1. 项目概述一个会“自言自语”的API调用AI智能体最近我完成了一个挺有意思的项目一个在调用你的API时会“自言自语”地展示其思考过程的AI智能体。听起来可能有点抽象简单来说这不是一个简单的API调用封装库而是一个具备“内省”和“规划”能力的自动化助手。它不仅能执行你给它的任务比如“通过用户ID获取其最近订单并计算平均消费金额”还能在执行每一步操作前像人类一样在后台“嘀咕”它的推理“嗯要完成这个任务我需要先调用/users/{id}接口获取用户信息但用户ID还没拿到得先从请求上下文里解析出来。解析出来后我需要检查响应里是否包含‘orders’字段如果没有可能得再调用/orders?userId{id}这个独立的接口……”这个项目的核心价值或者说“非显而易见”的部分并不在于它能调用API——任何脚本都能做到。真正的挑战和趣味在于如何让一个黑盒般的AI模型在面对复杂、多步骤、可能出错的真实API时能进行透明的、可解释的、链式的思考并最终可靠地完成任务。这涉及到提示工程、执行流程设计、错误处理以及如何将“思考”这个过程本身作为一种可监控、可调试的资产输出出来。对于开发者、测试人员甚至产品经理而言这样一个能“边做边想”的智能体在对接新API、构建自动化工作流或排查集成问题时价值是巨大的。2. 核心设计思路为什么“思考出声”比“默默执行”更难2.1 从“函数调用”到“任务分解”的范式转变传统的API自动化工具或SDK其范式是“函数调用”。你预先定义好接口的路径、参数、方法然后编写逻辑去按顺序调用它们。这种方式清晰、可控但缺乏灵活性。一旦任务流程发生变化或者遇到接口返回非预期数据代码就需要修改。而这个AI智能体的设计范式是“任务分解”。你给它一个用自然语言描述的高级目标比如“为注册超过30天但未下单的用户发送优惠券”。智能体需要自己理解这个目标并将其分解为一系列可执行的原子操作步骤。这个分解过程就是它需要“思考出声”的核心部分。它必须推理出首先我需要一个能筛选用户的接口其次筛选条件可能包括“注册时间”和“订单数量”然后我需要一个发放优惠券的接口最后我需要将前一步的结果作为后一步的输入。注意这里的“思考”并非真正的意识而是大型语言模型根据给定的上下文和指令进行的下一步动作预测和理由生成。我们通过特定的提示词设计强制要求模型在输出最终动作前先输出其推理链。2.2 “非显而易见”的挑战幻觉、状态与工具描述让AI“思考”听起来很酷但实现起来有几个不那么显而易见的坑幻觉与事实性LLM可能会“幻想”出一些不存在的API端点或参数。例如它可能自信地“思考”“接下来调用/api/v1/users/{{userId}}/recent-orders。” 但你的实际API可能是/api/orders?userId{{userId}}limit5。如何将模型的“思考”牢牢锚定在你提供的、真实的API文档上是首要挑战。状态管理一个多步骤任务中前一步的执行结果如获取到的用户ID需要传递给下一步使用。智能体需要有能力记住和使用这些中间状态。这不仅仅是变量传递模型在“思考”下一步时必须能正确引用这些状态变量名。工具描述的精确性你需要将每个API接口描述成一个“工具”供模型调用。这个描述不能只是简单的“获取用户信息”而必须包含精确的端点、方法、参数列表名称、类型、是否必填、描述、请求体格式以及成功响应的示例结构。模糊的描述会导致模型调用错误或参数传递混乱。我的设计思路是采用“规划-执行-观察”循环Plan-Execute-Observe Loop并在每个循环中强制输出“内部独白Internal Monologue”。3. 架构拆解构建“思考型”智能体的四大支柱3.1 支柱一结构化工具定义与动态上下文工具定义是整个系统的基石。我设计了一个比OpenAI Function Calling更丰富的描述格式不仅包含标准字段还增加了“使用场景示例”和“常见错误处理”提示。{ name: get_user_by_id, description: 根据用户唯一ID获取用户的详细信息。通常用于获取用户基础档案后进行后续操作如查询订单、更新信息。, endpoint: /api/v1/users/{userId}, method: GET, path_parameters: [ { name: userId, type: string, description: 用户的唯一标识符通常由系统分配。可以从‘list_users’接口的响应中获取或从任务上下文中解析。, required: true } ], query_parameters: [], request_body_schema: null, response_example: { success: true, data: { id: usr_123abc, name: 张三, email: zhangsanexample.com, registrationDate: 2023-10-01, tags: [vip, active] } }, potential_errors: [ {code: 404, reason: 提供的userId不存在。}, {code: 403, reason: 当前API密钥无权访问该用户信息。} ] }关键点description字段至关重要。我会用自然语言描述这个工具“在什么情况下使用”以及“它的输出通常用于什么”。这直接引导了模型的“思考”方向。例如“通常用于…进行后续操作”这句话会暗示模型这个工具的输出结果用户ID是后续步骤的关键输入。所有工具定义会被动态地注入到每次与LLM交互的上下文Prompt中。随着任务步骤的推进已经调用过的工具及其结果会被移出“可用工具列表”以避免模型重复调用同时被加入到“已执行历史”中作为后续思考的参考。3.2 支柱二分步式提示工程与思维链强制这是实现“思考出声”的核心技术。我设计的提示词模板包含几个强制部分系统指令设定AI的角色“你是一个专业的API自动化助手”、核心原则“在决定使用哪个工具前你必须逐步推理”和输出格式要求。任务描述用户输入的自然语言任务。可用工具当前步骤所有可用的、结构化的工具定义。执行历史之前步骤的“思考”和“执行结果”。输出格式指令严格要求模型按以下格式响应**思考过程** [模型在这里用第一人称写下它的推理步骤。例如用户想找未下单的老用户。首先我需要找到所有用户。但我没有用户列表所以我需要先调用‘list_users’工具。这个工具可能需要分页参数我先尝试默认第一页...] **下一步动作** { “tool_name”: “list_users”, “arguments”: {“page”: 1, “pageSize”: 50} }通过这种强制格式我们“劫持”了模型的输出使其必须先生成一段人类可读的推理文本再生成一个结构化的动作指令。这段“思考过程”就是项目标题中“Thinks Out Loud”的部分。3.3 支柱三执行引擎与状态管理执行引擎负责解析模型输出的“下一步动作”调用真实的API并将结果格式化后反馈给模型进行下一轮“思考”。这里有几个细节参数替换引擎需要支持从“执行历史”中提取值来动态填充参数。例如get_user_orders工具需要userId而这个userId可能来自上一步get_user_by_id响应的data.id字段。引擎需要支持类似{{steps.get_user_by_id.result.data.id}}的模板语法。错误处理与重试当API调用返回错误如4xx, 5xx时引擎不能直接崩溃。它需要将错误信息如状态码、错误消息格式化后作为“观察结果”反馈给模型。模型在下一轮思考中需要分析这个错误并决定是重试如调整参数、换一种方式还是承认失败。这模拟了人类的调试过程。状态追踪维护一个全局的“任务状态字典”记录每一步的输出。这个字典的键可以是步骤的自定义名称如user_profile方便后续引用。3.4 支柱四可观测性与“思考日志”这是项目的亮点之一。整个智能体的运行过程会产生一份完整的“思考日志”这份日志不是简单的API调用记录而是包含了时间戳本轮思考的完整提示词输入模型生成的“思考过程”文本模型决定执行的“动作”执行引擎调用API的实际请求和响应执行耗时和状态这份日志对于调试和优化至关重要。你可以清晰地看到模型在哪一步“想歪了”是因为工具描述不清还是因为历史上下文有误导你也可以把这份日志分享给同事让他们快速理解智能体是如何完成一个复杂任务的这比看代码直观得多。4. 实操构建从零搭建一个基础版本4.1 环境准备与工具选型我选择Python作为实现语言因为它有丰富的AI和Web开发库。核心依赖openai/litellm用于与LLM如GPT-4, Claude-3交互。LiteLLM的优势在于它统一了多个模型供应商的接口。pydantic用于数据验证和设置管理。定义工具、动作、响应的数据结构非常方便。requests/httpx用于执行HTTP API调用。Httpx支持异步性能更好。python-dotenv管理API密钥等配置。为什么不直接用LangChain或AutoGPT这些框架功能强大但抽象层级高有时不够透明。为了深入理解“思考出声”的机制并实现高度定制化尤其是思维链的强制输出格式我从更底层的组件开始构建。这能让我们对控制流有绝对掌控。4.2 核心类设计与实现我设计了三个核心类Tool、Agent和ExecutionEngine。1. Tool类封装一个API工具的所有信息。from pydantic import BaseModel, Field from typing import Optional, Dict, Any, List class APIParameter(BaseModel): name: str type: str description: str required: bool True class Tool(BaseModel): name: str description: str endpoint: str method: str path_parameters: List[APIParameter] [] query_parameters: List[APIParameter] [] request_body_schema: Optional[Dict[str, Any]] None response_example: Optional[Dict[str, Any]] None def to_openai_function_schema(self) - Dict[str, Any]: 转换为OpenAI Function Calling格式但我们会主要使用自己的格式 properties {} required [] for param in self.path_parameters self.query_parameters: properties[param.name] {type: param.type, description: param.description} if param.required: required.append(param.name) # ... 处理request_body return { name: self.name, description: self.description, parameters: { type: object, properties: properties, required: required } }2. Agent类负责与LLM交互生成“思考”和“动作”。class Agent: def __init__(self, llm_client, tools: List[Tool], system_prompt: str): self.llm llm_client self.tools tools self.system_prompt system_prompt self.conversation_history [] # 保存多轮交互 def format_prompt(self, task: str, history: List[Dict]) - str: # 1. 加入系统指令 prompt_parts [fSystem: {self.system_prompt}\n\n] # 2. 加入任务 prompt_parts.append(fTask: {task}\n\n) # 3. 加入可用工具描述 prompt_parts.append(## Available Tools:\n) for tool in self.tools: # 这里用更易读的方式格式化工具描述而非纯JSON prompt_parts.append(f- {tool.name}: {tool.description}\n) prompt_parts.append(f Endpoint: {tool.method} {tool.endpoint}\n) if tool.path_parameters: prompt_parts.append(f Path Params: {, .join([p.name for p in tool.path_parameters])}\n) # ... 格式化更多细节 prompt_parts.append(\n) # 4. 加入执行历史 if history: prompt_parts.append(## Execution History:\n) for step in history[-5:]: # 只保留最近几步防止上下文过长 prompt_parts.append(f- Step {step[step]}: {step.get(thought, N/A)}\n) prompt_parts.append(f Action: {step.get(action)}\n) prompt_parts.append(f Result: {str(step.get(result))[:200]}...\n) # 截断长结果 prompt_parts.append(\n) # 5. 加入严格的输出格式指令 prompt_parts.append(## Your Response MUST follow this format:\n) prompt_parts.append(**Thought Process:**\n[Your step-by-step reasoning here]\n\n) prompt_parts.append(**Next Action:**\njson\n{\n \tool_name\: \tool_name_here\,\n \arguments\: {\arg1\: \value1\}\n}\n) return .join(prompt_parts) async def think_and_plan(self, task: str, history: List[Dict]) - Dict[str, Any]: prompt self.format_prompt(task, history) self.conversation_history.append({role: user, content: prompt}) try: response await self.llm.acompletions.create( modelgpt-4, messagesself.conversation_history, temperature0.1, # 低温度保证输出稳定符合格式 max_tokens1500 ) raw_text response.choices[0].message.content # 解析响应分离“思考过程”和“下一步动作” thought, action_json self._parse_response(raw_text) return {thought: thought, action: action_json} except Exception as e: # 处理LLM调用失败 return {thought: fFailed to generate plan: {str(e)}, action: None}3. ExecutionEngine类负责执行动作调用真实API管理状态。class ExecutionEngine: def __init__(self, base_url: str, auth_token: Optional[str] None): self.base_url base_url.rstrip(/) self.auth_token auth_token self.state {} # 全局状态存储 async def execute(self, tool: Tool, arguments: Dict, step_name: str) - Dict: # 1. 参数渲染将形如 {{state.user_id}} 的模板替换为实际值 rendered_args self._render_arguments(arguments) # 2. 构建请求根据工具定义将参数填充到路径、查询字符串或请求体中 url, req_params, req_body self._build_request(tool, rendered_args) # 3. 发送HTTP请求 async with httpx.AsyncClient(timeout30.0) as client: headers {Authorization: fBearer {self.auth_token}} if self.auth_token else {} try: resp await client.request( methodtool.method, urlurl, paramsreq_params, jsonreq_body, headersheaders ) resp.raise_for_status() result resp.json() # 4. 可选将关键结果存入全局状态 if step_name: self.state[step_name] result return {success: True, data: result, status_code: resp.status_code} except httpx.HTTPStatusError as e: # API返回错误状态码 return {success: False, error: fHTTP {e.response.status_code}, details: e.response.text} except Exception as e: # 网络或其他错误 return {success: False, error: ExecutionFailed, details: str(e)} def _render_arguments(self, arguments: Dict) - Dict: rendered {} for k, v in arguments.items(): if isinstance(v, str) and v.startswith({{) and v.endswith(}}): key_path v[2:-2].strip() # 去掉 {{ 和 }} # 简单支持 state.key 的路径实际可以更复杂 if key_path.startswith(state.): state_key key_path.split(.)[1] rendered[k] self.state.get(state_key, v) # 取不到则保留原值 else: rendered[k] v else: rendered[k] v return rendered4.3 主循环与日志记录将上述组件串联起来的主循环逻辑如下async def run_agent(task_description: str, tools: List[Tool], max_steps10): agent Agent(llm_client, tools, SYSTEM_PROMPT) engine ExecutionEngine(base_urlhttps://api.yourservice.com, auth_tokenAPI_KEY) history [] step 1 # 初始化思考日志 thought_log [] while step max_steps: print(f\n Step {step} ) # 1. Agent思考 plan_result await agent.think_and_plan(task_description, history) thought plan_result[thought] action plan_result[action] log_entry {step: step, thought: thought, planned_action: action} print(fThought: {thought[:200]}...) # 打印部分思考 if not action: print(Agent decided to stop or failed to plan.) thought_log.append({**log_entry, result: No action planned.}) break # 2. 查找对应的Tool对象 tool_to_use next((t for t in tools if t.name action[tool_name]), None) if not tool_to_use: error_msg fTool {action[tool_name]} not found. result {success: False, error: ToolNotFound, details: error_msg} print(fError: {error_msg}) else: # 3. 执行引擎执行 print(fExecuting: {tool_to_use.name} with args {action[arguments]}) result await engine.execute(tool_to_use, action[arguments], step_namefstep_{step}) # 4. 记录结果到历史供下一轮思考 history.append({ step: step, thought: thought, action: action, result: result }) # 5. 保存到思考日志 thought_log.append({**log_entry, result: result, executed_tool: tool_to_use.name if tool_to_use else None}) # 6. 判断是否终止任务完成或出现致命错误 if result.get(success) and _is_task_complete(result.get(data), task_description): print(Task completed successfully!) break if step max_steps: print(Reached max steps without completion.) break step 1 # 将完整的thought_log保存为JSON文件便于分析 import json with open(fthought_log_{int(time.time())}.json, w) as f: json.dump(thought_log, f, indent2, ensure_asciiFalse) print(f\nFull thought log saved.) return thought_log5. 核心难点与避坑实录5.1 难点一如何让模型的“思考”保持专注且有用最初模型的“思考过程”常常天马行空会讨论很多与当前可用工具无关的背景知识或者陷入无限循环的自我提问。解决方案在系统提示词中明确约束加入强指令如“你的思考必须紧密围绕如何利用当前提供的工具来推进任务。不要假设不存在的工具。如果当前工具无法直接达成目标请思考如何组合它们。”提供“思考范例”在Few-Shot Prompting中在系统消息里提供1-2个高质量的“思考-行动”对作为示例。这能极大地校准模型的输出格式和思考深度。工具描述的引导性如前所述在工具描述的description字段中暗示其“上游”和“下游”关系。例如对于get_order_details工具描述写成“在已获得订单ID后用于获取订单的详细商品列表和金额。订单ID通常来自list_orders_by_user工具的结果。” 这样模型在思考时会自然形成链条。5.2 难点二复杂参数传递与状态引用当任务需要多步骤时如何让模型理解并正确引用上一步的结果作为下一步的参数解决方案在提示词中明确展示变量语法在给模型的“可用工具”描述或示例中明确写出参数可以引用历史结果例如arguments: {userId: {{steps.get_user.result.data.id}}”}。虽然实际替换由执行引擎完成但这让模型知道了这种能力的存在。在“执行历史”中格式化显示结果在Agent.format_prompt方法中展示历史记录时不仅显示原始JSON更用自然语言摘要关键输出字段。例如“Result: Success. Retrieved user profile: id’usr_123’, name’张三’.” 这比展示一大段JSON更容易让模型捕捉到关键信息。设计状态键名执行引擎在存储状态时self.state[step_name] result可以使用有意义的键名如user_profile、recent_orders并在提示词中告知模型这些键名可供引用。5.3 难点三错误处理与恢复API调用总会出错网络超时、认证失败、参数错误、数据不存在。智能体不能一遇错误就崩溃。解决方案将错误信息作为“观察”反馈执行引擎将任何非成功的执行结果包括HTTP错误和异常都格式化为一个标准结构反馈给Agent。例如{“success”: false, “error_type”: “HTTP_404”, “message”: “User not found with id ‘usr_xxx’.”}。在系统提示词中训练模型处理错误加入类似指令“如果执行结果返回错误请在你的‘思考过程’中分析错误原因。是参数错误还是需要先调用其他工具根据分析决定是重试调整参数、尝试替代方案还是终止任务并报告原因。”设置重试与回退逻辑可以在执行引擎层面对网络超时等临时性错误进行自动重试。对于业务逻辑错误如404则依赖模型的分析和决策。5.4 难点四控制成本与运行效率每次“思考”都需要调用一次LLM如果任务步骤多成本会很高。同时提示词上下文会随着历史增长而膨胀影响速度和效果。解决方案压缩历史记录不在每一轮都将完整历史塞进提示词。只保留最近3-5步的详细记录对于更早的步骤只保留一个高度概括的摘要例如“步骤1-3成功获取了用户ID为’usr_123’的档案及其最近5笔订单。”使用更便宜的模型进行简单步骤对于模式固定、决策简单的步骤如解析出ID后直接调用下一个GET接口可以尝试使用更便宜、更快的模型如GPT-3.5-Turbo仅在需要复杂推理和规划时使用GPT-4。设置最大步数防止任务陷入无限循环必须设置max_steps硬性限制。6. 进阶优化与扩展思路6.1 引入“验证器”与“后置条件”为了让智能体更可靠可以为每个工具定义“后置条件”。例如get_user_by_id工具的后置条件可以是“响应中必须包含’id’和’email’字段”。执行引擎在拿到结果后自动验证这些条件。如果验证失败则自动将失败信息如“获取用户成功但响应缺少’email’字段”作为“观察”反馈给模型触发它进行异常处理比如尝试从另一个接口获取邮箱。6.2 实现“子任务”分解与递归执行对于极其复杂的任务可以让主Agent具备“发布子任务”的能力。当主Agent“思考”后认为当前目标可以分解为几个独立的子目标时它可以生成子任务描述并启动一个新的、专注于此子任务的智能体实例递归调用。子任务执行完毕后将结果汇总给主Agent。这类似于人类项目经理将大项目拆分成小模块。6.3 与可视化界面结合“思考日志”是绝佳的可视化素材。可以开发一个简单的Web界面以时间线或流程图的方式展示智能体完整的执行过程每个节点的展开可以看到当时的“思考过程”、发出的请求和收到的响应。这对于演示、教学和调试来说体验远超查看控制台日志。6.4 工具的动态发现与学习目前的工具是静态定义的。一个更高级的版本是让智能体能够阅读Swagger/OpenAPI文档并自动将其转化为内部的Tool定义。更进一步智能体可以在执行过程中如果发现现有工具无法满足需求可以尝试向一个“工具生成器”服务请求根据自然语言描述生成一个新API调用的临时定义当然这需要非常谨慎的安全审查。构建这样一个“会思考的API智能体”的过程本质上是在LLM强大的语义理解能力和确定性的程序执行之间搭建一座坚固且透明的桥梁。最大的收获不是最终能自动完成某个任务而是在构建过程中被迫去深入思考如何将模糊的人类指令转化为清晰、可执行、可追溯的步骤序列。这份“思考日志”或许才是人机协作新时代最有价值的副产品。
构建会“思考”的API智能体：从任务分解到透明执行的工程实践
1. 项目概述一个会“自言自语”的API调用AI智能体最近我完成了一个挺有意思的项目一个在调用你的API时会“自言自语”地展示其思考过程的AI智能体。听起来可能有点抽象简单来说这不是一个简单的API调用封装库而是一个具备“内省”和“规划”能力的自动化助手。它不仅能执行你给它的任务比如“通过用户ID获取其最近订单并计算平均消费金额”还能在执行每一步操作前像人类一样在后台“嘀咕”它的推理“嗯要完成这个任务我需要先调用/users/{id}接口获取用户信息但用户ID还没拿到得先从请求上下文里解析出来。解析出来后我需要检查响应里是否包含‘orders’字段如果没有可能得再调用/orders?userId{id}这个独立的接口……”这个项目的核心价值或者说“非显而易见”的部分并不在于它能调用API——任何脚本都能做到。真正的挑战和趣味在于如何让一个黑盒般的AI模型在面对复杂、多步骤、可能出错的真实API时能进行透明的、可解释的、链式的思考并最终可靠地完成任务。这涉及到提示工程、执行流程设计、错误处理以及如何将“思考”这个过程本身作为一种可监控、可调试的资产输出出来。对于开发者、测试人员甚至产品经理而言这样一个能“边做边想”的智能体在对接新API、构建自动化工作流或排查集成问题时价值是巨大的。2. 核心设计思路为什么“思考出声”比“默默执行”更难2.1 从“函数调用”到“任务分解”的范式转变传统的API自动化工具或SDK其范式是“函数调用”。你预先定义好接口的路径、参数、方法然后编写逻辑去按顺序调用它们。这种方式清晰、可控但缺乏灵活性。一旦任务流程发生变化或者遇到接口返回非预期数据代码就需要修改。而这个AI智能体的设计范式是“任务分解”。你给它一个用自然语言描述的高级目标比如“为注册超过30天但未下单的用户发送优惠券”。智能体需要自己理解这个目标并将其分解为一系列可执行的原子操作步骤。这个分解过程就是它需要“思考出声”的核心部分。它必须推理出首先我需要一个能筛选用户的接口其次筛选条件可能包括“注册时间”和“订单数量”然后我需要一个发放优惠券的接口最后我需要将前一步的结果作为后一步的输入。注意这里的“思考”并非真正的意识而是大型语言模型根据给定的上下文和指令进行的下一步动作预测和理由生成。我们通过特定的提示词设计强制要求模型在输出最终动作前先输出其推理链。2.2 “非显而易见”的挑战幻觉、状态与工具描述让AI“思考”听起来很酷但实现起来有几个不那么显而易见的坑幻觉与事实性LLM可能会“幻想”出一些不存在的API端点或参数。例如它可能自信地“思考”“接下来调用/api/v1/users/{{userId}}/recent-orders。” 但你的实际API可能是/api/orders?userId{{userId}}limit5。如何将模型的“思考”牢牢锚定在你提供的、真实的API文档上是首要挑战。状态管理一个多步骤任务中前一步的执行结果如获取到的用户ID需要传递给下一步使用。智能体需要有能力记住和使用这些中间状态。这不仅仅是变量传递模型在“思考”下一步时必须能正确引用这些状态变量名。工具描述的精确性你需要将每个API接口描述成一个“工具”供模型调用。这个描述不能只是简单的“获取用户信息”而必须包含精确的端点、方法、参数列表名称、类型、是否必填、描述、请求体格式以及成功响应的示例结构。模糊的描述会导致模型调用错误或参数传递混乱。我的设计思路是采用“规划-执行-观察”循环Plan-Execute-Observe Loop并在每个循环中强制输出“内部独白Internal Monologue”。3. 架构拆解构建“思考型”智能体的四大支柱3.1 支柱一结构化工具定义与动态上下文工具定义是整个系统的基石。我设计了一个比OpenAI Function Calling更丰富的描述格式不仅包含标准字段还增加了“使用场景示例”和“常见错误处理”提示。{ name: get_user_by_id, description: 根据用户唯一ID获取用户的详细信息。通常用于获取用户基础档案后进行后续操作如查询订单、更新信息。, endpoint: /api/v1/users/{userId}, method: GET, path_parameters: [ { name: userId, type: string, description: 用户的唯一标识符通常由系统分配。可以从‘list_users’接口的响应中获取或从任务上下文中解析。, required: true } ], query_parameters: [], request_body_schema: null, response_example: { success: true, data: { id: usr_123abc, name: 张三, email: zhangsanexample.com, registrationDate: 2023-10-01, tags: [vip, active] } }, potential_errors: [ {code: 404, reason: 提供的userId不存在。}, {code: 403, reason: 当前API密钥无权访问该用户信息。} ] }关键点description字段至关重要。我会用自然语言描述这个工具“在什么情况下使用”以及“它的输出通常用于什么”。这直接引导了模型的“思考”方向。例如“通常用于…进行后续操作”这句话会暗示模型这个工具的输出结果用户ID是后续步骤的关键输入。所有工具定义会被动态地注入到每次与LLM交互的上下文Prompt中。随着任务步骤的推进已经调用过的工具及其结果会被移出“可用工具列表”以避免模型重复调用同时被加入到“已执行历史”中作为后续思考的参考。3.2 支柱二分步式提示工程与思维链强制这是实现“思考出声”的核心技术。我设计的提示词模板包含几个强制部分系统指令设定AI的角色“你是一个专业的API自动化助手”、核心原则“在决定使用哪个工具前你必须逐步推理”和输出格式要求。任务描述用户输入的自然语言任务。可用工具当前步骤所有可用的、结构化的工具定义。执行历史之前步骤的“思考”和“执行结果”。输出格式指令严格要求模型按以下格式响应**思考过程** [模型在这里用第一人称写下它的推理步骤。例如用户想找未下单的老用户。首先我需要找到所有用户。但我没有用户列表所以我需要先调用‘list_users’工具。这个工具可能需要分页参数我先尝试默认第一页...] **下一步动作** { “tool_name”: “list_users”, “arguments”: {“page”: 1, “pageSize”: 50} }通过这种强制格式我们“劫持”了模型的输出使其必须先生成一段人类可读的推理文本再生成一个结构化的动作指令。这段“思考过程”就是项目标题中“Thinks Out Loud”的部分。3.3 支柱三执行引擎与状态管理执行引擎负责解析模型输出的“下一步动作”调用真实的API并将结果格式化后反馈给模型进行下一轮“思考”。这里有几个细节参数替换引擎需要支持从“执行历史”中提取值来动态填充参数。例如get_user_orders工具需要userId而这个userId可能来自上一步get_user_by_id响应的data.id字段。引擎需要支持类似{{steps.get_user_by_id.result.data.id}}的模板语法。错误处理与重试当API调用返回错误如4xx, 5xx时引擎不能直接崩溃。它需要将错误信息如状态码、错误消息格式化后作为“观察结果”反馈给模型。模型在下一轮思考中需要分析这个错误并决定是重试如调整参数、换一种方式还是承认失败。这模拟了人类的调试过程。状态追踪维护一个全局的“任务状态字典”记录每一步的输出。这个字典的键可以是步骤的自定义名称如user_profile方便后续引用。3.4 支柱四可观测性与“思考日志”这是项目的亮点之一。整个智能体的运行过程会产生一份完整的“思考日志”这份日志不是简单的API调用记录而是包含了时间戳本轮思考的完整提示词输入模型生成的“思考过程”文本模型决定执行的“动作”执行引擎调用API的实际请求和响应执行耗时和状态这份日志对于调试和优化至关重要。你可以清晰地看到模型在哪一步“想歪了”是因为工具描述不清还是因为历史上下文有误导你也可以把这份日志分享给同事让他们快速理解智能体是如何完成一个复杂任务的这比看代码直观得多。4. 实操构建从零搭建一个基础版本4.1 环境准备与工具选型我选择Python作为实现语言因为它有丰富的AI和Web开发库。核心依赖openai/litellm用于与LLM如GPT-4, Claude-3交互。LiteLLM的优势在于它统一了多个模型供应商的接口。pydantic用于数据验证和设置管理。定义工具、动作、响应的数据结构非常方便。requests/httpx用于执行HTTP API调用。Httpx支持异步性能更好。python-dotenv管理API密钥等配置。为什么不直接用LangChain或AutoGPT这些框架功能强大但抽象层级高有时不够透明。为了深入理解“思考出声”的机制并实现高度定制化尤其是思维链的强制输出格式我从更底层的组件开始构建。这能让我们对控制流有绝对掌控。4.2 核心类设计与实现我设计了三个核心类Tool、Agent和ExecutionEngine。1. Tool类封装一个API工具的所有信息。from pydantic import BaseModel, Field from typing import Optional, Dict, Any, List class APIParameter(BaseModel): name: str type: str description: str required: bool True class Tool(BaseModel): name: str description: str endpoint: str method: str path_parameters: List[APIParameter] [] query_parameters: List[APIParameter] [] request_body_schema: Optional[Dict[str, Any]] None response_example: Optional[Dict[str, Any]] None def to_openai_function_schema(self) - Dict[str, Any]: 转换为OpenAI Function Calling格式但我们会主要使用自己的格式 properties {} required [] for param in self.path_parameters self.query_parameters: properties[param.name] {type: param.type, description: param.description} if param.required: required.append(param.name) # ... 处理request_body return { name: self.name, description: self.description, parameters: { type: object, properties: properties, required: required } }2. Agent类负责与LLM交互生成“思考”和“动作”。class Agent: def __init__(self, llm_client, tools: List[Tool], system_prompt: str): self.llm llm_client self.tools tools self.system_prompt system_prompt self.conversation_history [] # 保存多轮交互 def format_prompt(self, task: str, history: List[Dict]) - str: # 1. 加入系统指令 prompt_parts [fSystem: {self.system_prompt}\n\n] # 2. 加入任务 prompt_parts.append(fTask: {task}\n\n) # 3. 加入可用工具描述 prompt_parts.append(## Available Tools:\n) for tool in self.tools: # 这里用更易读的方式格式化工具描述而非纯JSON prompt_parts.append(f- {tool.name}: {tool.description}\n) prompt_parts.append(f Endpoint: {tool.method} {tool.endpoint}\n) if tool.path_parameters: prompt_parts.append(f Path Params: {, .join([p.name for p in tool.path_parameters])}\n) # ... 格式化更多细节 prompt_parts.append(\n) # 4. 加入执行历史 if history: prompt_parts.append(## Execution History:\n) for step in history[-5:]: # 只保留最近几步防止上下文过长 prompt_parts.append(f- Step {step[step]}: {step.get(thought, N/A)}\n) prompt_parts.append(f Action: {step.get(action)}\n) prompt_parts.append(f Result: {str(step.get(result))[:200]}...\n) # 截断长结果 prompt_parts.append(\n) # 5. 加入严格的输出格式指令 prompt_parts.append(## Your Response MUST follow this format:\n) prompt_parts.append(**Thought Process:**\n[Your step-by-step reasoning here]\n\n) prompt_parts.append(**Next Action:**\njson\n{\n \tool_name\: \tool_name_here\,\n \arguments\: {\arg1\: \value1\}\n}\n) return .join(prompt_parts) async def think_and_plan(self, task: str, history: List[Dict]) - Dict[str, Any]: prompt self.format_prompt(task, history) self.conversation_history.append({role: user, content: prompt}) try: response await self.llm.acompletions.create( modelgpt-4, messagesself.conversation_history, temperature0.1, # 低温度保证输出稳定符合格式 max_tokens1500 ) raw_text response.choices[0].message.content # 解析响应分离“思考过程”和“下一步动作” thought, action_json self._parse_response(raw_text) return {thought: thought, action: action_json} except Exception as e: # 处理LLM调用失败 return {thought: fFailed to generate plan: {str(e)}, action: None}3. ExecutionEngine类负责执行动作调用真实API管理状态。class ExecutionEngine: def __init__(self, base_url: str, auth_token: Optional[str] None): self.base_url base_url.rstrip(/) self.auth_token auth_token self.state {} # 全局状态存储 async def execute(self, tool: Tool, arguments: Dict, step_name: str) - Dict: # 1. 参数渲染将形如 {{state.user_id}} 的模板替换为实际值 rendered_args self._render_arguments(arguments) # 2. 构建请求根据工具定义将参数填充到路径、查询字符串或请求体中 url, req_params, req_body self._build_request(tool, rendered_args) # 3. 发送HTTP请求 async with httpx.AsyncClient(timeout30.0) as client: headers {Authorization: fBearer {self.auth_token}} if self.auth_token else {} try: resp await client.request( methodtool.method, urlurl, paramsreq_params, jsonreq_body, headersheaders ) resp.raise_for_status() result resp.json() # 4. 可选将关键结果存入全局状态 if step_name: self.state[step_name] result return {success: True, data: result, status_code: resp.status_code} except httpx.HTTPStatusError as e: # API返回错误状态码 return {success: False, error: fHTTP {e.response.status_code}, details: e.response.text} except Exception as e: # 网络或其他错误 return {success: False, error: ExecutionFailed, details: str(e)} def _render_arguments(self, arguments: Dict) - Dict: rendered {} for k, v in arguments.items(): if isinstance(v, str) and v.startswith({{) and v.endswith(}}): key_path v[2:-2].strip() # 去掉 {{ 和 }} # 简单支持 state.key 的路径实际可以更复杂 if key_path.startswith(state.): state_key key_path.split(.)[1] rendered[k] self.state.get(state_key, v) # 取不到则保留原值 else: rendered[k] v else: rendered[k] v return rendered4.3 主循环与日志记录将上述组件串联起来的主循环逻辑如下async def run_agent(task_description: str, tools: List[Tool], max_steps10): agent Agent(llm_client, tools, SYSTEM_PROMPT) engine ExecutionEngine(base_urlhttps://api.yourservice.com, auth_tokenAPI_KEY) history [] step 1 # 初始化思考日志 thought_log [] while step max_steps: print(f\n Step {step} ) # 1. Agent思考 plan_result await agent.think_and_plan(task_description, history) thought plan_result[thought] action plan_result[action] log_entry {step: step, thought: thought, planned_action: action} print(fThought: {thought[:200]}...) # 打印部分思考 if not action: print(Agent decided to stop or failed to plan.) thought_log.append({**log_entry, result: No action planned.}) break # 2. 查找对应的Tool对象 tool_to_use next((t for t in tools if t.name action[tool_name]), None) if not tool_to_use: error_msg fTool {action[tool_name]} not found. result {success: False, error: ToolNotFound, details: error_msg} print(fError: {error_msg}) else: # 3. 执行引擎执行 print(fExecuting: {tool_to_use.name} with args {action[arguments]}) result await engine.execute(tool_to_use, action[arguments], step_namefstep_{step}) # 4. 记录结果到历史供下一轮思考 history.append({ step: step, thought: thought, action: action, result: result }) # 5. 保存到思考日志 thought_log.append({**log_entry, result: result, executed_tool: tool_to_use.name if tool_to_use else None}) # 6. 判断是否终止任务完成或出现致命错误 if result.get(success) and _is_task_complete(result.get(data), task_description): print(Task completed successfully!) break if step max_steps: print(Reached max steps without completion.) break step 1 # 将完整的thought_log保存为JSON文件便于分析 import json with open(fthought_log_{int(time.time())}.json, w) as f: json.dump(thought_log, f, indent2, ensure_asciiFalse) print(f\nFull thought log saved.) return thought_log5. 核心难点与避坑实录5.1 难点一如何让模型的“思考”保持专注且有用最初模型的“思考过程”常常天马行空会讨论很多与当前可用工具无关的背景知识或者陷入无限循环的自我提问。解决方案在系统提示词中明确约束加入强指令如“你的思考必须紧密围绕如何利用当前提供的工具来推进任务。不要假设不存在的工具。如果当前工具无法直接达成目标请思考如何组合它们。”提供“思考范例”在Few-Shot Prompting中在系统消息里提供1-2个高质量的“思考-行动”对作为示例。这能极大地校准模型的输出格式和思考深度。工具描述的引导性如前所述在工具描述的description字段中暗示其“上游”和“下游”关系。例如对于get_order_details工具描述写成“在已获得订单ID后用于获取订单的详细商品列表和金额。订单ID通常来自list_orders_by_user工具的结果。” 这样模型在思考时会自然形成链条。5.2 难点二复杂参数传递与状态引用当任务需要多步骤时如何让模型理解并正确引用上一步的结果作为下一步的参数解决方案在提示词中明确展示变量语法在给模型的“可用工具”描述或示例中明确写出参数可以引用历史结果例如arguments: {userId: {{steps.get_user.result.data.id}}”}。虽然实际替换由执行引擎完成但这让模型知道了这种能力的存在。在“执行历史”中格式化显示结果在Agent.format_prompt方法中展示历史记录时不仅显示原始JSON更用自然语言摘要关键输出字段。例如“Result: Success. Retrieved user profile: id’usr_123’, name’张三’.” 这比展示一大段JSON更容易让模型捕捉到关键信息。设计状态键名执行引擎在存储状态时self.state[step_name] result可以使用有意义的键名如user_profile、recent_orders并在提示词中告知模型这些键名可供引用。5.3 难点三错误处理与恢复API调用总会出错网络超时、认证失败、参数错误、数据不存在。智能体不能一遇错误就崩溃。解决方案将错误信息作为“观察”反馈执行引擎将任何非成功的执行结果包括HTTP错误和异常都格式化为一个标准结构反馈给Agent。例如{“success”: false, “error_type”: “HTTP_404”, “message”: “User not found with id ‘usr_xxx’.”}。在系统提示词中训练模型处理错误加入类似指令“如果执行结果返回错误请在你的‘思考过程’中分析错误原因。是参数错误还是需要先调用其他工具根据分析决定是重试调整参数、尝试替代方案还是终止任务并报告原因。”设置重试与回退逻辑可以在执行引擎层面对网络超时等临时性错误进行自动重试。对于业务逻辑错误如404则依赖模型的分析和决策。5.4 难点四控制成本与运行效率每次“思考”都需要调用一次LLM如果任务步骤多成本会很高。同时提示词上下文会随着历史增长而膨胀影响速度和效果。解决方案压缩历史记录不在每一轮都将完整历史塞进提示词。只保留最近3-5步的详细记录对于更早的步骤只保留一个高度概括的摘要例如“步骤1-3成功获取了用户ID为’usr_123’的档案及其最近5笔订单。”使用更便宜的模型进行简单步骤对于模式固定、决策简单的步骤如解析出ID后直接调用下一个GET接口可以尝试使用更便宜、更快的模型如GPT-3.5-Turbo仅在需要复杂推理和规划时使用GPT-4。设置最大步数防止任务陷入无限循环必须设置max_steps硬性限制。6. 进阶优化与扩展思路6.1 引入“验证器”与“后置条件”为了让智能体更可靠可以为每个工具定义“后置条件”。例如get_user_by_id工具的后置条件可以是“响应中必须包含’id’和’email’字段”。执行引擎在拿到结果后自动验证这些条件。如果验证失败则自动将失败信息如“获取用户成功但响应缺少’email’字段”作为“观察”反馈给模型触发它进行异常处理比如尝试从另一个接口获取邮箱。6.2 实现“子任务”分解与递归执行对于极其复杂的任务可以让主Agent具备“发布子任务”的能力。当主Agent“思考”后认为当前目标可以分解为几个独立的子目标时它可以生成子任务描述并启动一个新的、专注于此子任务的智能体实例递归调用。子任务执行完毕后将结果汇总给主Agent。这类似于人类项目经理将大项目拆分成小模块。6.3 与可视化界面结合“思考日志”是绝佳的可视化素材。可以开发一个简单的Web界面以时间线或流程图的方式展示智能体完整的执行过程每个节点的展开可以看到当时的“思考过程”、发出的请求和收到的响应。这对于演示、教学和调试来说体验远超查看控制台日志。6.4 工具的动态发现与学习目前的工具是静态定义的。一个更高级的版本是让智能体能够阅读Swagger/OpenAPI文档并自动将其转化为内部的Tool定义。更进一步智能体可以在执行过程中如果发现现有工具无法满足需求可以尝试向一个“工具生成器”服务请求根据自然语言描述生成一个新API调用的临时定义当然这需要非常谨慎的安全审查。构建这样一个“会思考的API智能体”的过程本质上是在LLM强大的语义理解能力和确定性的程序执行之间搭建一座坚固且透明的桥梁。最大的收获不是最终能自动完成某个任务而是在构建过程中被迫去深入思考如何将模糊的人类指令转化为清晰、可执行、可追溯的步骤序列。这份“思考日志”或许才是人机协作新时代最有价值的副产品。
