从零手撸AI智能体：基于ReAct循环的自主决策与任务拆解实践

1. 项目概述：从“执行”到“思考”的跨越

最近在AI圈子里，“智能体”这个词的热度是越来越高。从各种AI应用平台到开发者社区，大家似乎都在讨论如何让大模型不止是“一问一答”，而是能像人一样，自主规划、执行任务。我手头这个项目，就是想抛开那些复杂的框架和平台，用最直接的Python代码，从零开始“手撸”一个具备基础“思考”能力的AI智能体原型。

所谓“思考”，在这里并不是指科幻电影里的强人工智能，而是指让程序具备一种自主决策和任务拆解的能力。比如，你给它一个模糊的指令“帮我分析一下最近的销售数据”，一个简单的聊天机器人可能只会回复“我无法处理文件”。但一个会“思考”的智能体，应该能理解这个指令背后的意图，然后自主规划出几个步骤：1. 请求用户上传数据文件；2. 读取并解析数据；3. 调用合适的分析工具（比如计算环比、找出Top产品）；4. 生成一份结构化的报告。这个过程，就是智能体区别于普通聊天程序的核心。

这个项目的核心价值在于理解其运作机理。市面上已经有Dify、Coze等优秀的智能体平台，它们封装得很好，拖拖拽拽就能搭建。但如果你想知道智能体到底是怎么“想”的，它的“大脑”里发生了什么，那么亲手用代码实现一遍，是无可替代的学习路径。通过这个项目，你将彻底搞懂提示词工程如何驱动决策、任务链是如何被拆解和执行的、以及如何与外部工具（API）进行交互。无论你是想深入AI应用开发，还是仅仅对背后的原理感到好奇，这个实践都能让你获益匪浅。

2. 核心架构设计：构建智能体的“大脑”与“四肢”

要构建一个能“思考”的智能体，我们需要为其设计一个清晰的架构。这个架构主要包含两个核心部分：决策中心（大脑）和工具集（四肢）。大脑负责理解、规划和决策，四肢负责执行具体的动作。

2.1 决策中心：基于大语言模型的“指挥官”

智能体的“思考”能力，本质上来源于大语言模型。我们不会去训练一个模型，而是通过API（例如DeepSeek、GPT等）来调用现成的强大模型。决策中心的核心工作是提示词工程。我们需要设计一套“系统提示词”，来塑造智能体的角色和行为模式。

这套提示词需要明确告诉模型：

1. 身份与目标：你是一个AI助手，目标是帮助用户完成任务。
2. 能力与限制：你可以通过使用工具来获取信息或执行操作，你不能直接知道实时信息或操作外部系统。
3. 思考范式：你必须遵循“思考-行动-观察”的循环。收到用户请求后，先思考需要做什么、分几步、用什么工具；然后选择并调用一个工具；获得工具返回的结果后，观察结果并决定下一步是继续调用工具，还是直接给用户最终答案。
4. 输出格式：你必须用严格的JSON格式来回应，以便我的程序能解析你的决策。

一个基础的提示词设计如下：

你是一个任务执行AI助手。你的决策必须遵循严格的JSON格式。 格式说明： { “thought”: “你的思考过程，分析当前状况和下一步计划。”, “action”: { “name”: “要执行的动作名称，必须是以下工具之一：[‘search_web’， ‘calculate’， ‘final_answer’]”, “args”: {“key”: “value”} // 动作所需的参数 } } 工具列表： - search_web(query): 执行网络搜索。参数：query（搜索关键词）。 - calculate(expression): 执行数学计算。参数：expression（数学表达式，如’(5+3)*2’）。 - final_answer(answer): 向用户给出最终答案。参数：answer（你的回答文本）。 工作流程： 1. 用户提出请求。 2. 你根据请求，在“thought”中分析。 3. 你决定一个“action”。如果需要多步，就一步一步来。 4. 执行action后，你会收到“observation”（工具执行结果）。 5. 基于observation，你再次进入“thought-action”循环，直到任务完成，使用final_answer。 现在，开始处理用户请求：{user_input}

这个提示词将大模型变成了一个遵守固定规则的决策引擎，其输出是结构化的数据，而非自由文本，这是我们程序能与之对话的基础。

2.2 工具集：赋予智能体“动手”能力

智能体不能只“想”不“做”。工具集就是它可调用的函数集合。每个工具都是一个Python函数，执行特定的、模型自身无法完成的任务，比如计算、查询数据库、调用第三方API等。

在我们的原型里，先实现两个简单的工具：

import requests import json import math def search_web(query: str) -> str: """模拟网络搜索（实际应用中需接入真实搜索API）""" # 此处为演示，返回模拟结果。真实场景可调用Serper、Google Search等API。 mock_results = { “Python安装”: “访问python.org官网，下载对应操作系统的安装包，按照向导安装即可。”, “今天天气”: “北京：晴，15-25摄氏度；上海：多云，18-28摄氏度。”, “圆周率”: “圆周率π是一个数学常数，约等于3.14159。” } return mock_results.get(query, f”未找到关于‘{query}’的模拟信息。“) def calculate(expression: str) -> str: """执行安全的数学表达式计算""" # 警告：实际生产中，直接eval极其危险！这里仅作演示。 # 应采用ast.literal_eval或专用数学表达式解析库（如numexpr）。 try: # 极其简化的安全过滤，切勿用于生产环境！ if any(c for c in expression if c in “__” or c.isalpha() and c not in ‘pi e’): return “错误：表达式包含不安全字符。” result = eval(expression, {“__builtins__”: None}, {“pi”: math.pi, “e”: math.e}) return str(result) except Exception as e: return f”计算错误：{e}” def final_answer(answer: str) -> str: """这是一个特殊工具，用于终止循环并返回最终答案给用户""" return answer

注意：calculate函数中使用了eval()，这在实际项目中是高危操作，极易导致代码注入漏洞。此处仅用于原理演示。真实项目必须使用ast.literal_eval进行严格限制，或使用numexpr这类安全的库来评估数学表达式。

工具函数的设计原则是：功能单一、接口明确、返回字符串。智能体的“大脑”通过工具名称和参数来调用它们。

3. 核心循环实现：让智能体“动”起来

有了“大脑”（提示词+大模型）和“四肢”（工具集），我们需要一个核心驱动循环将它们串联起来，这就是经典的ReAct (Reasoning + Acting)循环的简化版。

3.1 主循环逻辑与状态管理

这个循环是智能体的“心跳”，它不断重复“思考-行动-观察”的过程，直到任务完成。我们需要管理一个“状态”，记录当前的对话历史、工具调用结果等。

class SimpleAgent: def __init__(self, api_key: str, model: str = “deepseek-chat”): self.api_key = api_key self.model = model self.base_url = “https://api.deepseek.com/v1/chat/completions” # 示例URL self.conversation_history = [] # 记录对话和观察，用于上下文 def call_llm(self, messages: list) -> dict: """调用大模型API""" headers = { “Authorization”: f”Bearer {self.api_key}”, “Content-Type”: “application/json” } data = { “model”: self.model, “messages”: messages, “temperature”: 0.1, # 低温度，使输出更确定、更遵循格式 “max_tokens”: 1000 } try: response = requests.post(self.base_url, headers=headers, json=data, timeout=30) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: # 处理网络或API错误 print(f”API调用失败：{e}”) if hasattr(e, ‘response’) and e.response is not None: print(f”错误详情：{e.response.text}”) # 返回一个模拟响应，防止程序崩溃，便于演示 return {“choices”: [{“message”: {“content”: ‘{“thought”: “API连接失败”， “action”: {“name”: “final_answer”， “args”: {“answer”: “网络服务暂时不可用，请稍后再试。”}}}}’}]} def parse_llm_response(self, response_text: str) -> dict: """解析模型返回的JSON字符串""" try: # 模型返回的内容可能包含markdown代码块标记，需要清理 cleaned_text = response_text.strip() if cleaned_text.startswith(‘```json’): cleaned_text = cleaned_text[7:] if cleaned_text.endswith(‘```’): cleaned_text = cleaned_text[:-3] cleaned_text = cleaned_text.strip() return json.loads(cleaned_text) except json.JSONDecodeError as e: print(f”JSON解析失败！原始响应：{response_text}”) # 解析失败时，返回一个让智能体结束的指令 return { “thought”: “我的响应格式出错了。”, “action”: { “name”: “final_answer”, “args”: {“answer”: “抱歉，我处理您的请求时出现了内部错误。”} } } def run(self, user_input: str, max_steps: int = 10) -> str: """运行智能体主循环""" print(f”用户：{user_input}”) # 初始化系统提示词，包含工具定义和流程规则 system_prompt = “””你是一个任务执行AI助手...（此处填入2.1节完整的提示词）“””.format(user_input=user_input) messages = [{“role”: “system”， “content”: system_prompt}] self.conversation_history = messages.copy() step = 0 while step < max_steps: step += 1 print(f”\n—- 第{step}步 —-“) # 1. 思考：调用LLM获取决策 llm_response = self.call_llm(self.conversation_history) llm_message = llm_response[“choices”][0][“message”] llm_content = llm_message[“content”] print(f”AI思考：{llm_content}”) # 将AI的回复加入历史，维持上下文 self.conversation_history.append({“role”: “assistant”， “content”: llm_content}) # 2. 解析决策 decision = self.parse_llm_response(llm_content) thought = decision.get(“thought”， “无思考内容”) action = decision.get(“action”， {}) action_name = action.get(“name”) action_args = action.get(“args”， {}) print(f”解析结果：思考 - {thought}， 行动 - {action_name}， 参数 - {action_args}”) # 3. 执行行动 if action_name == “final_answer”: final_msg = action_args.get(“answer”， “任务完成。”) print(f”最终答案：{final_msg}”) return final_msg # 映射并调用工具 tool_map = { “search_web”: search_web, “calculate”: calculate, } if action_name in tool_map: tool_func = tool_map[action_name] try: # 将参数字典展开为关键字参数传入函数 observation = tool_func(**action_args) except TypeError as e: observation = f”工具调用参数错误：{e}” except Exception as e: observation = f”工具执行异常：{e}” else: observation = f”错误：未知的工具名称‘{action_name}’。可用工具有：{list(tool_map.keys())}” print(f”工具执行结果（观察）：{observation}”) # 4. 观察：将结果作为新一轮的“用户消息”加入历史，驱动下一轮思考 observation_msg = f”上一次你决定执行 {action_name}， 结果是：{observation}。请根据这个结果继续思考下一步。” self.conversation_history.append({“role”: “user”， “content”: observation_msg}) # 循环超过最大步数，强制结束 return “任务过于复杂或陷入循环，已终止。”

这个SimpleAgent类封装了整个智能体的生命周期。run方法中的while循环就是核心。它每次迭代都：请求模型思考 -> 解析JSON决策 -> 执行对应工具 -> 将结果反馈给模型。直到模型决定调用final_answer工具，循环终止。

3.2 关键参数解析与配置心得

在实现过程中，有几个关键参数和配置点直接影响智能体的表现：

1. 系统提示词的温度（temperature）：在call_llm函数中，我们设置了“temperature”: 0.1。这个参数控制模型输出的随机性。范围通常在0到2之间。值越低（如0.1），输出越确定、可预测，更严格地遵循指令格式，适合需要稳定JSON输出的场景。值越高，输出越有创造性，但也更可能不按格式来。对于任务型智能体，低温度是首选。

2. 最大步数（max_steps）：在run方法中，我们设置了max_steps=10。这是一个重要的安全阀，防止智能体陷入无限循环或执行过于冗长的任务链。如果10步之后还没得到final_answer，就强制终止。在实际应用中，你可以根据任务复杂度调整这个值，并设计更优雅的超时处理，比如提示用户任务可能太复杂。

3. 错误处理与鲁棒性：代码中包含了多处try...except块。这是智能体能否稳定运行的关键。API可能超时、返回非JSON内容、工具可能出错。良好的错误处理能将异常转化为智能体能理解的“观察”（observation），让它有机会调整策略，而不是让整个程序崩溃。例如，在parse_llm_response中，如果JSON解析失败，我们会返回一个导向final_answer的决策，优雅地结束会话并告知用户。

实操心得：调试智能体时，一定要把每一步的thought、action和observation打印出来。这是理解你智能体“思维过程”的唯一窗口。很多时候效果不好，不是逻辑问题，而是提示词没写清楚，或者模型的输出格式有轻微偏差（比如多了个换行符），导致JSON解析失败。逐步打印日志能帮你快速定位问题层。

4. 运行示例与效果分析

让我们用写好的智能体跑几个例子，看看它是如何“思考”的。

首先初始化智能体（你需要替换成自己的API Key）：

# 注意：以下为示例，你需要使用有效的API Key # 为安全起见，建议从环境变量读取API Key import os # api_key = os.getenv(“DEEPSEEK_API_KEY”) api_key = “your_api_key_here” # 临时测试用，切勿提交到代码仓库！ agent = SimpleAgent(api_key=api_key, model=“deepseek-chat”)

示例一：处理复合问题

result = agent.run(“先帮我计算一下(15+27)除以6等于多少，然后再搜索一下‘Python安装’的步骤。”) print(“\n=== 最终回复 ===\n”, result)

可能的执行日志输出：

用户：先帮我计算一下(15+27)除以6等于多少，然后再搜索一下‘Python安装’的步骤。 —- 第1步 —- AI思考：{“thought”: “用户提出了一个包含两个连续任务的请求。首先需要计算数学表达式(15+27)/6，然后需要搜索‘Python安装’。我应该按顺序执行。第一步，使用calculate工具。”, “action”: {“name”: “calculate”, “args”: {“expression”: “(15+27)/6”}}} 解析结果：思考 - 用户提出了一个包含两个连续任务的请求…， 行动 - calculate， 参数 - {‘expression’: ‘(15+27)/6’} 工具执行结果（观察）：7.0 —- 第2步 —- AI思考：{“thought”: “计算结果是7.0。第一步任务完成。现在开始第二步任务：搜索‘Python安装’。使用search_web工具。”, “action”: {“name”: “search_web”, “args”: {“query”: “Python安装”}}} 解析结果：思考 - 计算结果是7.0。第一步任务完成…， 行动 - search_web， 参数 - {‘query’: ‘Python安装’} 工具执行结果（观察）：访问python.org官网，下载对应操作系统的安装包，按照向导安装即可。 —- 第3步 —- AI思考：{“thought”: “两个任务都已完成。计算结果是7.0，搜索得到了Python安装的步骤。现在可以整合信息，给用户最终答案了。”, “action”: {“name”: “final_answer”, “args”: {“answer”: “计算结果是：7.0。\\n\\n关于Python安装：访问python.org官网，下载对应操作系统的安装包，按照向导安装即可。”}}} 解析结果：思考 - 两个任务都已完成…， 行动 - final_answer， 参数 - {‘answer’: ‘计算结果是：7.0…’} 最终答案：计算结果是：7.0。 关于Python安装：访问python.org官网，下载对应操作系统的安装包，按照向导安装即可。

效果分析：智能体成功识别了这是一个序列任务。它没有试图一次性完成所有事，而是先思考，决定第一步做什么（计算），执行后获得观察结果，再基于此思考第二步（搜索），最后整合信息给出答案。这完美体现了“思考-行动-观察”的循环价值。

示例二：处理无法直接回答的问题

result = agent.run(“今天的天气怎么样？”) print(“\n=== 最终回复 ===\n”, result)

可能的执行日志输出：

用户：今天的天气怎么样？ —- 第1步 —- AI思考：{“thought”: “用户询问实时天气。我无法直接获取实时信息，但我有search_web工具可以模拟搜索天气信息。我应该使用这个工具。”, “action”: {“name”: “search_web”, “args”: {“query”: “今天天气”}}} 解析结果：思考 - 用户询问实时天气…， 行动 - search_web， 参数 - {‘query’: ‘今天天气’} 工具执行结果（观察）：北京：晴，15-25摄氏度；上海：多云，18-28摄氏度。 —- 第2步 —- AI思考：{“thought”: “搜索工具返回了一些模拟的天气信息。由于用户没有指定城市，我提供了示例信息。现在可以将这个信息作为答案返回给用户。”, “action”: {“name”: “final_answer”, “args”: {“answer”: “根据模拟搜索的结果（示例数据）：北京：晴，15-25摄氏度；上海：多云，18-28摄氏度。请注意，要获取精确的实时天气，请使用专业的天气应用或网站。”}}} 解析结果：思考 - 搜索工具返回了一些模拟的天气信息…， 行动 - final_answer， 参数 - {‘answer’: ‘根据模拟搜索的结果…’} 最终答案：根据模拟搜索的结果（示例数据）：北京：晴，15-25摄氏度；上海：多云，18-28摄氏度。请注意，要获取精确的实时天气，请使用专业的天气应用或网站。

效果分析：智能体知道自己“不能直接知道天气”，但它知道自己“能通过工具搜索”。于是它规划了使用search_web工具的步骤，并在最终答案中诚实地说明了信息的来源和局限性。这展示了智能体对自身能力边界的认知和利用工具扩展能力的过程。

5. 进阶优化与扩展方向

我们实现了一个基础但完整的智能体原型。要让其更实用、更强大，可以从以下几个方向进行扩展：

5.1 增强工具能力与安全性

1. 接入真实API：将search_web替换为真实的搜索引擎API（如Serper Dev、Google Custom Search JSON API）。为calculate函数实现一个安全的表达式求值器（使用numexpr库）。
2. 增加更多工具：
   • 文件操作：读取用户上传的CSV/Excel文件并进行分析。
   • 代码执行：在一个安全的沙箱环境中执行简单的Python代码片段（需极度谨慎）。
   • 数据库查询：连接数据库，执行SQL查询（使用参数化查询防止注入）。
   • 网络请求：调用其他RESTful API，获取股票价格、新闻摘要等。
3. 工具描述自动化：目前工具列表是手写在提示词里的。可以写一个函数自动收集所有工具函数的名称、描述和参数schema，动态生成提示词部分，这样新增工具时就不必手动修改提示词了。

5.2 改进决策与记忆机制

1. 短期记忆（上下文管理）：我们当前的conversation_history会不断增长，可能很快超过模型的最大上下文长度（如DeepSeek V4-Pro的1048565 tokens）。需要实现一个上下文窗口管理策略，比如只保留最近N轮对话，或者对历史对话进行智能摘要（Summarization）。
2. 长期记忆（向量数据库）：为智能体配备一个“笔记本”。可以将重要的对话内容、工具执行结果转换成向量，存入像Chroma、Pinecone这样的向量数据库。当处理新任务时，先进行向量相似度搜索，找到相关的历史记忆，从而做出更连贯、个性化的决策。
3. 复杂任务规划：对于多步骤的复杂任务，可以引入更高级的规划器。例如，让模型先输出一个完整的任务计划（Task List），然后再逐步执行和勾选。这比一步一想的ReAct循环更适合宏观规划。

5.3 提升稳定性与用户体验

1. 输出格式加固：模型有时会不按JSON格式输出。除了代码中的清理逻辑，还可以在提示词中更加强调格式，甚至采用支持JSON Mode的API（如果所用模型支持）。另一种方案是使用“输出解析器”（Output Parser），例如LangChain提供的Pydantic解析器，它能以更强的约束引导模型输出。
2. 错误处理与重试：当工具调用失败或模型输出格式错误时，不应直接结束。可以设计一个重试机制，比如将错误信息反馈给模型，让它“反思”并尝试另一种方案，最多重试3次。
3. 人机交互与确认：对于高风险操作（如删除文件、发送邮件），可以让智能体在执行前，先调用一个ask_for_confirmation工具，将计划的操作呈现给用户，等待用户明确确认后再执行。

6. 常见问题与排查实录

在开发和调试这个智能体的过程中，我踩过不少坑。这里把一些典型问题和解决方法记录下来，希望能帮你节省时间。

一个典型的调试过程：当我第一次运行智能体时，它经常在第三步之后输出一些奇怪的文本而不是JSON。我打开了每一步的详细日志，发现模型在第二轮思考后，输出的内容开头有“好的，根据上一步的结果...”这样的自然语言，然后才是JSON。这污染了JSON解析。解决方法就是在parse_llm_response函数中加入更强大的文本清洗，并在系统提示词的开头用非常醒目的方式强调：“你必须且只能输出一个合法的JSON对象，不要有任何其他前缀、后缀或解释性文字。” 通常加粗或全大写能引起模型更多注意。

手撸一个智能体的过程，就像在教一个天赋异禀但缺乏常识的孩子如何一步步解决问题。你需要用最精确的语言（提示词）告诉它规则，为它准备好工具（函数），并设计好一个不会让它跑偏的流程（循环）。这个过程充满挑战，但当你看到它按照你的设计，有条不紊地拆解并完成一个复杂任务时，那种成就感是直接用现成平台无法比拟的。这个原型只是一个起点，你可以沿着上面提到的扩展方向，把它打造成一个真正能处理你日常工作流的得力助手。