1. 项目概述手搓LLM的ReAct模式去年在调试LangChain时第一次接触到ReAct模式这种将推理Reasoning和行动Action结合的交互方式让我眼前一亮。最近在开发本地知识库问答系统时发现单纯依靠prompt engineering难以处理复杂逻辑于是决定从零实现一个ReAct框架。本文将分享如何用Python原生代码构建支持ReAct模式的大语言模型交互系统包含完整的思维链实现和工具调用机制。2. ReAct模式核心原理2.1 模式结构解析ReActReasoningActing其工作流程呈现典型的循环特征观察阶段模型接收环境状态如用户问题、工具输出推理阶段生成包含思考过程的文本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_mapauto, torch_dtypetorch.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_])\|({.*?}), flagsre.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(maxsize100) 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中保留完整的推理链条比调参更重要。下一步计划加入自动工具组合学习功能让模型能自主发现工具的使用模式。