从ReAct到AI Agent：工程化实现与架构设计实战指南 1. 项目概述从ReAct到Agent的工程化之路最近在AI应用开发圈里ReAct框架和AI Agent成了高频词。很多朋友看了论文或者概念介绍觉得思路很清晰——让大模型LLM学会“思考-行动-观察”的循环再结合函数调用Function Call去执行具体操作一个能自主完成复杂任务的智能体Agent似乎就呼之欲出了。但真到了动手写代码的时候往往就卡壳了循环怎么设计工具怎么管理状态如何维护错误怎么处理这些问题论文里可不会细说。我花了不少时间基于LLM Function Call的模式从零搭建并迭代了好几个Agent项目。今天就想抛开那些宏大的概念聚焦在代码实现上聊聊如何把一个ReAct的理论框架落地成一个稳定、可扩展、真正能用的AI Agent。这不仅仅是调用几个API更是关于系统设计、状态管理和工程实践的思考。2. 核心架构设计构建Agent的“大脑”与“手脚”一个基于ReAct的AI Agent其核心架构可以类比为一个具备反思和执行能力的智能体。它的“大脑”是LLM负责规划和决策“手脚”则是通过Function Call暴露的各种工具Tools。而ReAct框架就是连接大脑和手脚、并使其协同工作的“神经系统”。2.1 ReAct循环的工程化拆解在论文中ReActReasoning Acting通常被描述为Thought - Action - Observation的循环。但在工程实现上我们需要将其细化、健壮化。一个更完整的工程化循环通常包括以下步骤任务解析与初始化Agent接收一个用户查询QueryLLM首先需要理解任务的最终目标并将其分解为可能的子步骤。这一步的输出是一个初始的“思考”Thought。思考ThoughtLLM基于当前任务目标、已有的历史观察Observation和可用的工具列表分析现状决定下一步要做什么。关键输出是a) 是否需要调用工具b) 如果调用调用哪个工具以及传入什么参数。行动Action如果思考决定要行动则根据上一步的决策格式化一个标准的工具调用请求。这通常是一个结构化的JSON包含tool_name和arguments。工具执行与观察Observation系统根据Action请求找到对应的函数并执行将执行结果成功或失败以及返回的数据整理成文本形式的“观察”反馈给LLM。这里极易出错比如工具执行异常、返回数据格式不符合LLM预期等。循环判断与终止LLM结合新的观察再次思考。判断任务是否已经完成生成最终答案或者需要继续下一步行动。需要设定明确的终止条件防止无限循环。注意在代码中这个循环需要一个“执行引擎”来驱动。这个引擎负责维护循环状态、调用LLM、分发工具调用、收集结果并判断循环是否应该继续。2.2 工具Function的管理与注册机制Function Call是Agent的“手脚”。如何优雅地管理这些手脚是架构设计的重点。我们不能把一堆函数散乱地扔在代码里。一个常见的工具管理模块应具备以下功能工具注册表一个中心化的地方比如一个Python字典或一个专门的类来注册所有可用的工具。每个工具需要提供唯一名称、功能描述、参数JSON Schema定义参数名称、类型、是否必需、描述。工具描述生成LLM需要知道它能用什么工具。因此我们需要将注册的工具列表按照LLM提供商如OpenAI, Anthropic要求的格式生成一个“工具描述列表”。这个列表会在每次调用LLM时附上相当于告诉LLM“这是你能用的所有工具说明书”。工具调用路由与执行当LLM返回一个工具调用请求时执行引擎需要根据tool_name从注册表中找到对应的Python函数并将arguments解析后传入执行。工具结果的标准化工具执行后可能返回任何Python对象字典、列表、字符串、甚至None。我们需要一个标准化的处理器将这些结果转换为LLM容易理解的文本格式Observation。对于复杂对象可能需要序列化为JSON字符串或进行摘要。# 一个简化的工具注册与执行示例 class ToolRegistry: def __init__(self): self._tools {} # {‘tool_name‘: {‘func‘: callable, ‘schema‘: dict}} def register(self, name: str, func: callable, description: str, params_schema: dict): 注册一个工具 self._tools[name] { func: func, schema: { name: name, description: description, parameters: params_schema } } def get_tool_schemas_for_llm(self): 生成LLM所需的工具描述列表 return [tool_info[schema] for tool_info in self._tools.values()] def execute(self, tool_name: str, arguments: dict): 执行指定工具 if tool_name not in self._tools: raise ValueError(fTool {tool_name} not found.) tool_func self._tools[tool_name][func] try: # 这里可以加入参数验证、类型转换等 result tool_func(**arguments) return str(result) # 简单转换为字符串作为观察 except Exception as e: return fError executing tool {tool_name}: {str(e)}实操心得工具的描述description和参数Schema的描述至关重要。LLM完全依赖这些文本来决定是否以及如何调用工具。描述要清晰、具体说明工具的用途、输入输出。参数Schema要尽可能严格这能极大减少LLM传参格式错误的概率。3. 与LLM的交互Prompt工程与消息流设计LLM是Agent的决策核心。如何与它“对话”直接影响Agent的表现。3.1 系统提示词System Prompt的精心设计系统提示词定义了Agent的角色、能力和行为规范。一个好的系统提示词是成功的一半。一个有效的ReAct Agent系统提示词通常包含角色定义明确告诉LLM它现在是一个AI助手并且具备调用外部工具的能力。任务流程说明清晰地阐述ReAct循环的步骤思考-行动-观察并给出格式示例。强制要求LLM以特定格式如JSON输出便于代码解析。工具使用说明指示LLM在需要时从提供的工具列表中选择并严格按照参数Schema来构造调用。终止条件告诉LLM当任务完成时应输出一个特殊的标记如Final Answer:来结束循环。约束与规范例如禁止编造工具、一次只调用一个工具、思考要简洁等。示例片段你是一个强大的AI助手可以通过调用工具来获取信息或执行操作以完成用户任务。 请遵循以下步骤 1. 思考分析当前情况和目标决定下一步。 2. 行动如果需要调用工具请以严格的JSON格式输出仅包含action和action_input字段。action是工具名action_input是参数字典。 3. 观察你将收到工具执行的结果。 工具列表{{TOOL_SCHEMAS}} 如果你认为已经收集到足够信息可以给出最终答案请以“Final Answer:”开头输出。 现在开始。用户查询是{{USER_QUERY}} 历史记录{{CONVERSATION_HISTORY}}3.2 对话历史Message History的管理Agent需要记忆。它必须能看到之前的思考、行动和观察才能进行连贯的推理。我们需要维护一个对话消息列表。常见的消息角色包括system: 系统提示词通常在对话开始前注入一次。user: 用户的最新查询以及拼接的历史观察和新的问题。assistant: LLM之前的回复包含思考和行动指令。tool(或function): 工具执行后返回的观察结果。在OpenAI等API中这是一个特殊的角色。管理策略长度控制LLM有上下文长度限制。需要设计策略来裁剪或总结过长的历史尤其是那些包含大量工具返回数据的观察。一种简单策略是只保留最近N轮循环。格式保持确保历史消息的格式与LLM API期望的一致。例如将Action Observation对正确地格式化为user和tool消息。实操心得直接将庞大的工具返回数据如一整篇网页内容塞进上下文会快速消耗Token并可能干扰LLM的思考。更好的做法是让工具本身或一个后处理函数对返回数据进行摘要提取只将关键信息作为Observation。例如一个搜索工具返回10条结果可以只提取每条结果的标题和摘要前100字。4. 状态机与执行引擎Agent的“心脏”这是将以上所有部分粘合起来的核心代码。我们可以将其概念化为一个状态机State Machine。4.1 核心循环的实现class ReActAgent: def __init__(self, llm_client, tool_registry, system_prompt): self.llm llm_client self.tools tool_registry self.system_prompt system_prompt self.max_steps 10 # 防止无限循环 self.conversation_history [] def run(self, user_query: str) - str: # 初始化对话历史 messages [{role: system, content: self.system_prompt}] # 将初始查询加入历史 self.conversation_history.append({role: user, content: user_query}) messages.append({role: user, content: self._format_current_context(user_query)}) for step in range(self.max_steps): # 1. 调用LLM进行思考/决策 llm_response self.llm.chat_completion(messages) assistant_message llm_response.choices[0].message messages.append(assistant_message) self.conversation_history.append(assistant_message) # 2. 解析LLM响应 content assistant_message.content # 2.1 检查是否最终答案 if Final Answer: in content: final_answer content.split(Final Answer:)[-1].strip() return final_answer # 2.2 尝试解析工具调用 (这里需要健壮的解析逻辑) tool_call self._parse_tool_call(content) if not tool_call: # LLM可能输出了无法解析的内容视为一次“观察”即它说了些话但没调用工具 # 我们可以将其作为虚拟观察加入历史并继续循环 observation fThe assistant said: {content} messages.append({role: user, content: fObservation: {observation}}) continue # 3. 执行工具调用 tool_name tool_call[action] tool_args tool_call[action_input] observation self.tools.execute(tool_name, tool_args) # 4. 将观察结果格式化并加入对话历史供下一轮思考 observation_msg fObservation: {observation} messages.append({role: user, content: observation_msg}) self.conversation_history.append({role: user, content: observation_msg}) # 循环超过最大步数 return fAgent stopped after {self.max_steps} steps. Unable to complete task. def _format_current_context(self, query): # 一个简单的上下文格式化函数可以整合历史 # 更复杂的实现可能会对长历史进行摘要 context fCurrent user query: {query}\n\n if self.conversation_history: context Previous conversation:\n for msg in self.conversation_history[-5:]: # 只保留最近5轮 context f{msg[role]}: {msg[content]}\n return context def _parse_tool_call(self, text): # 这是一个关键且脆弱的部分 # 需要从LLM的输出中可靠地提取JSON。 # 方法1使用正则表达式查找JSON块。 # 方法2要求LLM输出纯JSON并在提示词中强调。 # 方法3推荐使用支持结构化输出的LLM API如OpenAI的JSON Mode或Anthropic的tool_use。 import json, re json_match re.search(r\{.*\}, text, re.DOTALL) if json_match: try: data json.loads(json_match.group()) # 验证必要字段 if action in data and action_input in data: return data except json.JSONDecodeError: pass return None4.2 错误处理与鲁棒性增强上面的基础循环非常脆弱。生产级的Agent必须考虑各种错误。LLM输出解析失败LLM可能不按格式输出。策略1) 优化提示词2) 在解析失败时向LLM发送错误信息并要求重试例如“你上次的响应格式不正确请严格按照JSON格式输出行动指令。”。工具执行错误工具可能抛出异常网络错误、参数错误等。策略捕获异常将清晰的错误信息作为Observation返回给LLM让它有机会调整策略。循环停滞LLM可能陷入死循环反复调用同一个工具或进行无意义的思考。策略1) 设置最大步数2) 在历史中检测重复模式并中断3) 引入“超时”或“人工干预”机制。上下文超长随着循环进行历史消息会越来越长。策略实现一个“历史摘要”功能定期将旧的对话压缩成一段摘要而不是保留所有原始文本。5. 实战构建一个简单的网页搜索Agent让我们用一个具体例子把上面的零件组装起来。我们将构建一个能回答实时性问题的Agent它可以使用一个“搜索网络”的工具。5.1 定义工具我们使用一个模拟的搜索函数实际项目中可接入SerpAPI、Google Search API等。import json from typing import List, Dict import requests # 假设我们有一个搜索API def search_web(query: str, num_results: int 3) - str: 在互联网上搜索给定查询词并返回结果的摘要。 Args: query (str): 要搜索的关键词或问题。 num_results (int): 返回的最大结果数量默认为3。 Returns: str: 搜索结果的摘要文本每条结果包含标题和摘要。 # 这里是模拟代码实际应调用搜索API # 例如 response requests.get(fhttps://api.search.com/?q{query}num{num_results}) # 下面是一个模拟返回 mock_results [ {title: 什么是ReAct框架, snippet: ReAct是一种将推理和行动结合的大模型交互范式...}, {title: AI Agent开发指南, snippet: 本文介绍了使用LLM和工具调用构建自主Agent的步骤...}, {title: 最新天气预报, snippet: 北京晴15-25摄氏度。} ] formatted_results [] for i, r in enumerate(mock_results[:num_results], 1): formatted_results.append(f{i}. {r[title]}: {r[snippet]}) return \n.join(formatted_results) # 注册工具 tool_registry ToolRegistry() tool_registry.register( namesearch_web, funcsearch_web, description当需要获取最新的、非模型训练数据内的信息时使用此工具例如当前事件、天气、股价、新闻等。, params_schema{ type: object, properties: { query: {type: string, description: 搜索查询词}, num_results: {type: integer, description: 返回结果数量默认3, default: 3} }, required: [query] } )5.2 配置LLM客户端与提示词假设我们使用OpenAI的ChatCompletion API其他厂商类似。import openai # 初始化客户端 client openai.OpenAI(api_keyyour-api-key) system_prompt 你是一个有帮助的AI助手可以调用搜索工具来回答用户问题。 请严格按照以下格式输出 思考你的推理过程 行动{action: tool_name, action_input: {arg1: value1}} 仅在需要调用工具时输出此行 最终答案你的最终回答 当你能直接回答或通过工具获得信息后回答时以此开头 可用工具 {tools_description} 现在开始。记住一次只调用一个工具。 用户问题{user_input} 5.3 运行Agentagent ReActAgent(llm_clientclient, tool_registrytool_registry, system_promptsystem_prompt) question 北京今天的天气怎么样 answer agent.run(question) print(answer) # 预期输出基于我们的模拟搜索工具 # 最终答案根据搜索结果北京今天的天气是晴气温在15到25摄氏度之间。执行过程推演LLM看到问题“北京今天的天气怎么样”思考后认为需要最新信息。它输出行动{action: search_web, action_input: {query: 北京今天天气}}。引擎调用search_web工具获得模拟的天气结果。引擎将结果“Observation: 1. 最新天气预报: 北京晴15-25摄氏度。”加入历史并再次调用LLM。LLM看到观察结果判断信息已足够输出“最终答案北京今天晴15-25摄氏度。”。6. 高级话题与优化方向实现基础循环只是起点。要打造强大的Agent还需要考虑以下方面。6.1 复杂工具与工作流组合工具一个工具可以封装更复杂的子流程。例如一个“订机票”工具内部可能需要先调用“搜索航班”再调用“获取价格”最后调用“创建订单”。并行与顺序执行有些任务可以并行调用多个工具如同时查询天气和交通。这需要扩展Action的格式和执行引擎使其支持一个步骤内发起多个工具调用。子Agent对于极其复杂的任务可以设计一个“主Agent”来规划和协调多个“子Agent”每个子Agent负责一个特定领域如数据分析、文案撰写、代码生成。这本质上是将ReAct循环层级化。6.2 记忆与知识管理短期记忆即当前的对话历史。管理策略上文已讨论。长期记忆让Agent记住跨对话的信息。这通常需要引入向量数据库。将对话中的关键信息如用户偏好、事实结论提取并存入向量库在后续对话中通过检索增强RAG的方式提供给LLM。反思与学习高级Agent可以对自己的行动历史进行“反思”总结成功和失败的模式并更新自己的策略或知识。这属于更前沿的研究范畴。6.3 评估与监控可观测性在Agent运行时详细记录每一步的思考、行动、观察和耗时。这对于调试和优化至关重要。评估指标如何评价一个Agent的好坏除了最终任务的完成度成功率还可以评估其步骤效率用了多少步、工具调用准确性、成本Token消耗等。人机协同Human-in-the-loop在关键步骤或Agent不确定时暂停并请求人类反馈。这对于高风险或高价值任务非常必要。7. 常见陷阱与调试技巧在实际开发中你肯定会遇到各种问题。以下是一些常见坑点和解决思路。问题1LLM不按格式输出导致解析失败。排查首先检查系统提示词是否清晰、强制地规定了输出格式。使用“必须”、“严格”等词。提供更清晰的示例。技巧使用LLM的“结构化输出”功能如OpenAI的response_format{ type: json_object }。这能极大提高输出稳定性。兜底在解析代码中如果第一次解析失败可以将错误信息连同原始上下文再次发送给LLM要求它纠正。例如“你上次的响应不是有效的JSON。请重试只输出JSON。”问题2Agent陷入无效循环反复调用同一个工具或进行无意义思考。排查观察历史。LLM是否收到了足够的观察信息工具返回的结果是否清晰任务目标是否在上下文中被遗忘技巧在提示词中强调“避免重复行动”。在引擎中实现简单的重复检测如果连续N步行动相同则中断循环并返回错误。在每轮思考时都重新强调一遍用户的原始问题防止LLM跑偏。问题3工具调用参数总是出错。排查检查工具的参数Schema描述是否足够精确LLM是否理解每个参数的类型字符串、数字、布尔值和含义技巧在Schema中使用enum来限制参数的可选值。为每个参数提供详细的description和examples。在工具执行前可以加入一层参数验证和类型转换的预处理。问题4上下文太长导致响应慢或LLM性能下降。排查是工具返回的数据太大还是历史轮次太多技巧对于工具返回的大数据如长文档、多行数据强制要求进行摘要。可以设计一个“摘要”工具或者让工具函数本身返回摘要版本。对于历史实现一个滑动窗口只保留最近K轮交互。问题5Agent在简单任务上表现良好但复杂任务完全失败。排查复杂任务可能需要多步规划和信息整合。你的Agent是否具备“规划”能力还是只是走一步看一步技巧在系统提示词中引导LLM在第一步先进行“任务分解”。例如“请先分析这个问题并列出需要哪些步骤来解决它。” 然后让Agent逐步执行这些步骤。这相当于将ReAct循环提升了一个层次。构建一个可靠的AI Agent是一个迭代过程。从最简单的单一工具循环开始逐步增加复杂性并辅以完善的日志和监控你才能清晰地看到问题所在并持续改进。代码实现是骨架而提示词、工具设计和错误处理策略则是赋予其灵魂和韧性的血肉。