+
86
-

如何让大模型自动思考问题自动调用工具反思迭代几次输出结果?

如何让大模型自动思考问题自动调用工具反思迭代几次输出结果?

比如我要搜集最新的1500-2000范围的手机的品牌和参数等数据,并形成一个可视化的报告,工具有代码运行器、搜索api、pdf生成等工具,然后让大模型思考一步一步解决问题,最终给出一个pdf报告。

网友回复

+
4
-

设计一个通用的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. 分析问题并制定解...

点击查看剩余70%

我知道答案,我要回答