1. 项目概述为什么我们需要从零开始理解 Tool Calling如果你最近在折腾大语言模型应用尤其是想让它不只是“聊天”而是能真正帮你干点实事——比如查个天气、发封邮件、或者从数据库里拉份报表——那你肯定绕不开一个词Tool Calling。听起来挺高大上但说白了就是教 AI 学会“使用工具”。这就像给一个博学但手无缚鸡之力的学者配上了一套趁手的瑞士军刀让它能从“知道”进化到“做到”。市面上现成的框架比如 LangChain、Dify确实把 Tool Calling 封装得很好几行代码就能跑起来。但问题也来了当你遇到一个奇葩的业务需求或者框架的某个设计让你觉得别扭时那种“黑盒”感会让你非常无力。你只知道它“能跑”但不知道它“为什么这么跑”更不知道怎么让它“跑得更好”。这就是为什么我们要“从零手写”不是为了重复造轮子而是为了彻底搞懂轮子的每一个辐条是怎么受力、怎么转动的。只有亲手从零搭建一遍你才能真正理解 JSON Schema 如何描述工具、ReAct 思维链如何驱动决策、以及一个健壮的 Agent 工作流应该如何设计。这份理解是你在未来面对任何复杂 AI 应用架构时最宝贵的底气。2. 核心概念拆解Tool Calling 的四大基石在动手写代码之前我们必须把几个核心概念掰开揉碎了讲清楚。它们构成了 Tool Calling 这座大厦的地基。2.1 JSON Schema工具的“说明书”你可以把 JSON Schema 理解为一份给 AI 看的、极其严谨的“工具使用说明书”。AI 不像人类能通过模糊的自然语言理解“帮我查一下北京明天天气”。它需要明确知道这个工具叫什么名字get_weather需要什么参数city: 字符串date: 符合YYYY-MM-DD格式的字符串每个参数是什么意思。为什么必须是 JSON Schema因为它是机器可读的结构化标准。大语言模型在生成内容时本质上是在做“下一个词预测”。如果你要求它输出一个结构化的调用信息没有 Schema 的约束它可能会输出千奇百怪的格式你的程序根本无法解析。JSON Schema 就像是一个填空题的模板告诉 AI“请在这里填城市名那里填日期并且必须按照这个格式交卷。”一个简单的天气查询工具的 Schema 可能长这样{ name: get_weather, description: 获取指定城市在指定日期的天气信息, parameters: { type: object, properties: { city: { type: string, description: 城市名称例如北京、上海 }, date: { type: string, format: date, description: 日期格式为 YYYY-MM-DD } }, required: [city] } }注意description字段至关重要它是 AI 理解工具功能和参数含义的主要依据。写得模糊不清AI 就可能用错工具。2.2 ReAct 范式AI 的“思考过程”ReAct (Reason Act) 是驱动 Tool Calling 的核心推理框架。它的妙处在于强制 AI 将“思考”Reason和“行动”Act以文本形式交错输出。这让我们能够窥探 AI 的决策过程也使得整个流程更可控、可调试。一个典型的 ReAct 循环是这样的Thought思考AI 分析当前状况和用户目标决定下一步该做什么。Action行动AI 根据思考选择一个工具并生成符合 Schema 的调用参数。Observation观察执行工具并将执行结果成功或失败返回给 AI。回到第1步AI 根据观察结果进行下一轮思考直到问题解决或无法继续。例如用户问“刘德华的妻子是谁” AI 的 ReAct 过程可能是Thought: 用户问的是关于一个人的家庭成员信息。我内部的知识可能不是最新或最准确的我应该使用搜索工具来获取可靠信息。 Action: search_web 参数: query: 刘德华 妻子 Observation: 根据网络搜索结果刘德华的妻子是朱丽倩。 Thought: 我已经获得了准确信息可以直接回答用户了。 Answer: 刘德华的妻子是朱丽倩。这个过程清晰明了如果搜索工具返回了错误或无关信息我们可以从Observation中立刻发现并思考如何调整查询词而不是得到一个莫名其妙的错误答案。2.3 Agent 工作流从单次调用到持续会话单个 Tool Calling 解决了“用一次工具”的问题。但真实场景往往是多步骤、带状态的。比如用户说“帮我查一下北京明天的天气如果下雨就提醒我带伞。” 这至少需要两个步骤1. 调用天气工具2. 根据结果决定是否调用“发送提醒”工具。这就是Agent智能体的价值。一个简单的 Agent 工作流需要管理记忆Memory记住之前的对话历史、工具调用结果。工具集Toolkit所有可用的工具列表及其 Schema。决策循环ReAct Loop基于记忆和当前输入持续进行思考-行动-观察的循环。停止条件Stop Condition当 AI 认为任务已完成给出最终答案或陷入死循环、遇到无法解决的问题时优雅地停止。Agent Workflow 与 ReAct 的区别ReAct 是一个具体的推理模式而 Agent Workflow 是用这个模式构建的一个可以运行的程序。你可以把 ReAct 看作汽车的发动机工作原理进气、压缩、做功、排气而 Agent Workflow 是包含了发动机、变速箱、方向盘、油箱的整辆汽车。2.4 LLM 的角色核心的“推理引擎”在整个架构中大语言模型扮演着无可替代的“大脑”角色。它负责理解用户意图、进行逻辑推理Thought、生成符合规范的工具调用Action、以及整合信息生成最终回答。选择 LLM 的关键考量推理能力对于需要多步逻辑判断的复杂任务需要选择在推理基准上表现好的模型。指令遵循能力模型必须能严格遵守你给出的输出格式要求如“你必须以 Thought: 开头”。上下文长度决定了 Agent 能记住多长的对话历史和工具调用记录。成本与速度自建模型、开源模型还是商用 API需要权衡响应时间、费用和可控性。实操心得在开发初期建议使用 OpenAI 的 GPT-4 或 Anthropic 的 Claude 等顶级商用 API它们的指令遵循和推理能力非常稳定能极大降低调试难度。当流程跑通后再考虑用成本更低的开源模型如 DeepSeek、Qwen进行替换和优化但要做好在 Prompt 工程上投入更多精力的准备。3. 从零手写构建一个最小可用的 Tool Calling 系统理论说了一堆现在我们来点实在的。我们将用 Python 一步步构建一个核心系统它不依赖任何外部框架只使用requests调用 LLM API 和标准库。3.1 第一步定义工具基类与注册机制首先我们需要一个统一的方式来定义和管理工具。每个工具都应该有名字、描述、参数 Schema 和一个执行函数。import json from typing import Dict, Any, Callable, Optional class Tool: 工具基类 def __init__(self, name: str, description: str, parameters: Dict[str, Any], func: Callable): self.name name self.description description self.parameters parameters # 符合 JSON Schema 的参数字典 self.func func # 实际执行的函数 def execute(self, **kwargs) - str: 执行工具返回结果字符串 try: result self.func(**kwargs) # 确保返回的是字符串方便作为 Observation 传给 LLM return str(result) except Exception as e: return fError executing tool {self.name}: {str(e)} class ToolRegistry: 工具注册表单例模式管理所有工具 _instance None def __new__(cls): if cls._instance is None: cls._instance super().__new__(cls) cls._instance._tools {} return cls._instance def register(self, tool: Tool): if tool.name in self._tools: raise ValueError(fTool {tool.name} is already registered.) self._tools[tool.name] tool def get_tool(self, name: str) - Optional[Tool]: return self._tools.get(name) def get_tools_schema(self) - list: 获取所有工具的 Schema 列表用于构造 Prompt return [ { name: tool.name, description: tool.description, parameters: tool.parameters } for tool in self._tools.values() ]3.2 第二步实现两个具体的工具让我们实现两个简单的工具一个计算器和一个网络搜索模拟器。import math # 创建注册表实例 registry ToolRegistry() # 1. 计算器工具 def calculate(expression: str) - float: 安全地计算数学表达式。警告使用 eval 有风险仅用于演示。 # 在生产环境中应使用更安全的表达式解析库如 asteval allowed_names {k: v for k, v in math.__dict__.items() if not k.startswith(_)} allowed_names.update({abs: abs, round: round}) try: # 使用限制的全局变量进行 eval result eval(expression, {__builtins__: {}}, allowed_names) return float(result) if isinstance(result, (int, float)) else result except Exception as e: raise ValueError(f无法计算表达式 {expression}: {e}) calc_tool Tool( namecalculator, description计算一个数学表达式的结果。支持 , -, *, /, **, sqrt, sin, cos 等。, parameters{ type: object, properties: { expression: { type: string, description: 数学表达式例如3 5 * 2, sqrt(16), sin(3.14/2) } }, required: [expression] }, funccalculate ) registry.register(calc_tool) # 2. 模拟搜索工具真实项目应接入搜索引擎 API def mock_web_search(query: str) - str: 模拟网络搜索返回固定结果。 knowledge_base { 刘德华 妻子: 刘德华的妻子是朱丽倩。, Python 创始人: Python 语言的创始人是吉多·范罗苏姆。, 今天天气 北京: 北京今天晴转多云气温 15-25°C南风2-3级。, 什么是 Tool Calling: Tool Calling 是一种让大语言模型能够调用外部工具或API来完成特定任务的技术。 } return knowledge_base.get(query, f未找到关于 {query} 的明确信息。) search_tool Tool( namesearch_web, description在互联网上搜索信息。当你需要获取实时、事实性或内部知识库中没有的信息时使用此工具。, parameters{ type: object, properties: { query: { type: string, description: 搜索查询关键词 } }, required: [query] }, funcmock_web_search ) registry.register(search_tool)3.3 第三步构造驱动 LLM 的 Prompt 引擎这是最核心的部分之一。我们需要精心设计一个 Prompt让 LLM 按照 ReAct 格式进行输出。def build_system_prompt(tools_schema: list) - str: 构建系统指令 Prompt tools_text \n.join([ f- {s[name]}: {s[description]} 参数: {json.dumps(s[parameters], ensure_asciiFalse)} for s in tools_schema ]) prompt f你是一个强大的助手可以调用工具来解决问题。你可以使用的工具如下 {tools_text} 请严格按照以下格式进行回应 1. 如果你需要调用工具请输出 Thought: 解释你为什么需要调用这个工具以及你的思考过程。 Action: 工具名称 Action Input: {{参数名1: 参数值1, 参数名2: 参数值2}} # 必须是严格的 JSON 对象 2. 当你收到工具的执行结果后会以 Observation: 结果 的形式提供给你。然后你可以 - 如果问题已解决直接输出最终答案Answer: 你的最终回答 - 如果还需要继续调用工具则再次输出 Thought: ... Action: ... Action Input: ... 3. 如果你不需要调用任何工具就能直接回答问题请直接输出Answer: 你的回答 重要规则 - Action Input 必须是一个**有效的 JSON 对象**且参数必须与工具定义严格匹配。 - 一次只调用一个工具。 - 如果工具执行出错请根据错误信息调整你的思路。 - 你的最终输出只能是 Thought:、Action:、Action Input: 或 Answer: 开头的行。 现在开始。 return prompt3.4 第四步实现 LLM 调用与输出解析器我们需要一个函数来调用 LLM这里以 OpenAI API 为例并严格解析它的输出提取出 Thought, Action 等信息。import openai # 需要安装 openai 库 import re class LLMClient: def __init__(self, api_key: str, model: str gpt-3.5-turbo): self.client openai.OpenAI(api_keyapi_key) self.model model def chat_completion(self, messages: list) - str: response self.client.chat.completions.create( modelself.model, messagesmessages, temperature0.1, # 低温度保证输出格式稳定 streamFalse ) return response.choices[0].message.content def parse_llm_output(output: str) - Dict[str, str]: 解析 LLM 的输出提取 Thought, Action, Action Input 或 Answer。 result {type: None, thought: , action: , action_input: , answer: } # 使用正则表达式匹配关键行 thought_match re.search(r^Thought:\s*(.)$, output, re.MULTILINE) action_match re.search(r^Action:\s*(\w)$, output, re.MULTILINE) action_input_match re.search(r^Action Input:\s*(.)$, output, re.MULTILINE | re.DOTALL) answer_match re.search(r^Answer:\s*(.)$, output, re.MULTILINE | re.DOTALL) if answer_match: result[type] answer result[answer] answer_match.group(1).strip() elif action_match and action_input_match: result[type] action result[thought] thought_match.group(1).strip() if thought_match else result[action] action_match.group(1).strip() try: # 尝试解析 Action Input 为 JSON input_json json.loads(action_input_match.group(1).strip()) result[action_input] input_json except json.JSONDecodeError: # 如果解析失败保留原始字符串后续会报错 result[action_input] action_input_match.group(1).strip() result[type] error result[error] Action Input 不是有效的 JSON。 else: # 如果格式完全不符合可能 LLM 直接给出了答案 result[type] direct_answer result[answer] output.strip() return result3.5 第五步组装 ReAct 智能体主循环现在我们把所有零件组装起来形成一个可以运行的智能体。class SimpleReActAgent: def __init__(self, llm_client: LLMClient, max_steps: int 10): self.llm llm_client self.registry ToolRegistry() self.max_steps max_steps self.conversation_history [] # 简单的对话记忆 def run(self, user_query: str) - str: print(f用户: {user_query}) # 初始化对话 tools_schema self.registry.get_tools_schema() system_prompt build_system_prompt(tools_schema) messages [ {role: system, content: system_prompt}, {role: user, content: user_query} ] for step in range(self.max_steps): print(f\n--- 第 {step 1} 步 ---) # 1. 调用 LLM llm_output self.llm.chat_completion(messages) print(fLLM 原始输出:\n{llm_output}) # 2. 解析输出 parsed parse_llm_output(llm_output) # 3. 处理解析结果 if parsed[type] answer or parsed[type] direct_answer: final_answer parsed[answer] print(f智能体最终回答: {final_answer}) return final_answer elif parsed[type] action: tool_name parsed[action] tool self.registry.get_tool(tool_name) if not tool: observation fError: 工具 {tool_name} 不存在。 else: # 执行工具 try: # 确保 action_input 是字典 params parsed[action_input] if isinstance(params, str): # 如果之前解析失败这里再尝试一次 try: params json.loads(params) except: observation fError: 工具参数不是有效的 JSON: {params} if isinstance(params, dict): observation tool.execute(**params) else: observation fError: 工具参数必须是字典类型。 except Exception as e: observation fError executing tool: {str(e)} print(f工具执行结果 (Observation): {observation}) # 将本次交互加入历史用于下一轮 messages.append({role: assistant, content: llm_output}) messages.append({role: user, content: fObservation: {observation}}) elif parsed[type] error: observation fError: {parsed.get(error, 未知格式错误)} print(f解析错误: {observation}) messages.append({role: assistant, content: llm_output}) messages.append({role: user, content: fObservation: {observation}\n请检查你的输出格式Action Input 必须是有效的 JSON。}) else: # 意外情况终止循环 observation Error: 无法理解你的输出格式。 messages.append({role: assistant, content: llm_output}) messages.append({role: user, content: fObservation: {observation}}) # 循环结束仍未给出答案 return 抱歉我经过多次尝试仍未能解决这个问题。3.6 第六步运行你的第一个智能体现在让我们初始化并运行它。# 配置你的 OpenAI API Key import os # 假设你的 API Key 存储在环境变量中 api_key os.getenv(OPENAI_API_KEY) if not api_key: print(请设置 OPENAI_API_KEY 环境变量) # 为了演示我们用一个模拟的 LLM 客户端来避免真实调用 class MockLLMClient: def chat_completion(self, messages): # 模拟一个简单查询的响应 last_user_msg messages[-1][content] if 天气 in last_user_msg: return Thought: 用户想了解天气信息我需要使用搜索工具。 Action: search_web Action Input: {query: 今天天气 北京} elif 妻子 in last_user_msg: return Thought: 这是一个关于人物事实的问题使用搜索工具获取准确信息。 Action: search_web Action Input: {query: 刘德华 妻子} elif 计算 in last_user_msg or sqrt in last_user_msg: return Thought: 这是一个数学计算问题使用计算器工具。 Action: calculator Action Input: {expression: 3 5 * 2} else: return Answer: 我目前无法处理这个请求。 llm_client MockLLMClient() else: llm_client LLMClient(api_keyapi_key, modelgpt-3.5-turbo) # 创建并运行智能体 agent SimpleReActAgent(llm_clientllm_client, max_steps5) # 测试查询 queries [ 刘德华的妻子是谁, 计算一下 3 加 5 乘以 2 等于多少, 北京今天天气怎么样 ] for query in queries: print(\n *50) answer agent.run(query) print(*50)运行上述代码你将看到智能体一步步思考、调用工具、获取结果并最终给出答案的过程。虽然我们用了模拟的 LLM 和工具但整个架构和流程与真实环境完全一致。4. 迈向框架抽象设计可扩展的 Agent 核心架构手写版本跑通了但我们很快会发现它的局限性工具管理简陋、记忆只有简单列表、无法处理复杂对话历史、没有错误恢复机制、难以接入不同的 LLM 提供商。是时候进行抽象设计一个更健壮、更灵活的框架雏形了。4.1 抽象层设计定义核心接口一个好的抽象应该定义清晰的接口让具体实现可以灵活替换。我们至少需要抽象出以下几个部分from abc import ABC, abstractmethod from typing import List, Dict, Any, Optional class LLMProvider(ABC): LLM 提供商抽象接口 abstractmethod def generate(self, messages: List[Dict], tools: Optional[List] None) - Dict: 生成回复。 返回字典应至少包含: {content: str, tool_calls: Optional[List]} pass class Memory(ABC): 记忆抽象接口 abstractmethod def add_message(self, role: str, content: str): 添加一条消息到记忆 pass abstractmethod def get_messages(self, limit: Optional[int] None) - List[Dict]: 获取历史消息 pass abstractmethod def clear(self): 清空记忆 pass class Tool(ABC): 工具抽象接口对手写版本的增强 property abstractmethod def name(self) - str: pass property abstractmethod def description(self) - str: pass property abstractmethod def schema(self) - Dict: # 返回标准的 OpenAPI Tool 格式 pass abstractmethod def execute(self, **kwargs) - Any: pass class Agent(ABC): 智能体抽象接口 abstractmethod def run(self, input_text: str) - str: pass4.2 实现一个基于 OpenAI 格式的 LLM 适配器现代 LLM API如 OpenAI, Anthropic已经原生支持 Tool Calling 功能。我们应该利用这一点而不是完全依赖 Prompt 工程。import openai class OpenAIToolCallingProvider(LLMProvider): def __init__(self, api_key: str, model: str gpt-3.5-turbo-1106): self.client openai.OpenAI(api_keyapi_key) self.model model def generate(self, messages: List[Dict], tools: Optional[List[Tool]] None) - Dict: openai_tools None if tools: # 将我们的 Tool 对象转换为 OpenAI 的 tool 格式 openai_tools [{ type: function, function: { name: tool.name, description: tool.description, parameters: tool.schema } } for tool in tools] response self.client.chat.completions.create( modelself.model, messagesmessages, toolsopenai_tools, tool_choiceauto if tools else None, # 让模型自行决定是否调用工具 ) message response.choices[0].message result { content: message.content, tool_calls: [] } if message.tool_calls: for tc in message.tool_calls: result[tool_calls].append({ id: tc.id, name: tc.function.name, arguments: json.loads(tc.function.arguments) # 注意这里 API 已经保证了是合法 JSON }) return result核心优势使用原生 Tool CallingLLM 输出的tool_calls字段已经是结构化的数据完全无需我们之前写的复杂解析器。这大大提升了稳定性和开发效率。这也是为什么现代框架都采用这种方式。4.3 实现一个简单的对话记忆管理手写版本只用了一个列表现在我们需要一个更结构化的记忆管理能够处理长上下文。class SimpleConversationMemory(Memory): def __init__(self, max_turns: int 20): self.messages [] self.max_turns max_turns * 2 # 因为每轮有 user 和 assistant 两条消息 def add_message(self, role: str, content: str): self.messages.append({role: role, content: content}) # 简单的截断策略保留最近 N 条消息 if len(self.messages) self.max_turns: # 保留系统消息和最近的对话 system_msg [m for m in self.messages if m[role] system] recent_msgs self.messages[-self.max_turns:] self.messages system_msg recent_msgs def get_messages(self, limit: Optional[int] None) - List[Dict]: if limit: return self.messages[-limit:] return self.messages.copy() def clear(self): self.messages []4.4 组装高级 ReAct 智能体现在我们用抽象后的组件重新构建智能体。class AdvancedReActAgent(Agent): def __init__(self, llm_provider: LLMProvider, tools: List[Tool], memory: Optional[Memory] None): self.llm llm_provider self.tools {tool.name: tool for tool in tools} self.memory memory or SimpleConversationMemory() # 初始化系统提示 self.system_prompt 你是一个有帮助的助手可以调用工具来解决问题。请根据需要使用提供的工具。 def run(self, input_text: str) - str: # 1. 将用户输入加入记忆 self.memory.add_message(user, input_text) # 2. 准备对话历史和工具列表 messages self.memory.get_messages() if not any(m[role] system for m in messages): messages.insert(0, {role: system, content: self.system_prompt}) max_iterations 10 for i in range(max_iterations): # 3. 调用 LLM llm_response self.llm.generate(messagesmessages, toolslist(self.tools.values())) # 4. 处理 LLM 响应 assistant_message {role: assistant, content: llm_response[content]} tool_calls_for_memory [] if llm_response.get(tool_calls): assistant_message[tool_calls] llm_response[tool_calls] # 执行每个工具调用 for tc in llm_response[tool_calls]: tool_name tc[name] tool_args tc[arguments] if tool_name not in self.tools: tool_result fError: Tool {tool_name} not found. else: try: tool_result self.tools[tool_name].execute(**tool_args) except Exception as e: tool_result fError executing tool {tool_name}: {str(e)} # 将工具执行结果构造为“工具”角色的消息 tool_message { role: tool, content: tool_result, tool_call_id: tc[id] # 用于关联调用和结果 } messages.append(tool_message) self.memory.add_message(tool, tool_result) # 简化记忆 tool_calls_for_memory.append((tool_name, tool_args, tool_result)) # 将助手的回复加入记忆 self.memory.add_message(assistant, llm_response[content]) # 5. 判断是否结束 # 如果没有工具调用或者有最终答案则结束 if not llm_response.get(tool_calls) and llm_response[content]: return llm_response[content] # 如果达到最大迭代次数 if i max_iterations - 1: return 已达到最大思考步骤未能得出最终结论。 # 6. 将助手消息加入下一轮的消息列表 messages.append(assistant_message)这个版本的智能体更加健壮利用了 LLM 的原生工具调用能力记忆管理也更合理。它已经具备了现代 AI Agent 框架的核心雏形。5. 实战避坑指南与高级技巧在真实项目中你会遇到比示例复杂得多的情况。以下是我从实际项目中总结出的血泪经验。5.1 工具设计中的常见陷阱Schema 描述模糊不清这是最常见的错误。description字段写得太简单AI 无法准确理解何时使用以及如何传参。反面教材“获取数据”正确做法“根据用户ID从用户数据库中查询该用户的姓名、注册邮箱和账户状态。用户ID应为整数。”工具粒度过粗或过细过粗一个“处理用户请求”的工具内部包含几十个 if-else 分支。这会让 AI 难以驾驭也违背了单一职责原则。过细为“加一”、“减一”分别创建工具。这会导致工具列表膨胀增加 AI 的选择难度和上下文长度。原则一个工具应对应一个原子性的、可复用的操作。例如“创建订单”、“查询物流”、“发送邮件”都是不错的粒度。错误处理缺失工具执行时可能遇到各种错误网络超时、参数无效、权限不足。必须在工具内部做好错误捕获并返回对 AI友好、可操作的错误信息。不要只返回“Error: 500 Internal Server Error”应该返回“查询失败用户服务暂时不可用HTTP 500。建议稍后重试或检查用户ID是否正确。”5.2 提升 Tool Calling 稳定性的 Prompt 技巧即使使用原生 APIPrompt 依然重要。在系统指令中加入以下约束可以显著提升稳定性请你严格遵循以下规则 1. 如果你决定调用工具**必须且只能**使用我提供的工具列表中的工具。 2. 调用工具时请确保 Action Input 中的参数名称、类型与工具定义**完全一致**。例如如果工具要求 user_id下划线不要使用 userId驼峰。 3. 如果一个工具调用失败了请先仔细阅读错误信息思考失败原因再决定是重试、换一个工具还是向用户请求澄清。 4. 在得到所有必要信息之前不要急于给出最终答案。5.3 复杂工作流的编排超越简单 ReAct对于“查天气如果下雨就发提醒”这类多条件依赖的任务简单循环可能不够。我们需要引入更高级的工作流概念。这里介绍一个极其简单但有效的“目标-子任务”分解模式。class PlanningAgent: 一个能进行简单任务规划的智能体 def __init__(self, llm_provider, tools): self.core_agent AdvancedReActAgent(llm_provider, tools) self.planner_prompt 你的任务是将一个复杂请求分解成一系列清晰的、可顺序执行的子任务。 每个子任务应该可以直接由可用的工具或你的知识处理。 以 JSON 列表格式输出每个元素是一个子任务描述。 例如用户请求“查北京天气如果下雨就提醒我带伞。” 输出[ “使用 search_web 工具查询北京今天的天气情况。”, “分析天气查询结果判断是否包含‘雨’。”, “如果判断结果为下雨则调用 send_reminder 工具发送提醒内容‘今天有雨请带伞’。” ] 现在请分解以下请求 {user_input} def run(self, user_input: str): # 第一步规划 plan_messages [{role: user, content: self.planner_prompt.format(user_inputuser_input)}] plan_response self.llm.generate(plan_messages, tools[]) # 解析出子任务列表 (这里简化实际需要解析 JSON) import ast try: sub_tasks ast.literal_eval(plan_response[content]) except: sub_tasks [plan_response[content]] # 第二步串行执行子任务 context {} for task in sub_tasks: # 将子任务和当前上下文信息一起交给执行智能体 enriched_query f背景信息{context}。当前任务{task} result self.core_agent.run(enriched_query) context[f任务结果: {task}] result # 这里可以加入逻辑判断根据结果决定是否继续 # 第三步汇总结果 final_messages [ {role: user, content: user_input}, {role: assistant, content: f我已经按步骤执行了任务中间结果如下{context}}, {role: user, content: 请根据以上所有步骤的执行结果给用户一个完整、清晰的最终答复。} ] final_response self.llm.generate(final_messages, tools[]) return final_response[content]这个模式将“规划”与“执行”分离让 LLM 先当“项目经理”拆解任务再当“工程师”逐个完成非常适合逻辑链条长的任务。5.4 评估与监控你的智能体真的可靠吗上线前必须对智能体进行系统化评估。构建测试集覆盖常见问题、边界情况和之前出错的案例。定义评估指标工具调用准确率AI 是否在正确的时机调用了正确的工具参数填充准确率调用工具时参数是否正确任务完成率最终是否给出了用户满意的答案平均交互轮数完成任务需要多少步轮数过多可能意味着效率低下或陷入循环。实现自动化测试编写脚本用测试集批量运行智能体并自动计算上述指标。加入人工审核与干预对于关键任务如金融交易、内容发布设计“人工确认”环节让 AI 在最终执行前将计划提交给人审核。从零手写到框架抽象的过程本质上是一个从“知其然”到“知其所以然”的深度探索。你不再是一个只会调用agent.run()的 API 用户而是成为了能洞察其内部运转、能随心所欲定制和优化的架构师。当你在未来面对 LangChain、Dify 甚至更复杂的框架时这份亲手搭建的经验会让你一眼看穿其核心设计快速定位问题并自信地将其改造成最适合你业务的样子。这才是深入技术本质带来的真正自由。