2026/9/12 17:52:13

Python实现极简智能体:200行代码构建Agent核心框架

Python实现极简智能体:200行代码构建Agent核心框架 1. 项目概述用Python构建极简Agent的核心思路200行代码实现一个智能体听起来像天方夜谭其实只要抓住Agent技术的三个核心要素意图理解、工具调用和决策循环就能搭建出具备基础智能的微型系统。这个项目将使用纯Python实现不依赖任何复杂框架通过最精简的代码揭示智能体运作的本质。现代智能体开发通常有两种技术路线基于ReAct的复杂规划型和基于Function Calling的轻量执行型。我们选择后者作为实现方案因为它更符合极简的定位且能直观展示LLM如何与外部工具交互。整个系统的工作流程可以简化为接收用户输入→分析意图→选择工具→执行动作→返回结果。2. 核心组件拆解与实现2.1 意图理解模块class IntentParser: def __init__(self, llm_client): self.llm llm_client def parse(self, user_input): prompt f分析用户意图并返回JSON格式结果。可选意图类型 - weather_query: 天气查询 - math_calculation: 数学计算 - general_qa: 通用问答 输入{user_input} 输出格式{intent: 类型, params: {参数名: 参数值}} response self.llm.generate(prompt) try: return json.loads(response) except json.JSONDecodeError: return {intent: general_qa, params: {}}这个简易解析器展示了几个关键设计点限定意图范围提高准确率结构化输出便于后续处理添加异常处理保证鲁棒性2.2 工具调用系统class ToolManager: def __init__(self): self.tools { get_weather: { func: self._get_weather, desc: 获取指定城市的天气信息 }, calculate: { func: self._calculate, desc: 执行数学计算 } } def execute(self, tool_name, params): if tool_name not in self.tools: raise ValueError(f未知工具{tool_name}) return self.tools[tool_name][func](params) def _get_weather(self, params): # 模拟天气API调用 return f{params[location]}天气晴25℃ def _calculate(self, params): try: result eval(params[expression]) return f计算结果{result} except: return 计算失败请检查表达式工具系统的设计要点统一注册管理机制每个工具包含执行函数和描述安全的参数传递方式3. 决策循环与执行引擎3.1 主控制循环实现class AgentEngine: def __init__(self): self.llm LiteLLMClient() # 简化的LLM客户端 self.parser IntentParser(self.llm) self.tools ToolManager() def run(self, user_input): # 第一轮意图解析 intent self.parser.parse(user_input) # 第二轮工具选择与执行 if intent[intent] weather_query: result self.tools.execute(get_weather, intent[params]) elif intent[intent] math_calculation: result self.tools.execute(calculate, intent[params]) else: result self.llm.generate(user_input) return result这个核心引擎展示了智能体的基本工作模式将自然语言转化为结构化意图根据意图路由到具体工具处理工具返回结果3.2 增强型决策流程基础版本只能处理单轮交互我们可以添加反思机制来增强能力def enhanced_run(self, user_input): context [] for _ in range(3): # 最多3轮交互 intent self.parser.parse(user_input, context) if intent[intent] final_answer: return intent[params][answer] tool_result self.tools.execute(intent[tool], intent[params]) context.append({ user: user_input, tool_used: intent[tool], result: tool_result }) # 让LLM判断是否需要继续 continue_decision self.llm.generate( f根据以下对话历史是否需要继续追问\n f{json.dumps(context, ensure_asciiFalse)}\n f只需回答yes或no ) if no in continue_decision.lower(): break return self.llm.generate(f综合以下信息回答问题\n{context})4. 实战优化技巧4.1 工具描述的魔法工具的描述文字直接影响LLM的选择准确率。好的描述应该明确说明工具的适用场景列举典型的输入输出示例指出与其他工具的区别例如天气查询工具可以这样描述获取实时天气数据。适用于用户询问某地当前或未来天气的场景。 输入示例{location: 北京, date: 2023-10-01} 输出示例{status: 晴, temp: 22℃} 区别于天气预报工具本工具只返回当前实时数据。4.2 参数校验策略在工具执行前添加参数校验层def validate_params(self, tool_name, params): schemas { get_weather: { location: {type: str, required: True}, date: {type: str, required: False} } } if tool_name not in schemas: return False for param, config in schemas[tool_name].items(): if config[required] and param not in params: return False if param in params and not isinstance(params[param], config[type]): return False return True4.3 性能优化技巧工具预热提前加载常用工具的资源结果缓存对相同参数的查询缓存结果批量处理合并多个工具调用请求class CachedToolManager(ToolManager): def __init__(self): super().__init__() self.cache {} def execute(self, tool_name, params): cache_key f{tool_name}_{json.dumps(params, sort_keysTrue)} if cache_key in self.cache: return self.cache[cache_key] result super().execute(tool_name, params) self.cache[cache_key] result return result5. 典型问题排查指南5.1 工具选择错误症状LLM总是选择错误的工具 排查步骤检查工具描述是否清晰明确验证意图解析的prompt是否合理测试不同表述方式对结果的影响解决方案示例# 在意图解析prompt中添加示例 prompt f分析用户意图示例 输入北京明天天气如何 输出{intent: weather_query, params: {location: 北京, date: 明天}} 当前输入{user_input} 5.2 参数提取不准症状工具执行时缺少必要参数 解决方案添加参数校验中间件实现参数回填机制def param_fallback(self, tool_name, params): required_params self.get_required_params(tool_name) for param in required_params: if param not in params: # 让LLM补充缺失参数 value self.llm.generate( f需要补充参数{param}的值用户原话{params.get(_original_input,)} ) params[param] value return params5.3 循环失控症状Agent陷入无限循环 防护措施设置最大循环次数添加超时中断实现循环检测def safe_run(self, user_input): start_time time.time() loop_count 0 while loop_count self.max_loops and time.time()-start_time self.timeout: loop_count 1 # ...原有逻辑... # 检查是否陷入循环 if len(context) 2 and context[-1][tool_used] context[-3][tool_used]: break6. 扩展方向与进阶思考这个基础框架可以沿多个方向扩展多工具组合实现工具间的数据传递class PipelineTool: def execute(self, tools_sequence, initial_input): data initial_input for tool in tools_sequence: data tool.execute(data) return data动态工具加载运行时添加新工具def hot_load_tool(self, tool_def): self.tools[tool_def[name]] { func: tool_def[func], desc: tool_def.get(desc, ) }可视化监控实时展示Agent决策过程class VisualAgent(AgentEngine): def run(self, user_input): self.visualize(开始处理输入, user_input) intent self.parser.parse(user_input) self.visualize(解析意图, intent) # ...其余步骤添加可视化点...在200行代码的约束下我们做出了多个折中设计使用简易的LLM客户端替代完整API省略了复杂的错误恢复机制采用内存存储而非数据库这些设计使得代码保持简洁的同时完整呈现了智能体的核心原理。要构建生产级Agent还需要考虑分布式执行、持久化存储、权限控制等企业级特性但那将是另一个故事了。