回答

设计一个通用的Python框架，使大模型API能够自动思考问题、调用工具、反思迭代，并最终输出满足用户需求的结果。

框架设计思路

这个框架需要实现以下核心功能：

问题分析：理解用户输入的问题步骤规划：将问题分解为可执行的步骤工具调用：根据需要调用适当的工具反思评估：评估当前进展是否满足用户需求迭代优化：如果不满足需求，继续思考下一步

框架实现代码

下面是一个完整的Python框架实现：

import os
import json
import time
import requests
from typing import Dict, List, Any, Optional, Union, Callable

class Tool:
    """工具基类，所有工具都应继承此类"""
    def __init__(self, name: str, description: str):
        self.name = name
        self.description = description

    def execute(self, **kwargs) -> Dict[str, Any]:
        """执行工具功能，需要在子类中实现"""
        raise NotImplementedError("Tool子类必须实现execute方法")

    def to_dict(self) -> Dict[str, Any]:
        """返回工具的描述信息，用于API调用"""
        return {
            "name": self.name,
            "description": self.description
        }

class SearchTool(Tool):
    """搜索工具，用于获取互联网信息"""
    def __init__(self, api_key: str = None):
        super().__init__(
            name="search",
            description="搜索互联网获取信息，参数：query (搜索查询字符串)"
        )
        self.api_key = api_key

    def execute(self, query: str) -> Dict[str, Any]:
        """执行搜索操作"""
        print(f"正在搜索: {query}")
        # 这里应该是实际的搜索API调用
        # 为了演示，返回模拟数据
        return {
            "status": "success",
            "results": [
                {
                    "title": f"搜索结果 - {query}",
                    "content": f"这是关于 {query} 的搜索结果示例。在实际应用中，这里应该返回真实的搜索结果。"
                }
            ]
        }

class CodeRunnerTool(Tool):
    """代码运行工具，用于执行Python代码"""
    def __init__(self):
        super().__init__(
            name="run_code",
            description="运行Python代码，参数：code (要执行的Python代码字符串)"
        )

    def execute(self, code: str) -> Dict[str, Any]:
        """执行Python代码"""
        print("正在运行代码...")
        try:
            # 创建一个安全的本地环境来执行代码
            local_vars = {}
            # 注意：在实际应用中，应该使用更安全的沙箱环境
            exec(code, {}, local_vars)

            return {
                "status": "success",
                "result": str(local_vars.get('result', '代码执行成功，但没有返回值'))
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }

class PDFGeneratorTool(Tool):
    """PDF生成工具，用于创建PDF报告"""
    def __init__(self, output_dir: str = "output"):
        super().__init__(
            name="generate_pdf",
            description="生成PDF报告，参数：content (报告内容), title (报告标题), charts (可选，图表文件路径列表)"
        )
        self.output_dir = output_dir
        if not os.path.exists(output_dir):
            os.makedirs(output_dir)

    def execute(self, content: str, title: str = "自动生成报告", charts: List[str] = None) -> Dict[str, Any]:
        """生成PDF报告"""
        print("正在生成PDF报告...")
        # 这里应该是实际的PDF生成代码
        # 为了演示，只返回成功信息
        filename = f"report_{int(time.time())}.pdf"
        filepath = os.path.join(self.output_dir, filename)

        # 模拟PDF生成
        with open(filepath, 'w') as f:
            f.write(f"标题: {title}\n\n")
            f.write(content)
            if charts:
                f.write("\n\n图表列表:\n")
                for chart in charts:
                    f.write(f"- {chart}\n")

        return {
            "status": "success",
            "filename": filename,
            "filepath": filepath
        }

class AutoAgent:
    """自动代理类，负责协调大模型API和工具调用"""
    def __init__(self, api_key: str, model: str = "gpt-4", max_iterations: int = 10):
        self.api_key = api_key
        self.model = model
        self.max_iterations = max_iterations
        self.conversation_history = []
        self.tools = {}
        self.tool_descriptions = []

    def register_tool(self, tool: Tool) -> None:
        """注册工具"""
        self.tools[tool.name] = tool
        self.tool_descriptions.append(tool.to_dict())

    def add_message(self, role: str, content: Union[str, Dict[str, Any]]) -> None:
        """添加消息到对话历史"""
        self.conversation_history.append({"role": role, "content": content})

    def call_llm_api(self, messages: List[Dict[str, Any]], functions: List[Dict[str, Any]] = None) -> Dict[str, Any]:
        """调用大模型API"""
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}"
        }

        payload = {
            "model": self.model,
            "messages": messages,
            "temperature": 0.7
        }

        if functions:
            payload["functions"] = functions
            payload["function_call"] = "auto"

        try:
            response = requests.post(
                "https://api.openai.com/v1/chat/completions",
                headers=headers,
                json=payload
            )
            return response.json()
        except Exception as e:
            print(f"API调用错误: {e}")
            return None

    def execute_tool(self, tool_name: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """执行指定的工具"""
        if tool_name in self.tools:
            return self.tools[tool_name].execute(**params)
        else:
            return {"status": "error", "error": f"未知工具: {tool_name}"}

    def solve_problem(self, problem: str) -> Dict[str, Any]:
        """解决问题的主要流程"""
        print(f"开始解决问题: {problem}")

        # 初始化对话
        self.conversation_history = []

        # 设置系统提示
        self.add_message("system", """
        你是一个自动解决问题的AI助手。你需要：
        1. 分析问题并制定解决方案
        2. 使用提供的工具收集信息和处理数据
        3. 反思你的进展并迭代改进
        4. 最终生成一个完整的解决方案

        在每一步，你都应该：
        - 思考：分析当前状态和下一步行动
        - 行动：调用适当的工具
        - 观察：分析工具返回的结果
        - 反思：评估进展并调整计划

        请一步一步地解决问题，并在每一步清晰地解释你的思考过程。
        """)

        # 添加用户问题
        self.add_message("user", problem)

        # 准备工具函数描述
        functions = []
        for tool in self.tools.values():
            tool_dict = {
                "name": tool.name,
                "description": tool.description,
                "parameters": {
                    "type": "object",
                    "properties": {}
                }
            }

            # 根据工具名称添加参数
            if tool.name == "search":
                tool_dict["parameters"]["properties"]["query"] = {
                    "type": "string",
                    "description": "搜索查询"
                }
                tool_dict["parameters"]["required"] = ["query"]

            elif tool.name == "run_code":
                tool_dict["parameters"]["properties"]["code"] = {
                    "type": "string",
                    "description": "要执行的Python代码"
                }
                tool_dict["parameters"]["required"] = ["code"]

            elif tool.name == "generate_pdf":
                tool_dict["parameters"]["properties"] = {
                    "content": {
                        "type": "string",
                        "description": "PDF内容"
                    },
                    "title": {
                        "type": "string",
                        "description": "报告标题"
                    },
                    "charts": {
                        "type": "array",
                        "items": {
                            "type": "string"
                        },
                        "description": "图表文件路径列表"
                    }
                }
                tool_dict["parameters"]["required"] = ["content"]

            functions.append(tool_dict)

        # 迭代解决问题
        final_result = None
        for iteration in range(self.max_iterations):
            print(f"\n--- 迭代 {iteration + 1} ---")

            # 调用LLM获取下一步行动
            response = self.call_llm_api(self.conversation_history, functions)

            if not response or "choices" not in response:
                print("API返回错误，终止迭代")
                break

            message = response["choices"][0]["message"]

            # 检查是否调用工具
            if "function_call" in message:
                function_call = message["function_call"]
                tool_name = function_call["name"]
                tool_params = json.loads(function_call["arguments"])

                print(f"调用工具: {tool_name}")
                self.add_message("assistant", message)

                # 执行工具调用
                tool_result = self.execute_tool(tool_name, tool_params)

                # 添加工具结果到对话
                self.add_message("function", {
                    "name": tool_name,
                    "content": json.dumps(tool_result)
                })

                # 如果是生成PDF的工具调用，可能是最终步骤
                if tool_name == "generate_pdf" and tool_result["status"] == "success":
                    print(f"已生成PDF报告: {tool_result['filename']}")
                    final_result = tool_result

                    # 获取最终总结
                    self.add_message("user", "请总结你完成的工作和生成的报告内容。")
                    final_response = self.call_llm_api(self.conversation_history)
                    if final_response and "choices" in final_response:
                        final_message = final_response["choices"][0]["message"]["content"]
                        print("\n最终总结:")
                        print(final_message)

                    # 检查是否满足用户需求
                    self.add_message("user", "你认为当前的解决方案是否满足了用户的需求？如果是，请说明理由；如果不是，请说明还需要做什么。")
                    eval_response = self.call_llm_api(self.conversation_history)

                    if eval_response and "choices" in eval_response:
                        eval_message = eval_response["choices"][0]["message"]["content"]
                        print("\n需求满足评估:")
                        print(eval_message)

                        # 简单判断是否满足需求
                        if "满足" in eval_message and "不" not in eval_message[:eval_message.find("满足")]:
                            print("✅ 解决方案已满足用户需求，终止迭代")
                            break
                        else:
                            print("❌ 解决方案尚未满足用户需求，继续迭代")
                            self.add_message("assistant", eval_response["choices"][0]["message"])
                            self.add_message("user", "请继续完善解决方案，直到满足用户需求。")
            else:
                # 模型返回了文本回复而不是工具调用
                content = message["content"]
                print(f"AI思考: {content[:100]}...")
                self.add_message("assistant", message)

                # 提示模型继续解决问题
                self.add_message("user", "请继续解决问题，必要时调用适当的工具。")

        if iteration >= self.max_iterations - 1:
            print("达到最大迭代次数，终止")

        return final_result

def create_auto_agent(api_key: str) -> AutoAgent:
    """创建并配置自动代理"""
    agent = AutoAgent(api_key)

    # 注册工具
    agent.register_tool(SearchTool())
    agent.register_tool(CodeRunnerTool())
    agent.register_tool(PDFGeneratorTool())

    return agent

# 使用示例
if __name__ == "__main__":
    # 替换为你的API密钥
    api_key = "your_openai_api_key"

    # 创建自动代理
    agent = create_auto_agent(api_key)

    # 定义问题
    problem = """
    我需要收集价格在1500-2000元范围内的最新手机品牌和参数数据，并形成一个可视化的报告。
    报告应包含：
    1. 这个价格区间的主要手机品牌和型号
    2. 各品牌手机的主要参数对比（如处理器、内存、电池容量、屏幕等）
    3. 性价比分析
    4. 数据可视化图表
    """

    # 解决问题
    result = agent.solve_problem(problem)

    if result:
        print(f"\n问题已解决，最终报告: {result['filepath']}")
    else:
        print("\n未能完全解决问题")

框架扩展：添加更多工具

您可以轻松扩展这个框架，添加更多自定义工具：

from auto_agent_framework import Tool
import matplotlib.pyplot as plt
import pandas as pd
import requests
from bs4 import BeautifulSoup

class WebScraperTool(Tool):
    """网页抓取工具，用于从网页获取数据"""
    def __init__(self):
        super().__init__(
            name="web_scraper",
            description="从指定URL抓取网页内容，参数：url (网页地址)"
        )

    def execute(self, url: str) -> Dict[str, Any]:
        """抓取网页内容"""
        print(f"正在抓取网页: {url}")
        try:
            headers = {
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
            }
            response = requests.get(url, headers=headers)
            response.raise_for_status()

            soup = BeautifulSoup(response.text, 'html.parser')

            # 提取页面标题和正文内容
            title = soup.title.string if soup.title else "无标题"

            # 提取正文段落
            paragraphs = [p.text for p in soup.find_all('p')]
            content = "\n".join(paragraphs)

            return {
                "status": "success",
                "title": title,
                "content": content[:1000] + "..." if len(content) > 1000 else content
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }

class DataVisualizationTool(Tool):
    """数据可视化工具，用于创建图表"""
    def __init__(self, output_dir: str = "charts"):
        super().__init__(
            name="visualize_data",
            description="创建数据可视化图表，参数：data (数据字典或JSON字符串), chart_type (图表类型: bar, line, pie, scatter), title (图表标题)"
        )
        self.output_dir = output_dir
        if not os.path.exists(output_dir):
            os.makedirs(output_dir)

    def execute(self, data: Union[Dict, str], chart_type: str, title: str) -> Dict[str, Any]:
        """创建数据可视化图表"""
        print(f"正在创建{chart_type}图表: {title}")
        try:
            # 如果数据是字符串，尝试解析为JSON
            if isinstance(data, str):
                data = json.loads(data)

            # 创建DataFrame
            df = pd.DataFrame(data)

            # 创建图表
            plt.figure(figsize=(10, 6))

            if chart_type == "bar":
                df.plot(kind='bar')
            elif chart_type == "line":
                df.plot(kind='line')
            elif chart_type == "pie":
                df.plot(kind='pie', y=df.columns[1], autopct='%1.1f%%')
            elif chart_type == "scatter":
                df.plot(kind='scatter', x=df.columns[0], y=df.columns[1])
            else:
                return {
                    "status": "error",
                    "error": f"不支持的图表类型: {chart_type}"
                }

            plt.title(title)
            plt.tight_layout()

            # 保存图表
            filename = f"{chart_type}_{int(time.time())}.png"
            filepath = os.path.join(self.output_dir, filename)
            plt.savefig(filepath)
            plt.close()

            return {
                "status": "success",
                "chart_type": chart_type,
                "title": title,
                "filepath": filepath
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }

使用示例

下面是一个完整的使用示例，展示如何使用这个框架解决您提到的手机数据收集和报告生成问题：

from auto_agent_framework import create_auto_agent
from additional_tools import WebScraperTool, DataVisualizationTool

# 替换为你的API密钥
api_key = "your_openai_api_key"

# 创建自动代理
agent = create_auto_agent(api_key)

# 注册额外的工具
agent.register_tool(WebScraperTool())
agent.register_tool(DataVisualizationTool())

# 定义问题
problem = """
我需要收集价格在1500-2000元范围内的最新手机品牌和参数数据，并形成一个可视化的报告。
报告应包含：
1. 这个价格区间的主要手机品牌和型号
2. 各品牌手机的主要参数对比（如处理器、内存、电池容量、屏幕等）
3. 性价比分析
4. 数据可视化图表
"""

# 解决问题
result = agent.solve_problem(problem)

if result:
    print(f"\n问题已解决，最终报告: {result['filepath']}")
    # 可以自动打开生成的PDF
    import os
    os.system(f"start {result['filepath']}")  # Windows系统
    # 对于Mac: os.system(f"open {result['filepath']}")
    # 对于Linux: os.system(f"xdg-open {result['filepath']}")
else:
    print("\n未能完全解决问题")

框架核心组件解析

1. 工具抽象层

- Tool 基类定义了所有工具的通用接口

- 每个具体工具（如 SearchTool 、 CodeRunnerTool ）实现特定功能

- 工具注册机制允许动态添加新工具

2. 对话管理

- 维护与大模型的对话历史

- 支持多轮交互和上下文理解

3. 迭代反思机制

- 每次工具调用后评估进展

- 通过提问"是否满足用户需求"进行自我评估

- 根据评估结果决定是否继续迭代

4. 工具调用流程

- 大模型决定调用哪个工具

- 框架执行工具并返回结果

- 结果被添加到对话历史中供下一轮决策使用

扩展建议

1. 增强安全性

- 为代码执行工具添加沙箱环境

- 实现更严格的参数验证

- 添加用户权限控制

2. 提高效率

- 实现并行工具调用

- 添加缓存机制减少重复API调用

- 优化对话历史管理，减少token消耗

3. 增强评估能力

- 添加更细致的评估标准

- 实现基于用户反馈的学习机制

- 支持多维度评估（准确性、完整性、效率等）

4. 用户交互

- 添加中间结果展示

- 支持用户干预和引导

- 实现进度报告和估计完成时间

总结

这个框架实现了一个完整的自动思考、工具调用和反思迭代的系统，可以用于解决各种复杂问题。它的核心优势在于：

1. 模块化设计 ：易于扩展和定制

2. 自主决策 ：大模型自主决定下一步行动

3. 反思迭代 ：通过自我评估不断改进解决方案

4. 工具灵活性 ：可以根据需要添加各种工具

通过这个框架，您可以让大模型API自动解决各种复杂问题，从数据收集、分析到报告生成，全程自动化完成。