1. 项目概述：手搓LLM的ReAct模式
去年在调试LangChain时第一次接触到ReAct模式，这种将推理（Reasoning）和行动（Action）结合的交互方式让我眼前一亮。最近在开发本地知识库问答系统时，发现单纯依靠prompt engineering难以处理复杂逻辑，于是决定从零实现一个ReAct框架。本文将分享如何用Python原生代码构建支持ReAct模式的大语言模型交互系统，包含完整的思维链实现和工具调用机制。
2. ReAct模式核心原理
2.1 模式结构解析
ReAct=Reasoning+Acting，其工作流程呈现典型的循环特征：
- 观察阶段：模型接收环境状态（如用户问题、工具输出）
- 推理阶段：生成包含思考过程的文本（"I need to search...")
- 行动阶段：输出可执行的行动指令（如调用搜索引擎）
- 反馈循环：将行动结果作为新输入继续处理
这种模式相比传统few-shot prompt的优势在于：
- 显式保留了中间推理过程
- 支持多工具组合调用
- 具备自我修正能力
2.2 关键技术组件
实现时需要三个核心模块：
class ReActAgent: def __init__(self): self.memory = [] # 对话历史记录 self.tools = {} # 可用工具集 def _parse_action(self, text): # 解析模型输出中的行动指令 pass def _run_tool(self, tool_name, params): # 执行具体工具调用 pass3. 完整实现步骤
3.1 基础环境搭建
建议使用transformers库加载本地模型：
pip install transformers torch测试用的7B量级模型配置：
from transformers import AutoModelForCausalLM, AutoTokenizer model = AutoModelForCausalLM.from_pretrained( "Llama-2-7b-chat-hf", device_map="auto", torch_dtype=torch.float16 ) tokenizer = AutoTokenizer.from_pretrained(model_path)3.2 提示词工程
设计包含以下要素的system prompt：
你是一个具备工具调用能力的AI助手，请按照以下格式响应： 思考：<你的推理过程> 行动：<工具名>|<JSON参数> 观察：<工具返回结果>示例用户提问："北京和上海哪个城市人口更多？"
理想输出：
思考：需要比较两个城市的人口数据，应该查询权威统计资料 行动：search_engine|{"query":"北京 2023年常住人口"}3.3 行动解析器实现
关键的正则匹配逻辑：
import re action_pattern = re.compile( r"行动：([a-z_]+)\|({.*?})", flags=re.DOTALL ) def parse_action(text): match = action_pattern.search(text) if match: return match.group(1), json.loads(match.group(2)) return None, None3.4 工具管理系统
注册工具的装饰器实现：
def register_tool(name): def decorator(func): self.tools[name] = func return func return decorator @register_tool("search_engine") def search(query: str): # 实际接入搜索引擎API return f"找到{len(results)}条结果"4. 核心问题解决方案
4.1 思维链中断处理
常见问题：模型忘记输出行动指令 解决方案：在每次推理时注入历史交互记录
def build_prompt(question): history = "\n".join(self.memory[-5:]) return f"{history}\n问题：{question}"4.2 工具参数验证
使用Pydantic进行强类型校验：
from pydantic import BaseModel class SearchParams(BaseModel): query: str limit: int = 3 def validate_params(params, model): try: return model(**params).dict() except ValidationError as e: return {"error": str(e)}4.3 多轮对话管理
通过对话状态机维护上下文：
class DialogState: INIT = 0 AWAITING_ACTION = 1 AWAITING_OBSERVATION = 2 def process(self, input_text): if self.state == DialogState.INIT: prompt = self.build_prompt(input_text) output = self.llm.generate(prompt) self.state = DialogState.AWAITING_ACTION5. 性能优化技巧
5.1 流式输出处理
使用生成器减少等待时间：
def stream_response(prompt): for chunk in self.llm.stream(prompt): if "行动：" in chunk: yield "[ACTION DETECTED]" break yield chunk5.2 工具调用并行化
对于独立工具使用多线程：
from concurrent.futures import ThreadPoolExecutor with ThreadPoolExecutor() as executor: futures = { name: executor.submit(tool, **params) for name, params in actions.items() } results = { k: f.result() for k, f in futures.items() }5.3 缓存机制
使用LRU缓存重复查询：
from functools import lru_cache @register_tool("calculator") @lru_cache(maxsize=100) def calculate(expression: str): return eval(expression) # 注意安全风险！6. 安全防护方案
6.1 工具沙箱
限制危险操作：
import restrictedpython def safe_exec(code): """限制可访问的Python内置函数""" locals_dict = {"__builtins__": safe_builtins} bytecode = restrictedpython.compile_restricted(code) exec(bytecode, {}, locals_dict) return locals_dict.get("result")6.2 输出过滤
防止敏感信息泄露：
BLACKLIST = ["API_KEY", "password"] def sanitize_output(text): for term in BLACKLIST: text = text.replace(term, "[REDACTED]") return text7. 效果评估与调优
7.1 测试用例设计
应覆盖以下场景：
- 单工具调用
- 多工具串联
- 参数传递错误
- 模糊问题处理
示例测试集：
test_cases = [ ("今天北京天气怎样？", ["weather_api"]), ("李白和杜甫谁年龄大？", ["search_engine", "calculator"]), ("请画一只猫", ["image_generator"]) ]7.2 评估指标
建议监控：
- 工具调用准确率
- 平均交互轮次
- 异常处理成功率
- 响应延迟P99值
8. 生产环境部署建议
8.1 服务化封装
使用FastAPI暴露HTTP接口：
from fastapi import FastAPI app = FastAPI() agent = ReActAgent() @app.post("/chat") async def chat(query: str): return {"response": agent.process(query)}8.2 持久化方案
对话历史存储方案对比：
| 方案 | 优点 | 缺点 |
|---|---|---|
| SQLite | 零配置 | 扩展性差 |
| Redis | 高性能 | 需要独立服务 |
| PostgreSQL | 功能完善 | 运维复杂 |
9. 扩展方向
9.1 多模态支持
扩展行动指令类型：
{ "action": "image_generation", "params": { "prompt": "a cat wearing sunglasses", "style": "cartoon" } }9.2 动态工具加载
实现热插拔工具：
def load_tool_module(path): spec = importlib.util.spec_from_file_location("tool", path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module经过三周的迭代开发，这个ReAct框架已成功应用于我们的客服系统，复杂问题解决率提升40%。最关键的收获是：一定要给模型充足的"思考空间"，在prompt中保留完整的推理链条比调参更重要。下一步计划加入自动工具组合学习功能，让模型能自主发现工具的使用模式。
