手撸AI对话助手：基于ReAct框架实现可解释的思考过程展示

1. 项目概述：从“黑盒”到“白盒”的AI对话体验

最近在捣鼓大模型应用开发，发现一个挺有意思的现象：市面上的AI对话助手，绝大多数都是“黑盒”操作。你输入问题，它直接给你答案，中间发生了什么，它“想”了些什么，你一概不知。这就像让一个顶尖的数学家帮你解题，他只给你最终答案，却不展示草稿纸。对于学习者和开发者来说，这中间缺失的“思考过程”恰恰是价值最高的部分。

于是，我决定自己动手，“手撸”一个能带上思考过程的AI对话助手。这个项目的核心目标不是做一个功能多强大的聊天机器人，而是实现一个“白盒化”的推理过程展示。让AI在回答问题时，能像人一样，把内心的推理链条、权衡取舍、甚至可能的错误尝试都“说”出来。这对于理解大模型的工作原理、调试提示词（Prompt）、甚至用于教学演示，都极具价值。

这个项目适合所有对AI应用开发感兴趣的朋友，无论你是想深入理解大模型推理机制的研究者，还是希望为自己的产品增加可解释性功能的开发者，亦或是单纯好奇AI“脑子里”在想什么的爱好者，都能从中获得启发。接下来，我将从设计思路、核心实现、到避坑经验，完整分享这次“手撸”之旅。

2. 核心思路拆解：如何让AI“说出”它的思考？

要让AI展示思考过程，我们不能依赖现成的、封装好的API，因为它们通常只返回最终结果。我们需要深入到与大模型交互的“原始”层面，并对对话流程进行重新设计。

2.1 设计范式选择：ReAct与Chain-of-Thought

目前，让AI展示结构化思考过程的主流范式有两种：

1. 思维链（Chain-of-Thought, CoT）： 要求模型在给出最终答案前，先一步步地展示其推理步骤。通常通过在提示词中要求“让我们一步步思考”来实现。这种方式简单直接，但思考过程是线性的、内省的，不涉及对外部工具或知识的调用。

2. 推理与行动（Reasoning and Acting, ReAct）： 这是一个更强大的框架。它让模型循环执行“思考（Thought）- 行动（Action）- 观察（Observation）”的步骤。

   • Thought: AI分析当前状况，决定下一步要做什么。
   • Action: 根据Thought，执行一个具体动作，比如调用一个计算器API、搜索网络、查询数据库等。
   • Observation: 获取Action执行后的结果。 这个循环会一直持续，直到模型认为已经收集到足够信息，可以给出最终答案（Answer）。

对于我们的“带思考过程的对话助手”，ReAct范式更为合适。因为它不仅能展示“想”的过程（Thought），还能展示“做”的过程（Action & Observation），整个推理轨迹更加完整和具有可操作性。我们的系统将扮演一个“调度员”的角色，管理这个ReAct循环。

2.2 系统架构设计

一个基础的、带思考过程展示的对话助手，其核心架构可以分为三层：

1. 交互层： 负责接收用户问题，并实时流式输出AI的整个思考过程（包括Thought, Action, Observation）和最终答案。这里的关键是流式输出，让用户能看到“逐字吐出”的思考，而不是等待很久后一次性显示一大段文字。

2. 推理引擎层（核心）： 这是本项目的大脑。它包含一个提示词模板，该模板定义了ReAct的格式，并指示模型遵循这个格式输出。引擎的工作是：

   • 将用户问题与对话历史、ReAct模板结合，生成完整的提示词。
   • 调用大模型API。
   • 解析模型的返回结果，严格按照“Thought: ...\nAction: ...\nObservation: ...”的格式进行切分。
   • 根据Action的类型（如search,calculate），调用相应的工具层功能。
   • 将Observation结果反馈给模型，进行下一轮循环。

3. 工具层： 为AI提供“手”和“眼”。根据Action指令执行具体任务。例如：

   • search: 调用搜索引擎API（如Serper， Tavily）获取实时信息。
   • calculate: 调用Python的eval在一个安全沙箱中执行计算（注意：生产环境需要严格的安全处理）。
   • lookup: 查询内部知识库或数据库。
   • finish: 特殊Action，表示推理结束，后面跟着的就是最终Answer。

这个架构清晰地将“思考”、“决策”、“执行”分离，使得每一步都变得可见、可记录、可调试。

3. 核心实现细节与工具选型

有了架构蓝图，接下来就是选用合适的“砖瓦”来搭建。这里我选择Python生态，因为它拥有最丰富的大模型相关库。

3.1 大模型选择与提示词工程

模型选择： 要实现高质量的ReAct推理，模型必须具备良好的指令遵循（Instruction Following）能力和格式输出（Structured Output）能力。开源模型中，DeepSeek-V2-Chat、Qwen2.5-72B-Instruct、Llama 3.1 70B表现都非常出色。闭源API中，GPT-4o、Claude 3.5 Sonnet是顶级选择。考虑到成本和调试便利性，我在开发初期使用了Qwen2.5-7B-Instruct的本地部署版本，后期测试切换到了GPT-4o的API。

提示词模板设计： 这是项目的灵魂。一个糟糕的提示词会导致模型不按格式输出，整个解析循环就会崩溃。我的核心模板如下：

你是一个智能助手，必须使用以下格式回答问题： Question: 用户输入的问题 Thought: 你需要思考如何一步步解决问题。你可以使用工具。这是你的内心独白。 Action: 你需要执行的动作，必须是以下之一：`search[查询词]`, `calculate[数学表达式]`, `lookup[关键词]`, `finish[最终答案]`。 Observation: 动作执行后的结果。 ... (这个 Thought/Action/Observation 循环可以重复多次) Thought: 我现在有了所有信息，可以给出最终答案了。 Action: finish Answer: 给用户的最终答案，应详尽且友好。

关键技巧：

• 明确角色和格式： 开头就定下基调，强调“必须使用以下格式”。
• 详细定义Action： 将Action限定为几个明确的选项，并给出示例，这大大降低了模型“胡言乱语”的概率。
• 示例（Few-shot）注入： 在模板中加入1-2个完整的ReAct循环示例，能极大提升模型的格式遵循能力。例如，可以先展示一个“计算北京到上海距离”的完整过程。
• 分隔符清晰： 使用Thought:、Action:、Observation:这样的明确标签，便于后续用正则表达式进行解析。

3.2 流式输出与解析器实现

流式输出： 直接使用大模型API的流式接口（如OpenAI的stream=True）。每当收到一个token（词元），就将其追加到当前正在输出的部分（可能是Thought，也可能是Answer）。前端（或命令行）需要实时渲染这些内容。这里的一个体验优化点是：将Thought、Action、Observation用不同的颜色或缩进区分显示，让思考过程一目了然。

解析器实现： 这是最需要鲁棒性的部分。模型并不总是完美遵守格式。我的解析逻辑如下：

import re def parse_model_output(text: str): """ 解析模型返回的文本，提取出 Thought, Action, Observation。 使用正则表达式进行容错处理。 """ patterns = { 'thought': r'Thought:\s*(.*?)(?=\nAction:|$)', 'action': r'Action:\s*(\w+)\[([^\]]+)\]', 'observation': r'Observation:\s*(.*?)(?=\nThought:|$)', 'answer': r'Answer:\s*(.*)' } result = {} for key, pattern in patterns.items(): match = re.search(pattern, text, re.DOTALL) if match: if key == 'action': result['action_type'] = match.group(1) result['action_input'] = match.group(2) else: result[key] = match.group(1).strip() return result

注意： 正则表达式虽然强大，但面对模型千奇百怪的输出（比如多一个空格，换行符不一致），仍然可能失败。因此，必须加入重试和降级机制。例如，如果解析Action失败，可以尝试用更宽松的正则，或者直接截取“Action:”后面的第一行文本作为备选。在最坏情况下，应能优雅地回退到直接输出模型原始回复，而不是让程序崩溃。

3.3 工具执行与安全考量

工具层是AI与真实世界交互的桥梁，也是安全风险的高发地。

• 搜索工具： 我选择了Tavily Search API，它是为AI Agent优化的搜索引擎，返回的结果已经是结构化的摘要，非常适合给模型阅读。你也可以用Serper或自己搭建一个Google Custom Search JSON API。

  import requests def tool_search(query: str): api_key = "YOUR_TAVILY_KEY" url = "https://api.tavily.com/search" payload = {"query": query, "api_key": api_key} response = requests.post(url, json=payload) data = response.json() # 通常返回 data['results'][0]['content'] 作为观察结果 return data['results'][0]['content'] if data.get('results') else "未找到相关信息。"

• 计算工具：这是最高风险点！绝对不能让模型生成的数学表达式直接在你的主机上执行。

  import ast import operator as op # 允许的安全操作符 allowed_operators = {ast.Add: op.add, ast.Sub: op.sub, ast.Mult: op.mul, ast.Div: op.truediv, ast.Pow: op.pow, ast.USub: op.neg} def safe_eval(expr: str): """极其简化的安全计算，仅用于演示。生产环境需使用沙箱如`pyodide`或在独立容器中运行。""" try: node = ast.parse(expr, mode='eval').body def _eval(node): if isinstance(node, ast.Num): # Python 3.8+ return node.n elif isinstance(node, ast.Constant): return node.value elif isinstance(node, ast.BinOp): return allowed_operators[type(node.op)](_eval(node.left), _eval(node.right)) elif isinstance(node, ast.UnaryOp): return allowed_operators[type(node.op)](_eval(node.operand)) else: raise TypeError(f"不支持的表达式类型: {node}") return _eval(node) except Exception as e: return f"计算错误: {e}"

  重要警告： 上面的safe_eval函数只是一个极其基础的示例，远未达到生产级安全要求。对于涉及用户输入的任何代码执行，必须使用完全隔离的沙箱环境，例如Pyodide（在浏览器WASM中运行）、Docker容器或专门的沙箱服务（如E2B的代码执行环境）。永远不要信任来自模型的任意代码。

4. 完整实现流程与代码剖析

让我们用一个具体的例子，串联起整个系统的运行流程。假设用户问题是：“梅西在2022年卡塔尔世界杯决赛中进了几个球？这场比赛阿根廷的对手是谁？”

4.1 系统初始化与对话循环

首先，我们需要维护一个对话历史列表，用于存储多轮交互的上下文。

class ReActChatAssistant: def __init__(self, model_api, tools): self.model_api = model_api # 封装好的模型调用函数 self.tools = tools # 工具字典，如 {'search': tool_search, 'calculate': safe_eval} self.conversation_history = [] # 格式：[{"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}] def _build_prompt(self, user_input): """构建包含历史、指令和当前问题的完整提示词""" system_prompt = """（这里是上面定义的详细ReAct指令模板）""" history_text = "\n".join([f"{msg['role']}: {msg['content']}" for msg in self.conversation_history[-6:]]) # 保留最近3轮对话 full_prompt = f"{system_prompt}\n\n{history_text}\n\nQuestion: {user_input}\n\n" return full_prompt

4.2 单轮ReAct循环执行

当用户输入问题后，系统进入核心循环：

def generate_response(self, user_input): # 1. 构建提示词 prompt = self._build_prompt(user_input) full_chain_text = "" # 用于记录模型本次生成的所有文本 final_answer = None # 2. 开始ReAct循环（设置最大步数防止无限循环） max_steps = 10 for step in range(max_steps): # 2.1 调用模型，获取流式响应 model_response = "" for chunk in self.model_api.call_streaming(prompt + full_chain_text): # 这里是流式处理，每个chunk都实时输出到前端 model_response += chunk # 实时解析并高亮显示当前输出的部分（如Thought） full_chain_text += model_response # 2.2 解析本次模型的输出 parsed = parse_model_output(model_response) # 2.3 处理Action if parsed.get('action_type') == 'finish': final_answer = parsed.get('answer', '') break # 循环结束 action_type = parsed.get('action_type') action_input = parsed.get('action_input') if action_type in self.tools: # 2.4 执行工具，获取Observation observation = f"调用工具 {action_type}[{action_input}] 的结果：" try: tool_result = self.tools[action_type](action_input) observation += f"\n{tool_result}" except Exception as e: observation += f"\n工具执行出错: {e}" else: observation = f"错误：未知的Action类型 '{action_type}'。" # 2.5 将Observation追加到链中，作为下一轮模型的输入 full_chain_text += f"\nObservation: {observation}\n" # 3. 保存到历史 self.conversation_history.append({"role": "user", "content": user_input}) # 保存的是完整的思考链，还是只保存最终答案？取决于你想让历史记住什么。 # 如果保存完整链，下次对话模型会看到之前所有的思考，可能有助于复杂任务，但会消耗大量token。 # 通常只保存最终答案更经济。 self.conversation_history.append({"role": "assistant", "content": final_answer or "未能得出最终答案。"}) return final_answer, full_chain_text # 返回答案和完整的思考过程记录

4.3 运行实例推演

对于我们的问题，系统运行过程可能如下（简化显示）：

1. 第一轮模型输出:

   Thought: 用户问了两个问题：1. 梅西进了几个球。2. 阿根廷的对手是谁。这两个问题都关于2022年世界杯决赛，我需要查找这场比赛的详细信息。 Action: search[2022年卡塔尔世界杯决赛 梅西 进球 阿根廷 对手]

2. 系统执行Action： 调用tool_search，获取到搜索结果摘要：“2022年卡塔尔世界杯决赛于2022年12月18日举行，对阵双方是阿根廷和法国。阿根廷通过点球大战获胜。梅西在比赛中梅开二度，打入两球。”
3. 系统将Observation加入上下文：Observation: 调用工具 search[2022年卡塔尔世界杯决赛 梅西 进球 阿根廷 对手] 的结果：2022年卡塔尔世界杯决赛于2022年12月18日举行，对阵双方是阿根廷和法国。阿根廷通过点球大战获胜。梅西在比赛中梅开二度，打入两球。
4. 第二轮模型输出:

   Thought: 从观察中我已经得到了答案。梅西进了两个球（梅开二度）。阿根廷的对手是法国队。现在我可以给出最终答案了。 Action: finish Answer: 在2022年卡塔尔世界杯决赛中，梅西为阿根廷队打入了两球。这场决赛阿根廷队的对手是法国队。

5. 循环结束，返回最终答案和完整的思考链文本。

通过这个流程，用户不仅得到了答案，还看到了AI“决定去搜索”、“搜索了什么关键词”、“从搜索结果中提炼了什么信息”以及“如何根据信息组织答案”的全过程。

5. 避坑指南与实战经验

在实际开发中，我遇到了不少坑，这里总结出最重要的几点经验。

5.1 模型不遵循格式的应对策略

这是最常见的问题。模型可能会输出“我认为应该先搜索...”，而不是“Thought: 我认为应该先搜索...”。

• 强化提示词： 在系统指令中反复强调格式，并使用分隔符。例如用---将指令和上下文分开。可以写成：“你必须严格按照以下格式输出，每一轮都包含Thought、Action、Observation三个部分，用换行隔开：\n\nThought: ...\nAction: ...\nObservation: ...\n\n---\n\n现在开始回答。”
• 后处理与重试： 如果解析失败，可以将解析失败的那部分模型输出，连同一条修正指令（如“你刚才的输出格式有误，请严格按照Thought/Action/Observation格式重新回答”）一起，再次发送给模型。这通常能纠正错误。
• 使用支持结构化输出的模型/库： 这是终极解决方案。像OpenAI的response_format参数（指定为JSON Schema）、Anthropic的tools/tool_choice参数，或者LangChain的StructuredOutputParser，可以强制模型以指定的JSON格式输出，从根本上杜绝格式错误。这是生产环境的推荐做法。

5.2 循环失控与超时处理

模型有时会陷入“思考-行动”的死循环，或者提出无法执行的动作。

• 设置最大循环次数： 如上文代码所示，必须设置max_steps（如10次），超过则强制终止，并返回已收集的信息和超时提示。
• 超时控制： 对每一次模型调用和工具调用都设置超时（timeout），避免因某个环节卡死导致整个服务无响应。
• 定义清晰的工具列表和终止条件： 在提示词中明确列出所有可用的Action类型，并强调finish是唯一的结束方式。对于无效的Action，在Observation中明确反馈错误，引导模型回到正轨。

5.3 成本与性能优化

流式输出和多次模型调用（ReAct循环）会导致Token消耗增加和响应时间变长。

• 压缩历史上下文： 对话历史是Token消耗大户。可以对历史消息进行摘要（Summarization），而不是完整保存。例如，在每轮对话后，用一个小模型将冗长的思考过程总结成几句话存入历史。
• 选择性流式输出： 不一定所有内容都需要流式输出。可以考虑只对最终的Answer进行流式输出，而将Thought和Observation一次性返回。或者，先快速流式输出Thought，等工具执行和模型生成下一步时再输出后续内容，提升用户体验上的“流畅感”。
• 使用更小、更快的模型进行思考： 可以采用“大小模型协同”的策略。让一个低成本、快速的小模型（如Qwen2.5-1.5B）负责生成Thought和Action，而让一个能力强的大模型在最后一步根据所有Observation来合成最终Answer。这能在保证答案质量的同时，显著降低中间步骤的成本和延迟。

5.4 前端展示的体验打磨

思考过程的展示方式直接影响用户体验。

• 差异化视觉呈现： 在Web界面或命令行中，用不同颜色区分Thought（灰色/斜体）、Action（蓝色/加粗）、Observation（绿色）、Answer（高亮）。这能让用户一眼看清推理脉络。
• 可折叠/展开的细节： 对于很长的Observation（如搜索返回的大段文本），可以默认折叠，只显示摘要，用户点击后再展开详情，保持界面清爽。
• 交互与干预： 高级模式下，可以允许用户在AI思考过程中进行干预。例如，当AI准备执行一个看起来不靠谱的Action时，用户可以手动修改或确认。这实现了“人在回路”（Human-in-the-loop），让AI成为真正的助手。

手撸一个带思考过程的AI对话助手，远不止是调用API那么简单。它要求你深入理解提示词工程、流程控制、安全编程和用户体验。这个过程本身，就是一次对AI如何“思考”的绝佳探索。当你看到自己构建的系统，像剥洋葱一样将AI的黑盒推理一层层展现出来时，那种成就感和对技术的理解，是使用现成产品无法比拟的。这个项目可以作为一个强大的基础，未来你可以轻松地为它增加更多工具（如画图、写邮件、操作数据库），将其扩展成一个功能丰富的个人AI智能体（Agent）。