1. 从概念到代码一个最简化AI Agent的构建动机最近和不少同行交流发现大家一提到AI Agent脑海里浮现的要么是AutoGPT、LangChain这类庞大框架要么是各种“智能体平台”的复杂概念。框架虽好但有时也像黑箱把核心逻辑包裹得严严实实让人知其然不知其所以然。很多朋友想入门却被“工具调用”、“记忆”、“规划”这些术语劝退感觉无从下手。其实一个能自主思考、使用工具、并记住上下文的AI Agent其最核心的骨架远比想象中简单。今天我就想抛开那些重型框架带你从零开始亲手搭建一个最简化、但五脏俱全的AI Agent实现。这个实现将聚焦四个最核心的模块ReAct思考循环、工具调用、可观测性和持久化。通过这个“麻雀虽小五脏俱全”的项目你不仅能透彻理解Agent的工作机制更能获得一个可以随意扩展、用于实验和学习的坚实基础。无论你是想面试时侃侃而谈React Agent的原理还是想为自己的项目快速集成一个智能助手这篇手把手的指南都将为你提供清晰的路径和可直接运行的代码。2. 核心架构拆解理解“最简化”Agent的四大支柱在开始写代码之前我们必须先统一认知什么是一个“最简化”但“可用”的AI Agent它不应该依赖任何特定的庞大框架其核心能力应该由几个清晰、解耦的模块协同完成。基于这个思路我将其核心抽象为以下四个支柱它们共同构成了Agent的“大脑”与“身体”。2.1 ReAct循环Agent的“思考”引擎ReActReasoning Acting是让Agent变得“智能”的关键范式。它不是一个复杂的算法而是一个清晰的思维模式。你可以把它想象成一个优秀工程师的Debug过程观察Observe先看当前的状态和问题比如用户提问、工具执行结果、系统状态。思考Reason基于观察在大脑LLM里分析一下“我现在知道了什么我还需要什么下一步应该做什么”行动Act根据思考的结论执行一个具体的动作比如调用一个计算器工具或者查询数据库。循环将行动的结果作为新的“观察”回到第一步继续思考直到问题被解决或达到终止条件。这个循环的精髓在于它将“想”和“做”分开了。LLM只负责“想”生成包含思考过程和行动指令的文本而执行系统负责“做”解析指令并调用工具。这样既发挥了LLM的推理优势又通过工具弥补了其事实性和实时性的不足。在我们的最简化实现中ReAct循环就是那个驱动一切的主控程序。2.2 工具调用Agent的“手脚”与外部世界交互LLM本身是“纸上谈兵”的专家但它无法直接操作计算器、搜索网络、读写文件。工具调用Tool Calling就是为LLM装上“手脚”。一个工具本质上是一个函数它有明确的名称、描述、参数格式。当LLM在“思考”阶段决定要使用工具时它会按照预定格式如JSON输出一个工具调用请求。我们的系统会解析这个请求找到对应的工具函数并执行然后将执行结果返回给LLM供其下一轮思考使用。例如我们可以定义一个calculator工具描述为“执行数学计算”。当用户问“123乘以456等于多少”时LLM在ReAct循环中可能会输出“我需要计算123*456。我将调用计算器工具。” 然后生成一个结构化的调用请求{“action”: “calculator”, “action_input”: {“expression”: “123*456”}}。系统执行计算返回结果“56088”LLM再基于这个结果组织最终答案。2.3 可观测性Agent的“透明手术台”当你运行一个Agent尤其是它开始自主循环时最让人头疼的就是“它现在到底在想什么它做了什么决定为什么卡住了” 这就是可观测性Observability要解决的问题。它不同于简单的打印日志而是需要结构化的、分层次的洞察能力。一个具备良好可观测性的Agent系统应该能让我们看到思维链Chain of ThoughtLLM每一轮的完整推理文本包括其内心的“独白”和最终的决定。行动历史Action History它调用了哪些工具传入什么参数返回了什么结果。状态快照State Snapshot当前循环的轮次、使用的提示词模板、会话的元数据等。性能指标Metrics单轮耗时、Token消耗、工具调用成功率等。在我们的实现中我们会设计一个轻量级的“事件总线”或“回调系统”让这些内部状态能够以结构化的方式暴露出来方便我们实时监控、调试甚至在出现异常时进行干预。这是开发稳定、可靠Agent的必备基础设施也是面试中常被问到的“如何监控你的Agent”这一问题的实践答案。2.4 持久化Agent的“记忆”与状态留存一个没有记忆的Agent每次对话都是全新的开始这显然不符合“智能助手”的预期。持久化Persistence就是为了让Agent拥有“记忆”。这里的记忆分为几个层面会话记忆Conversation Memory记住当前对话的历史以便理解上下文。这是最基本的需求。长期记忆Long-term Memory跨越会话记住一些关键信息比如用户的偏好、之前达成的结论等。这通常需要向量数据库等外部存储。状态持久化State Persistence将Agent的完整运行状态包括当前循环索引、工具调用历史、内部变量等保存下来。这在Agent需要长时间运行或可能被中断时至关重要可以实现“暂停”和“继续”。在我们的最简化版本中我们会先实现最核心的会话记忆和状态快照保存/加载功能。这能让我们的Agent在单次运行中保持连贯性并且允许我们将一个复杂的任务状态保存到文件或数据库下次接着运行。这是构建真正可用Agent服务的基础。3. 手把手实现从零搭建最简化AI Agent理论讲完了现在进入最激动人心的实操环节。我们将使用Python不依赖LangChain等重型框架仅用openai库或兼容OpenAI API的库和一些基础数据结构构建出这个四合一的核心Agent。3.1 第一步定义工具系统——Agent的能力集工具系统是Agent的基石。我们先定义一个工具的基类然后实现几个常用工具。import json import requests from datetime import datetime from typing import Any, Dict, Callable class Tool: 工具基类定义工具的接口 def __init__(self, name: str, description: str, func: Callable): self.name name self.description description self.func func def run(self, **kwargs) - str: 执行工具返回字符串格式的结果 try: result self.func(**kwargs) return str(result) except Exception as e: return fError executing tool {self.name}: {str(e)} def to_dict(self) - Dict[str, Any]: 返回工具的元信息用于构造提示词 return { name: self.name, description: self.description, # 这里可以进一步细化参数schema为了简化先省略 } # 实现几个具体的工具 def calculator(expression: str) - float: 执行安全的数学表达式计算使用eval需谨慎此处仅作演示 # 警告在生产环境中应对expression做严格的安全检查和过滤 allowed_chars set(0123456789-*/(). ) if not all(c in allowed_chars for c in expression): raise ValueError(Expression contains invalid characters) return eval(expression) def get_current_time(*args, **kwargs) - str: 获取当前日期和时间 return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def search_web(query: str) - str: 模拟网络搜索实际可接入SerperAPI、Google Search等 # 此处为模拟真实情况需要调用搜索API print(f[模拟搜索] 搜索关键词: {query}) # 模拟返回一些结果 return f关于{query}的模拟搜索结果相关文章1相关文章2。 # 创建工具集 TOOLS { calculator: Tool(calculator, 执行数学计算例如3 5 * 2。, calculator), get_current_time: Tool(get_current_time, 获取当前的日期和时间。, get_current_time), search_web: Tool(search_web, 在互联网上搜索信息。需要提供搜索查询词。, search_web), }为什么这样设计统一的接口每个工具都有run方法Agent核心无需关心工具内部实现。自描述性to_dict方法提供了工具的元信息可以动态地插入到给LLM的提示词中告诉LLM现在有哪些工具可用。安全性考虑在calculator工具中我们简单演示了输入检查。在实际生产环境中直接使用eval是极度危险的必须替换为安全的表达式解析库如ast.literal_eval配合自定义解析器或沙箱环境。3.2 第二步设计ReAct提示词与解析逻辑——Agent的“思维模板”接下来我们需要设计一个提示词Prompt引导LLM按照ReAct的格式进行输出。同时我们需要一个解析器从LLM的回复中提取出“思考”和“行动”。def create_react_prompt(question: str, history: str, available_tools: list) - str: 构造ReAct格式的提示词 tools_desc \n.join([f- {t[name]}: {t[description]} for t in available_tools]) prompt f你是一个善于思考并可以使用工具的AI助手。你的任务是通过思考Thought、行动Action、观察Observation的循环来回答问题。 你可以使用的工具如下 {tools_desc} 对话历史 {history} 当前问题{question} 请严格按照以下格式回应 Thought: 你需要在这里描述你对问题的思考过程分析当前情况并决定下一步做什么。 Action: 你需要在这里指定要调用的工具名称。必须是以下之一{[t[name] for t in available_tools]}。 Action Input: 你需要在这里以JSON格式提供调用工具所需的输入参数。 或者如果你认为问题已经解决不需要再使用工具则使用以下格式 Thought: 我通过之前的步骤已经得到了答案。 Final Answer: 你的最终答案放在这里。 现在开始 return prompt def parse_llm_response(response: str) - Dict[str, Any]: 解析LLM的回复提取Thought, Action, Action Input 或 Final Answer lines response.strip().split(\n) result {thought: , action: None, action_input: None, final_answer: None} current_key None for line in lines: line line.strip() if line.startswith(Thought:): current_key thought result[current_key] line.replace(Thought:, , 1).strip() elif line.startswith(Action:): current_key action result[current_key] line.replace(Action:, , 1).strip() elif line.startswith(Action Input:): current_key action_input input_str line.replace(Action Input:, , 1).strip() # 尝试解析JSON try: result[current_key] json.loads(input_str) except json.JSONDecodeError: # 如果解析失败可能LLM没有输出严格JSON这里做简单处理 result[current_key] input_str elif line.startswith(Final Answer:): current_key final_answer result[current_key] line.replace(Final Answer:, , 1).strip() elif current_key and line: # 处理多行内容 result[current_key] \n line return result设计要点与避坑指南格式强制提示词中必须明确、严格地规定输出格式。LLM有时会“放飞自我”清晰的格式能极大提高解析成功率。工具描述清晰工具的描述description至关重要它是LLM决定是否及如何使用该工具的主要依据。描述应简洁、准确并包含参数示例。解析的鲁棒性parse_llm_response函数需要有一定的容错能力。LLM的输出可能有多余的空行、格式微调。我们的解析逻辑先按行分割再根据关键词提取并对JSON解析做了异常处理这是一个在实践中很有效的简单策略。3.3 第三步实现可观测性模块——给Agent装上“仪表盘”可观测性模块负责收集和展示Agent运行过程中的所有关键信息。我们实现一个简单的事件记录器。class AgentObserver: Agent可观测性模块记录运行全过程 def __init__(self): self.episodes [] # 记录每一轮Thought-Action-Observation的信息 self.current_episode {} def on_thought_start(self, question, prompt): 开始新一轮思考时调用 self.current_episode { question: question, prompt_snippet: prompt[:200] ... if len(prompt) 200 else prompt, # 记录部分提示词 thought: , action: , action_input: , observation: , final_answer: , timestamp: datetime.now().isoformat(), token_usage: {} # 可在此处记录token消耗 } def on_thought_generated(self, thought): 记录思考内容 self.current_episode[thought] thought def on_action_decided(self, action, action_input): 记录决策的行动 self.current_episode[action] action self.current_episode[action_input] action_input def on_observation_received(self, observation): 记录工具执行后的观察 self.current_episode[observation] observation def on_final_answer(self, answer): 记录最终答案 self.current_episode[final_answer] answer # 本轮结束存入历史 self.episodes.append(self.current_episode.copy()) def get_latest_episode(self): 获取最新一轮的详细信息 return self.current_episode def get_summary(self): 获取运行摘要 summary { total_episodes: len(self.episodes), tools_used: {}, last_question: self.episodes[-1][question] if self.episodes else None, last_answer: self.episodes[-1][final_answer] if self.episodes else None, } for ep in self.episodes: tool ep.get(action) if tool: summary[tools_used][tool] summary[tools_used].get(tool, 0) 1 return summary def print_trace(self): 以友好格式打印完整的思维轨迹用于调试 print(\n *60) print(AI Agent 思维轨迹追踪) print(*60) for i, ep in enumerate(self.episodes): print(f\n[第 {i1} 轮]) print(f 问题: {ep[question]}) print(f 思考: {ep[thought]}) if ep[action]: print(f 行动: {ep[action]}({ep[action_input]})) print(f 观察: {ep[observation]}) if ep[final_answer]: print(f 最终答案: {ep[final_answer]}) print(*60)这个观察器有什么用调试神器当Agent行为不符合预期时调用observer.print_trace()可以一目了然地看到它每一步的“想法”和“行动”快速定位是提示词问题、工具问题还是LLM理解问题。监控看板get_summary()方法可以提供关键指标如工具调用频率、循环轮次未来可以接入Grafana等监控系统。审计日志所有episodes记录了完整的历史可用于复盘分析或满足合规性要求。3.4 第四步集成持久化模块——赋予Agent“记忆”现在我们让Agent能够记住对话历史并能保存/加载整个会话状态。import pickle from typing import List class ConversationMemory: 简单的对话记忆管理上下文 def __init__(self, max_turns10): self.messages: List[Dict] [] # 存储{role: user/assistant, content: ...} self.max_turns max_turns # 控制上下文长度防止token超限 def add_user_message(self, content: str): self.messages.append({role: user, content: content}) self._trim_memory() def add_assistant_message(self, content: str): self.messages.append({role: assistant, content: content}) self._trim_memory() def _trim_memory(self): 修剪记忆保留最近的对话轮次 if len(self.messages) self.max_turns * 2: # 每轮包含user和assistant两条 # 保留系统提示如果有和最近的对话 self.messages self.messages[-self.max_turns*2:] def get_history_for_prompt(self) - str: 将记忆格式化为提示词中可用的历史字符串 history_str for msg in self.messages[-self.max_turns*2:]: # 确保不超过最大轮次 role Human if msg[role] user else Assistant history_str f{role}: {msg[content]}\n return history_str.strip() def clear(self): self.messages.clear() class PersistentAgentState: Agent持久化状态可保存到文件或数据库 def __init__(self, memory: ConversationMemory, observer: AgentObserver, other_stateNone): self.memory memory self.observer_data observer.episodes # 保存观察器的历史记录 self.other_state other_state or {} # 可扩展其他状态 self.saved_at datetime.now().isoformat() def save(self, filepath: str): 将状态保存到文件 with open(filepath, wb) as f: pickle.dump(self, f) print(fAgent状态已保存至: {filepath}) staticmethod def load(filepath: str) - PersistentAgentState: 从文件加载状态 with open(filepath, rb) as f: state pickle.load(f) print(fAgent状态已从 {filepath} 加载保存时间: {state.saved_at}) return state记忆与持久化的关键考量上下文长度管理ConversationMemory中的max_turns和_trim_memory方法至关重要。LLM有上下文窗口限制无限制地增长历史会导致后续请求失败或成本激增。常见的策略是保留最近N轮对话或使用更复杂的“摘要记忆”方式将久远的历史压缩成一段摘要。状态序列化我们使用pickle进行简单演示它可以将整个Python对象保存到文件。但在生产环境中pickle存在安全风险可能执行恶意代码且与语言/框架绑定。更安全的做法是将状态转化为安全的字典结构如JSON或使用ORM框架直接存入数据库。状态恢复加载状态后需要将memory和observer_data重新注入到运行中的Agent实例使其能“无缝”继续工作。这为实现长时间运行的任务如自动化工作流提供了可能。3.5 第五步组装核心ReAct循环——让Agent“动”起来最后我们将所有模块组装起来实现主循环逻辑。import openai # 或其他兼容OpenAI API的客户端 class SimpleReActAgent: 最简化的ReAct Agent核心实现 def __init__(self, llm_client, modelgpt-3.5-turbo, max_iterations5): self.llm_client llm_client self.model model self.max_iterations max_iterations # 防止无限循环 self.tools TOOLS self.memory ConversationMemory(max_turns5) self.observer AgentObserver() self.available_tools_info [tool.to_dict() for tool in self.tools.values()] def run(self, query: str) - str: 执行ReAct循环处理用户查询 print(f\n[Agent] 开始处理查询: {query}) self.memory.add_user_message(query) for i in range(self.max_iterations): print(f\n--- ReAct循环 第 {i1} 轮 ---) # 1. 准备提示词包含历史 history self.memory.get_history_for_prompt() prompt create_react_prompt(query, history, self.available_tools_info) # 记录可观测性事件 self.observer.on_thought_start(query, prompt) # 2. 调用LLM进行思考 try: response self.llm_client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], temperature0.1, # 低温度保证输出格式稳定 ) llm_output response.choices[0].message.content print(fLLM原始输出:\n{llm_output}) except Exception as e: return f调用LLM API时出错: {str(e)} # 3. 解析LLM输出 parsed parse_llm_response(llm_output) self.observer.on_thought_generated(parsed.get(thought, )) # 4. 判断是否结束 if parsed.get(final_answer): final_answer parsed[final_answer] print(f[Agent] 最终答案: {final_answer}) self.memory.add_assistant_message(final_answer) self.observer.on_final_answer(final_answer) # 打印完整的思维轨迹 self.observer.print_trace() return final_answer # 5. 执行工具调用 action parsed.get(action) action_input parsed.get(action_input) if not action or action not in self.tools: error_msg fLLM指定了无效或未知的行动: {action} print(f[错误] {error_msg}) # 可以将错误信息作为Observation反馈给LLM让其重试 observation error_msg else: print(f[Agent] 执行工具: {action}, 输入: {action_input}) self.observer.on_action_decided(action, action_input) # 执行工具 tool self.tools[action] if isinstance(action_input, dict): observation tool.run(**action_input) else: # 如果action_input不是dict尝试作为字符串参数传递 observation tool.run(action_input) print(f[工具 {action}] 返回: {observation}) self.observer.on_observation_received(observation) # 6. 将观察结果加入记忆准备下一轮循环 # 注意这里我们将“思考-行动-观察”作为一个整体回合加入记忆简化处理 # 更精细的做法可以分别记录 assistant_message fThought: {parsed.get(thought)}\nAction: {action}\nObservation: {observation} self.memory.add_assistant_message(assistant_message) # 循环达到最大次数仍未得出最终答案 timeout_msg f经过{self.max_iterations}轮思考仍未得出结论。 print(f[Agent] {timeout_msg}) self.observer.on_final_answer(timeout_msg) self.observer.print_trace() return timeout_msg def save_state(self, filepath: str): 保存当前Agent状态记忆、观察历史 state PersistentAgentState(self.memory, self.observer) state.save(filepath) def load_state(self, filepath: str): 加载Agent状态 state PersistentAgentState.load(filepath) self.memory state.memory self.observer.episodes state.observer_data print(状态加载完成记忆和观察历史已恢复。)主循环的核心逻辑与避坑点循环终止条件必须有max_iterations限制这是防止Agent陷入死循环或产生过高成本的安全阀。同时依赖LLM自己输出Final Answer来正常结束。错误处理在工具调用和LLM API调用环节必须有充分的try-except。特别是当LLM输出格式不符合预期或指定了不存在的工具时要有降级处理如将错误信息作为observation反馈回去让LLM自我纠正。温度Temperature设置在ReAct这类需要严格遵循输出格式的场景建议设置较低的temperature如0.1或0以增加输出的确定性和格式稳定性。记忆的粒度示例中将一整轮“Thought-Action-Observation”作为一条助手消息存入记忆。在实际复杂任务中你可能需要更精细的控制比如只将关键的observation和final_answer存入长期对话记忆而将中间思考过程仅用于可观测性记录。4. 运行示例与深度调试看Agent如何“思考”让我们用一个具体的例子来完整跑通这个Agent并利用可观测性模块深入其“内心世界”。# 假设你已经配置好了OpenAI API客户端 # client openai.OpenAI(api_keyyour-api-key) # 1. 初始化Agent agent SimpleReActAgent(llm_clientclient, modelgpt-3.5-turbo) # 2. 运行一个需要多步推理和工具调用的查询 question “今天是2024年5月20日请问150天后的星期几另外那天北京的天气怎么样模拟” answer agent.run(question) print(f\n最终回复给用户: {answer}) # 3. 保存当前会话状态模拟长时间任务中断 agent.save_state(agent_session_20240520.pkl) # 4. 模拟在另一个进程或时间点加载状态并继续 print(\n--- 模拟重新加载Agent状态并继续对话 ---) new_agent SimpleReActAgent(llm_clientclient, modelgpt-3.5-turbo) new_agent.load_state(agent_session_20240520.pkl) follow_up_question “把刚才计算出来的那天加上‘国际日’后缀重新告诉我日期。” follow_up_answer new_agent.run(follow_up_question) print(f\n后续问题回复: {follow_up_answer})预期输出与观察当你运行上述代码通过observer.print_trace()或直接查看控制台打印你会看到类似以下的思维轨迹[Agent] 开始处理查询: 今天是2024年5月20日请问150天后的星期几另外那天北京的天气怎么样模拟 --- ReAct循环 第 1 轮 --- LLM原始输出: Thought: 用户问了两个问题。第一个是计算150天后的日期和星期几这需要计算。第二个是问那天的天气这需要搜索。我应该先解决第一个问题因为它是一个确定性的计算然后再搜索天气。 Action: calculator Action Input: {expression: 150} ... [工具 calculator] 返回: 150 --- ReAct循环 第 2 轮 --- LLM原始输出: Thought: 我得到了150。但我需要从2024年5月20日开始计算150天后的具体日期。这涉及到日期运算简单的计算器工具可能不够。我需要一个能处理日期的工具或者用另一种方式。我意识到我可能没有日期计算工具。我可以先估算或者告诉用户我无法精确计算日期。但也许我可以先搜索“2024年5月20日 150天后”看看有没有在线计算器或信息。我先搜索一下。 Action: search_web Action Input: {query: 2024年5月20日 150天后 日期 计算} ... [工具 search_web] 返回: 关于2024年5月20日 150天后 日期 计算的模拟搜索结果相关文章1相关文章2。 --- ReAct循环 第 3 轮 --- LLM原始输出: Thought: 搜索返回了模拟结果但没有给出具体日期。我需要换一种思路。我可以手动计算5月有31天从5月20日到5月31日是11天。6月30天7月31天8月31天。11303131103天。150-10347天。9月有30天所以47天后是10月17日等等我需要更精确。这太容易出错了。我最好承认这个限制并专注于我能回答的部分——天气查询。我可以先回答天气部分并说明日期计算需要更专业的工具。 Action: search_web Action Input: {query: 2024年10月17日 北京 天气 模拟} ... [工具 search_web] 返回: 关于2024年10月17日 北京 天气 模拟的模拟搜索结果相关文章1相关文章2。 --- ReAct循环 第 4 轮 --- LLM原始输出: Thought: 我获得了关于那天北京天气的模拟搜索结果。结合我之前不精确的日期估算2024年10月17日我可以给出一个综合回答。对于日期计算的不精确我应该向用户说明。 Final Answer: 根据计算从2024年5月20日往后150天大致是2024年10月17日左右这是一个估算精确计算需要专门的日期计算工具。关于那天的天气根据模拟搜索北京在2024年10月17日左右的天气信息通常显示为秋季气候可能凉爽干燥但具体预报需要临近时查询权威气象网站。从这个轨迹我们能学到什么ReAct的威力Agent确实在“思考”。它先拆解问题识别出需要计算和搜索两个子任务。它尝试使用calculator但发现不适合日期计算。然后它尝试搜索但我们的模拟搜索工具没有返回具体日期。最终它基于现有信息估算的日期和模拟的天气结果组合出了一个诚实且有用的答案并说明了局限性。工具能力的边界我们的工具集是有限的。Agent在遇到“日期计算”这个它不具备精确能力的任务时表现出了“自知之明”没有强行给出错误答案而是说明了工具的局限。这提示我们设计工具时清晰的职责描述和准确的错误处理反馈至关重要。可观测性的价值如果没有这个思维轨迹我们只会看到一个最终答案。但通过轨迹我们清晰地看到了Agent决策的每一步、遇到的困难以及它如何调整策略。这对于调试、优化提示词、补充工具能力具有不可估量的价值。持久化的意义在后续问题中新加载的Agent因为拥有了之前的记忆知道之前估算的日期是10月17日能够理解“刚才计算出来的那天”指代的是什么从而正确回答了后续问题。5. 从“玩具”到“工具”扩展思路与生产级考量我们完成了一个可运行的最简化Agent但它距离生产级应用还有距离。以下是几个关键的扩展方向和深度思考也是你在面试或实际项目中可能会被问到的点。5.1 工具系统的增强从静态注册到动态发现目前的工具是硬编码在TOOLS字典里的。一个更高级的Agent应该能动态发现和调用工具。工具注册机制可以设计一个装饰器让开发者能方便地将任何函数注册为工具。例如agent_tool(nameget_weather, description获取指定城市的天气) def fetch_weather(city: str, date: str None) - str: # 调用真实天气API ...工具语义检索当工具数量成百上千时LLM可能记不住所有工具名。可以引入向量数据库根据用户问题的语义动态检索最相关的几个工具供LLM选择。工具组合与工作流让Agent能够规划并顺序或并行执行多个工具形成工作流Workflow。这需要增强其规划Planning能力。5.2 提示工程优化提升可靠性与效率我们使用了最简单的提示词模板。要提升Agent的可靠性提示词需要精心设计少样本示例Few-shot在提示词中提供1-3个完整的ReAct循环示例能极大地引导LLM遵循正确的格式和推理路径。结构化输出要求除了我们使用的关键词Thought/Action可以要求LLM以更严格的JSON格式输出便于解析。许多最新的LLM如GPT-4原生支持工具调用Function Calling的JSON格式可以无缝集成。反思Reflection提示在每一轮观察之后可以加入一个“反思”步骤让LLM评估上一步行动是否有效是否偏离目标从而进行自我纠正。5.3 可观测性与监控的深化我们的AgentObserver只是一个开始。生产系统需要结构化日志将事件记录到如JSON Lines文件或日志系统ELK stack便于后续分析和告警。关键指标监控监控每个会话的循环轮次、工具调用耗时、Token消耗、最终答案满意度可通过简单规则或后续模型评估。设置阈值告警例如单次会话循环超过10轮可能意味着Agent“卡住”了。可视化界面开发一个简单的Web界面实时展示活跃Agent的思维轨迹这对于演示和调试非常直观。5.4 记忆与状态管理的进阶向量记忆Vector Memory对于长期记忆和知识库检索需要集成向量数据库如Chroma, Pinecone。将对话片段或外部知识编码成向量存储在需要时进行语义检索注入到上下文中。记忆摘要Memory Summarization当对话很长时定期用LLM对之前的对话历史进行摘要用摘要替代原始长文本既能保留关键信息又能节省上下文Token。状态恢复的健壮性我们的pickle方式很脆弱。生产环境应将状态存储为数据库中的记录包含会话ID、序列化的记忆、观察数据、当前任务状态等并处理好并发访问和状态冲突。5.5 处理复杂性与稳定性超时与看门狗Watchdog除了最大循环次数还应设置总耗时超时。可以设计一个看门狗线程监控主循环在超时或异常时强制终止并清理资源。优雅降级当核心工具如搜索失败时Agent应能感知并调整计划而不是崩溃或陷入死循环。可以在提示词中教导LLM处理“工具暂时不可用”的情况。验证与护栏Guardrails在Agent输出最终答案前或执行某些敏感工具如发送邮件、操作数据库前可以加入一层验证。例如用一个简单的规则引擎或另一个轻量级LLM调用来检查输出是否安全、是否符合规范。构建一个健壮的AI Agent系统是一个在“智能”与“可控”、“灵活”与“稳定”之间不断权衡的过程。我们这个最简化的实现为你揭示了最核心的运作机制。基于这个清晰的内核你可以根据具体的业务场景像搭积木一样逐步添加上面提到的各种高级功能最终打造出真正强大、实用的智能体。 从零构建AI Agent：ReAct循环、工具调用与可观测性实战指南
1. 从概念到代码一个最简化AI Agent的构建动机最近和不少同行交流发现大家一提到AI Agent脑海里浮现的要么是AutoGPT、LangChain这类庞大框架要么是各种“智能体平台”的复杂概念。框架虽好但有时也像黑箱把核心逻辑包裹得严严实实让人知其然不知其所以然。很多朋友想入门却被“工具调用”、“记忆”、“规划”这些术语劝退感觉无从下手。其实一个能自主思考、使用工具、并记住上下文的AI Agent其最核心的骨架远比想象中简单。今天我就想抛开那些重型框架带你从零开始亲手搭建一个最简化、但五脏俱全的AI Agent实现。这个实现将聚焦四个最核心的模块ReAct思考循环、工具调用、可观测性和持久化。通过这个“麻雀虽小五脏俱全”的项目你不仅能透彻理解Agent的工作机制更能获得一个可以随意扩展、用于实验和学习的坚实基础。无论你是想面试时侃侃而谈React Agent的原理还是想为自己的项目快速集成一个智能助手这篇手把手的指南都将为你提供清晰的路径和可直接运行的代码。2. 核心架构拆解理解“最简化”Agent的四大支柱在开始写代码之前我们必须先统一认知什么是一个“最简化”但“可用”的AI Agent它不应该依赖任何特定的庞大框架其核心能力应该由几个清晰、解耦的模块协同完成。基于这个思路我将其核心抽象为以下四个支柱它们共同构成了Agent的“大脑”与“身体”。2.1 ReAct循环Agent的“思考”引擎ReActReasoning Acting是让Agent变得“智能”的关键范式。它不是一个复杂的算法而是一个清晰的思维模式。你可以把它想象成一个优秀工程师的Debug过程观察Observe先看当前的状态和问题比如用户提问、工具执行结果、系统状态。思考Reason基于观察在大脑LLM里分析一下“我现在知道了什么我还需要什么下一步应该做什么”行动Act根据思考的结论执行一个具体的动作比如调用一个计算器工具或者查询数据库。循环将行动的结果作为新的“观察”回到第一步继续思考直到问题被解决或达到终止条件。这个循环的精髓在于它将“想”和“做”分开了。LLM只负责“想”生成包含思考过程和行动指令的文本而执行系统负责“做”解析指令并调用工具。这样既发挥了LLM的推理优势又通过工具弥补了其事实性和实时性的不足。在我们的最简化实现中ReAct循环就是那个驱动一切的主控程序。2.2 工具调用Agent的“手脚”与外部世界交互LLM本身是“纸上谈兵”的专家但它无法直接操作计算器、搜索网络、读写文件。工具调用Tool Calling就是为LLM装上“手脚”。一个工具本质上是一个函数它有明确的名称、描述、参数格式。当LLM在“思考”阶段决定要使用工具时它会按照预定格式如JSON输出一个工具调用请求。我们的系统会解析这个请求找到对应的工具函数并执行然后将执行结果返回给LLM供其下一轮思考使用。例如我们可以定义一个calculator工具描述为“执行数学计算”。当用户问“123乘以456等于多少”时LLM在ReAct循环中可能会输出“我需要计算123*456。我将调用计算器工具。” 然后生成一个结构化的调用请求{“action”: “calculator”, “action_input”: {“expression”: “123*456”}}。系统执行计算返回结果“56088”LLM再基于这个结果组织最终答案。2.3 可观测性Agent的“透明手术台”当你运行一个Agent尤其是它开始自主循环时最让人头疼的就是“它现在到底在想什么它做了什么决定为什么卡住了” 这就是可观测性Observability要解决的问题。它不同于简单的打印日志而是需要结构化的、分层次的洞察能力。一个具备良好可观测性的Agent系统应该能让我们看到思维链Chain of ThoughtLLM每一轮的完整推理文本包括其内心的“独白”和最终的决定。行动历史Action History它调用了哪些工具传入什么参数返回了什么结果。状态快照State Snapshot当前循环的轮次、使用的提示词模板、会话的元数据等。性能指标Metrics单轮耗时、Token消耗、工具调用成功率等。在我们的实现中我们会设计一个轻量级的“事件总线”或“回调系统”让这些内部状态能够以结构化的方式暴露出来方便我们实时监控、调试甚至在出现异常时进行干预。这是开发稳定、可靠Agent的必备基础设施也是面试中常被问到的“如何监控你的Agent”这一问题的实践答案。2.4 持久化Agent的“记忆”与状态留存一个没有记忆的Agent每次对话都是全新的开始这显然不符合“智能助手”的预期。持久化Persistence就是为了让Agent拥有“记忆”。这里的记忆分为几个层面会话记忆Conversation Memory记住当前对话的历史以便理解上下文。这是最基本的需求。长期记忆Long-term Memory跨越会话记住一些关键信息比如用户的偏好、之前达成的结论等。这通常需要向量数据库等外部存储。状态持久化State Persistence将Agent的完整运行状态包括当前循环索引、工具调用历史、内部变量等保存下来。这在Agent需要长时间运行或可能被中断时至关重要可以实现“暂停”和“继续”。在我们的最简化版本中我们会先实现最核心的会话记忆和状态快照保存/加载功能。这能让我们的Agent在单次运行中保持连贯性并且允许我们将一个复杂的任务状态保存到文件或数据库下次接着运行。这是构建真正可用Agent服务的基础。3. 手把手实现从零搭建最简化AI Agent理论讲完了现在进入最激动人心的实操环节。我们将使用Python不依赖LangChain等重型框架仅用openai库或兼容OpenAI API的库和一些基础数据结构构建出这个四合一的核心Agent。3.1 第一步定义工具系统——Agent的能力集工具系统是Agent的基石。我们先定义一个工具的基类然后实现几个常用工具。import json import requests from datetime import datetime from typing import Any, Dict, Callable class Tool: 工具基类定义工具的接口 def __init__(self, name: str, description: str, func: Callable): self.name name self.description description self.func func def run(self, **kwargs) - str: 执行工具返回字符串格式的结果 try: result self.func(**kwargs) return str(result) except Exception as e: return fError executing tool {self.name}: {str(e)} def to_dict(self) - Dict[str, Any]: 返回工具的元信息用于构造提示词 return { name: self.name, description: self.description, # 这里可以进一步细化参数schema为了简化先省略 } # 实现几个具体的工具 def calculator(expression: str) - float: 执行安全的数学表达式计算使用eval需谨慎此处仅作演示 # 警告在生产环境中应对expression做严格的安全检查和过滤 allowed_chars set(0123456789-*/(). ) if not all(c in allowed_chars for c in expression): raise ValueError(Expression contains invalid characters) return eval(expression) def get_current_time(*args, **kwargs) - str: 获取当前日期和时间 return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def search_web(query: str) - str: 模拟网络搜索实际可接入SerperAPI、Google Search等 # 此处为模拟真实情况需要调用搜索API print(f[模拟搜索] 搜索关键词: {query}) # 模拟返回一些结果 return f关于{query}的模拟搜索结果相关文章1相关文章2。 # 创建工具集 TOOLS { calculator: Tool(calculator, 执行数学计算例如3 5 * 2。, calculator), get_current_time: Tool(get_current_time, 获取当前的日期和时间。, get_current_time), search_web: Tool(search_web, 在互联网上搜索信息。需要提供搜索查询词。, search_web), }为什么这样设计统一的接口每个工具都有run方法Agent核心无需关心工具内部实现。自描述性to_dict方法提供了工具的元信息可以动态地插入到给LLM的提示词中告诉LLM现在有哪些工具可用。安全性考虑在calculator工具中我们简单演示了输入检查。在实际生产环境中直接使用eval是极度危险的必须替换为安全的表达式解析库如ast.literal_eval配合自定义解析器或沙箱环境。3.2 第二步设计ReAct提示词与解析逻辑——Agent的“思维模板”接下来我们需要设计一个提示词Prompt引导LLM按照ReAct的格式进行输出。同时我们需要一个解析器从LLM的回复中提取出“思考”和“行动”。def create_react_prompt(question: str, history: str, available_tools: list) - str: 构造ReAct格式的提示词 tools_desc \n.join([f- {t[name]}: {t[description]} for t in available_tools]) prompt f你是一个善于思考并可以使用工具的AI助手。你的任务是通过思考Thought、行动Action、观察Observation的循环来回答问题。 你可以使用的工具如下 {tools_desc} 对话历史 {history} 当前问题{question} 请严格按照以下格式回应 Thought: 你需要在这里描述你对问题的思考过程分析当前情况并决定下一步做什么。 Action: 你需要在这里指定要调用的工具名称。必须是以下之一{[t[name] for t in available_tools]}。 Action Input: 你需要在这里以JSON格式提供调用工具所需的输入参数。 或者如果你认为问题已经解决不需要再使用工具则使用以下格式 Thought: 我通过之前的步骤已经得到了答案。 Final Answer: 你的最终答案放在这里。 现在开始 return prompt def parse_llm_response(response: str) - Dict[str, Any]: 解析LLM的回复提取Thought, Action, Action Input 或 Final Answer lines response.strip().split(\n) result {thought: , action: None, action_input: None, final_answer: None} current_key None for line in lines: line line.strip() if line.startswith(Thought:): current_key thought result[current_key] line.replace(Thought:, , 1).strip() elif line.startswith(Action:): current_key action result[current_key] line.replace(Action:, , 1).strip() elif line.startswith(Action Input:): current_key action_input input_str line.replace(Action Input:, , 1).strip() # 尝试解析JSON try: result[current_key] json.loads(input_str) except json.JSONDecodeError: # 如果解析失败可能LLM没有输出严格JSON这里做简单处理 result[current_key] input_str elif line.startswith(Final Answer:): current_key final_answer result[current_key] line.replace(Final Answer:, , 1).strip() elif current_key and line: # 处理多行内容 result[current_key] \n line return result设计要点与避坑指南格式强制提示词中必须明确、严格地规定输出格式。LLM有时会“放飞自我”清晰的格式能极大提高解析成功率。工具描述清晰工具的描述description至关重要它是LLM决定是否及如何使用该工具的主要依据。描述应简洁、准确并包含参数示例。解析的鲁棒性parse_llm_response函数需要有一定的容错能力。LLM的输出可能有多余的空行、格式微调。我们的解析逻辑先按行分割再根据关键词提取并对JSON解析做了异常处理这是一个在实践中很有效的简单策略。3.3 第三步实现可观测性模块——给Agent装上“仪表盘”可观测性模块负责收集和展示Agent运行过程中的所有关键信息。我们实现一个简单的事件记录器。class AgentObserver: Agent可观测性模块记录运行全过程 def __init__(self): self.episodes [] # 记录每一轮Thought-Action-Observation的信息 self.current_episode {} def on_thought_start(self, question, prompt): 开始新一轮思考时调用 self.current_episode { question: question, prompt_snippet: prompt[:200] ... if len(prompt) 200 else prompt, # 记录部分提示词 thought: , action: , action_input: , observation: , final_answer: , timestamp: datetime.now().isoformat(), token_usage: {} # 可在此处记录token消耗 } def on_thought_generated(self, thought): 记录思考内容 self.current_episode[thought] thought def on_action_decided(self, action, action_input): 记录决策的行动 self.current_episode[action] action self.current_episode[action_input] action_input def on_observation_received(self, observation): 记录工具执行后的观察 self.current_episode[observation] observation def on_final_answer(self, answer): 记录最终答案 self.current_episode[final_answer] answer # 本轮结束存入历史 self.episodes.append(self.current_episode.copy()) def get_latest_episode(self): 获取最新一轮的详细信息 return self.current_episode def get_summary(self): 获取运行摘要 summary { total_episodes: len(self.episodes), tools_used: {}, last_question: self.episodes[-1][question] if self.episodes else None, last_answer: self.episodes[-1][final_answer] if self.episodes else None, } for ep in self.episodes: tool ep.get(action) if tool: summary[tools_used][tool] summary[tools_used].get(tool, 0) 1 return summary def print_trace(self): 以友好格式打印完整的思维轨迹用于调试 print(\n *60) print(AI Agent 思维轨迹追踪) print(*60) for i, ep in enumerate(self.episodes): print(f\n[第 {i1} 轮]) print(f 问题: {ep[question]}) print(f 思考: {ep[thought]}) if ep[action]: print(f 行动: {ep[action]}({ep[action_input]})) print(f 观察: {ep[observation]}) if ep[final_answer]: print(f 最终答案: {ep[final_answer]}) print(*60)这个观察器有什么用调试神器当Agent行为不符合预期时调用observer.print_trace()可以一目了然地看到它每一步的“想法”和“行动”快速定位是提示词问题、工具问题还是LLM理解问题。监控看板get_summary()方法可以提供关键指标如工具调用频率、循环轮次未来可以接入Grafana等监控系统。审计日志所有episodes记录了完整的历史可用于复盘分析或满足合规性要求。3.4 第四步集成持久化模块——赋予Agent“记忆”现在我们让Agent能够记住对话历史并能保存/加载整个会话状态。import pickle from typing import List class ConversationMemory: 简单的对话记忆管理上下文 def __init__(self, max_turns10): self.messages: List[Dict] [] # 存储{role: user/assistant, content: ...} self.max_turns max_turns # 控制上下文长度防止token超限 def add_user_message(self, content: str): self.messages.append({role: user, content: content}) self._trim_memory() def add_assistant_message(self, content: str): self.messages.append({role: assistant, content: content}) self._trim_memory() def _trim_memory(self): 修剪记忆保留最近的对话轮次 if len(self.messages) self.max_turns * 2: # 每轮包含user和assistant两条 # 保留系统提示如果有和最近的对话 self.messages self.messages[-self.max_turns*2:] def get_history_for_prompt(self) - str: 将记忆格式化为提示词中可用的历史字符串 history_str for msg in self.messages[-self.max_turns*2:]: # 确保不超过最大轮次 role Human if msg[role] user else Assistant history_str f{role}: {msg[content]}\n return history_str.strip() def clear(self): self.messages.clear() class PersistentAgentState: Agent持久化状态可保存到文件或数据库 def __init__(self, memory: ConversationMemory, observer: AgentObserver, other_stateNone): self.memory memory self.observer_data observer.episodes # 保存观察器的历史记录 self.other_state other_state or {} # 可扩展其他状态 self.saved_at datetime.now().isoformat() def save(self, filepath: str): 将状态保存到文件 with open(filepath, wb) as f: pickle.dump(self, f) print(fAgent状态已保存至: {filepath}) staticmethod def load(filepath: str) - PersistentAgentState: 从文件加载状态 with open(filepath, rb) as f: state pickle.load(f) print(fAgent状态已从 {filepath} 加载保存时间: {state.saved_at}) return state记忆与持久化的关键考量上下文长度管理ConversationMemory中的max_turns和_trim_memory方法至关重要。LLM有上下文窗口限制无限制地增长历史会导致后续请求失败或成本激增。常见的策略是保留最近N轮对话或使用更复杂的“摘要记忆”方式将久远的历史压缩成一段摘要。状态序列化我们使用pickle进行简单演示它可以将整个Python对象保存到文件。但在生产环境中pickle存在安全风险可能执行恶意代码且与语言/框架绑定。更安全的做法是将状态转化为安全的字典结构如JSON或使用ORM框架直接存入数据库。状态恢复加载状态后需要将memory和observer_data重新注入到运行中的Agent实例使其能“无缝”继续工作。这为实现长时间运行的任务如自动化工作流提供了可能。3.5 第五步组装核心ReAct循环——让Agent“动”起来最后我们将所有模块组装起来实现主循环逻辑。import openai # 或其他兼容OpenAI API的客户端 class SimpleReActAgent: 最简化的ReAct Agent核心实现 def __init__(self, llm_client, modelgpt-3.5-turbo, max_iterations5): self.llm_client llm_client self.model model self.max_iterations max_iterations # 防止无限循环 self.tools TOOLS self.memory ConversationMemory(max_turns5) self.observer AgentObserver() self.available_tools_info [tool.to_dict() for tool in self.tools.values()] def run(self, query: str) - str: 执行ReAct循环处理用户查询 print(f\n[Agent] 开始处理查询: {query}) self.memory.add_user_message(query) for i in range(self.max_iterations): print(f\n--- ReAct循环 第 {i1} 轮 ---) # 1. 准备提示词包含历史 history self.memory.get_history_for_prompt() prompt create_react_prompt(query, history, self.available_tools_info) # 记录可观测性事件 self.observer.on_thought_start(query, prompt) # 2. 调用LLM进行思考 try: response self.llm_client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], temperature0.1, # 低温度保证输出格式稳定 ) llm_output response.choices[0].message.content print(fLLM原始输出:\n{llm_output}) except Exception as e: return f调用LLM API时出错: {str(e)} # 3. 解析LLM输出 parsed parse_llm_response(llm_output) self.observer.on_thought_generated(parsed.get(thought, )) # 4. 判断是否结束 if parsed.get(final_answer): final_answer parsed[final_answer] print(f[Agent] 最终答案: {final_answer}) self.memory.add_assistant_message(final_answer) self.observer.on_final_answer(final_answer) # 打印完整的思维轨迹 self.observer.print_trace() return final_answer # 5. 执行工具调用 action parsed.get(action) action_input parsed.get(action_input) if not action or action not in self.tools: error_msg fLLM指定了无效或未知的行动: {action} print(f[错误] {error_msg}) # 可以将错误信息作为Observation反馈给LLM让其重试 observation error_msg else: print(f[Agent] 执行工具: {action}, 输入: {action_input}) self.observer.on_action_decided(action, action_input) # 执行工具 tool self.tools[action] if isinstance(action_input, dict): observation tool.run(**action_input) else: # 如果action_input不是dict尝试作为字符串参数传递 observation tool.run(action_input) print(f[工具 {action}] 返回: {observation}) self.observer.on_observation_received(observation) # 6. 将观察结果加入记忆准备下一轮循环 # 注意这里我们将“思考-行动-观察”作为一个整体回合加入记忆简化处理 # 更精细的做法可以分别记录 assistant_message fThought: {parsed.get(thought)}\nAction: {action}\nObservation: {observation} self.memory.add_assistant_message(assistant_message) # 循环达到最大次数仍未得出最终答案 timeout_msg f经过{self.max_iterations}轮思考仍未得出结论。 print(f[Agent] {timeout_msg}) self.observer.on_final_answer(timeout_msg) self.observer.print_trace() return timeout_msg def save_state(self, filepath: str): 保存当前Agent状态记忆、观察历史 state PersistentAgentState(self.memory, self.observer) state.save(filepath) def load_state(self, filepath: str): 加载Agent状态 state PersistentAgentState.load(filepath) self.memory state.memory self.observer.episodes state.observer_data print(状态加载完成记忆和观察历史已恢复。)主循环的核心逻辑与避坑点循环终止条件必须有max_iterations限制这是防止Agent陷入死循环或产生过高成本的安全阀。同时依赖LLM自己输出Final Answer来正常结束。错误处理在工具调用和LLM API调用环节必须有充分的try-except。特别是当LLM输出格式不符合预期或指定了不存在的工具时要有降级处理如将错误信息作为observation反馈回去让LLM自我纠正。温度Temperature设置在ReAct这类需要严格遵循输出格式的场景建议设置较低的temperature如0.1或0以增加输出的确定性和格式稳定性。记忆的粒度示例中将一整轮“Thought-Action-Observation”作为一条助手消息存入记忆。在实际复杂任务中你可能需要更精细的控制比如只将关键的observation和final_answer存入长期对话记忆而将中间思考过程仅用于可观测性记录。4. 运行示例与深度调试看Agent如何“思考”让我们用一个具体的例子来完整跑通这个Agent并利用可观测性模块深入其“内心世界”。# 假设你已经配置好了OpenAI API客户端 # client openai.OpenAI(api_keyyour-api-key) # 1. 初始化Agent agent SimpleReActAgent(llm_clientclient, modelgpt-3.5-turbo) # 2. 运行一个需要多步推理和工具调用的查询 question “今天是2024年5月20日请问150天后的星期几另外那天北京的天气怎么样模拟” answer agent.run(question) print(f\n最终回复给用户: {answer}) # 3. 保存当前会话状态模拟长时间任务中断 agent.save_state(agent_session_20240520.pkl) # 4. 模拟在另一个进程或时间点加载状态并继续 print(\n--- 模拟重新加载Agent状态并继续对话 ---) new_agent SimpleReActAgent(llm_clientclient, modelgpt-3.5-turbo) new_agent.load_state(agent_session_20240520.pkl) follow_up_question “把刚才计算出来的那天加上‘国际日’后缀重新告诉我日期。” follow_up_answer new_agent.run(follow_up_question) print(f\n后续问题回复: {follow_up_answer})预期输出与观察当你运行上述代码通过observer.print_trace()或直接查看控制台打印你会看到类似以下的思维轨迹[Agent] 开始处理查询: 今天是2024年5月20日请问150天后的星期几另外那天北京的天气怎么样模拟 --- ReAct循环 第 1 轮 --- LLM原始输出: Thought: 用户问了两个问题。第一个是计算150天后的日期和星期几这需要计算。第二个是问那天的天气这需要搜索。我应该先解决第一个问题因为它是一个确定性的计算然后再搜索天气。 Action: calculator Action Input: {expression: 150} ... [工具 calculator] 返回: 150 --- ReAct循环 第 2 轮 --- LLM原始输出: Thought: 我得到了150。但我需要从2024年5月20日开始计算150天后的具体日期。这涉及到日期运算简单的计算器工具可能不够。我需要一个能处理日期的工具或者用另一种方式。我意识到我可能没有日期计算工具。我可以先估算或者告诉用户我无法精确计算日期。但也许我可以先搜索“2024年5月20日 150天后”看看有没有在线计算器或信息。我先搜索一下。 Action: search_web Action Input: {query: 2024年5月20日 150天后 日期 计算} ... [工具 search_web] 返回: 关于2024年5月20日 150天后 日期 计算的模拟搜索结果相关文章1相关文章2。 --- ReAct循环 第 3 轮 --- LLM原始输出: Thought: 搜索返回了模拟结果但没有给出具体日期。我需要换一种思路。我可以手动计算5月有31天从5月20日到5月31日是11天。6月30天7月31天8月31天。11303131103天。150-10347天。9月有30天所以47天后是10月17日等等我需要更精确。这太容易出错了。我最好承认这个限制并专注于我能回答的部分——天气查询。我可以先回答天气部分并说明日期计算需要更专业的工具。 Action: search_web Action Input: {query: 2024年10月17日 北京 天气 模拟} ... [工具 search_web] 返回: 关于2024年10月17日 北京 天气 模拟的模拟搜索结果相关文章1相关文章2。 --- ReAct循环 第 4 轮 --- LLM原始输出: Thought: 我获得了关于那天北京天气的模拟搜索结果。结合我之前不精确的日期估算2024年10月17日我可以给出一个综合回答。对于日期计算的不精确我应该向用户说明。 Final Answer: 根据计算从2024年5月20日往后150天大致是2024年10月17日左右这是一个估算精确计算需要专门的日期计算工具。关于那天的天气根据模拟搜索北京在2024年10月17日左右的天气信息通常显示为秋季气候可能凉爽干燥但具体预报需要临近时查询权威气象网站。从这个轨迹我们能学到什么ReAct的威力Agent确实在“思考”。它先拆解问题识别出需要计算和搜索两个子任务。它尝试使用calculator但发现不适合日期计算。然后它尝试搜索但我们的模拟搜索工具没有返回具体日期。最终它基于现有信息估算的日期和模拟的天气结果组合出了一个诚实且有用的答案并说明了局限性。工具能力的边界我们的工具集是有限的。Agent在遇到“日期计算”这个它不具备精确能力的任务时表现出了“自知之明”没有强行给出错误答案而是说明了工具的局限。这提示我们设计工具时清晰的职责描述和准确的错误处理反馈至关重要。可观测性的价值如果没有这个思维轨迹我们只会看到一个最终答案。但通过轨迹我们清晰地看到了Agent决策的每一步、遇到的困难以及它如何调整策略。这对于调试、优化提示词、补充工具能力具有不可估量的价值。持久化的意义在后续问题中新加载的Agent因为拥有了之前的记忆知道之前估算的日期是10月17日能够理解“刚才计算出来的那天”指代的是什么从而正确回答了后续问题。5. 从“玩具”到“工具”扩展思路与生产级考量我们完成了一个可运行的最简化Agent但它距离生产级应用还有距离。以下是几个关键的扩展方向和深度思考也是你在面试或实际项目中可能会被问到的点。5.1 工具系统的增强从静态注册到动态发现目前的工具是硬编码在TOOLS字典里的。一个更高级的Agent应该能动态发现和调用工具。工具注册机制可以设计一个装饰器让开发者能方便地将任何函数注册为工具。例如agent_tool(nameget_weather, description获取指定城市的天气) def fetch_weather(city: str, date: str None) - str: # 调用真实天气API ...工具语义检索当工具数量成百上千时LLM可能记不住所有工具名。可以引入向量数据库根据用户问题的语义动态检索最相关的几个工具供LLM选择。工具组合与工作流让Agent能够规划并顺序或并行执行多个工具形成工作流Workflow。这需要增强其规划Planning能力。5.2 提示工程优化提升可靠性与效率我们使用了最简单的提示词模板。要提升Agent的可靠性提示词需要精心设计少样本示例Few-shot在提示词中提供1-3个完整的ReAct循环示例能极大地引导LLM遵循正确的格式和推理路径。结构化输出要求除了我们使用的关键词Thought/Action可以要求LLM以更严格的JSON格式输出便于解析。许多最新的LLM如GPT-4原生支持工具调用Function Calling的JSON格式可以无缝集成。反思Reflection提示在每一轮观察之后可以加入一个“反思”步骤让LLM评估上一步行动是否有效是否偏离目标从而进行自我纠正。5.3 可观测性与监控的深化我们的AgentObserver只是一个开始。生产系统需要结构化日志将事件记录到如JSON Lines文件或日志系统ELK stack便于后续分析和告警。关键指标监控监控每个会话的循环轮次、工具调用耗时、Token消耗、最终答案满意度可通过简单规则或后续模型评估。设置阈值告警例如单次会话循环超过10轮可能意味着Agent“卡住”了。可视化界面开发一个简单的Web界面实时展示活跃Agent的思维轨迹这对于演示和调试非常直观。5.4 记忆与状态管理的进阶向量记忆Vector Memory对于长期记忆和知识库检索需要集成向量数据库如Chroma, Pinecone。将对话片段或外部知识编码成向量存储在需要时进行语义检索注入到上下文中。记忆摘要Memory Summarization当对话很长时定期用LLM对之前的对话历史进行摘要用摘要替代原始长文本既能保留关键信息又能节省上下文Token。状态恢复的健壮性我们的pickle方式很脆弱。生产环境应将状态存储为数据库中的记录包含会话ID、序列化的记忆、观察数据、当前任务状态等并处理好并发访问和状态冲突。5.5 处理复杂性与稳定性超时与看门狗Watchdog除了最大循环次数还应设置总耗时超时。可以设计一个看门狗线程监控主循环在超时或异常时强制终止并清理资源。优雅降级当核心工具如搜索失败时Agent应能感知并调整计划而不是崩溃或陷入死循环。可以在提示词中教导LLM处理“工具暂时不可用”的情况。验证与护栏Guardrails在Agent输出最终答案前或执行某些敏感工具如发送邮件、操作数据库前可以加入一层验证。例如用一个简单的规则引擎或另一个轻量级LLM调用来检查输出是否安全、是否符合规范。构建一个健壮的AI Agent系统是一个在“智能”与“可控”、“灵活”与“稳定”之间不断权衡的过程。我们这个最简化的实现为你揭示了最核心的运作机制。基于这个清晰的内核你可以根据具体的业务场景像搭积木一样逐步添加上面提到的各种高级功能最终打造出真正强大、实用的智能体。