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_ascii=False)}\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_keys=True)}" 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,还需要考虑分布式执行、持久化存储、权限控制等企业级特性,但那将是另一个故事了。