30行Python代码实现AI Agent核心：从工具调用到任务规划

最近在尝试各种AI Agent框架时我发现了一个有趣的现象很多开发者把Agent想得太复杂了。他们一上来就研究LangChain、AutoGPT这些重型框架却忽略了Agent最核心的机制其实可以用几十行代码说清楚。今天我就用30行Python代码带你从零实现一个具备工具调用能力的AI Agent核心。这不是玩具代码而是真正能工作的Agent骨架理解了它你就能看清所有复杂框架背后的共同逻辑。1. 先搞清楚Agent到底在解决什么问题很多人误以为Agent就是能调用工具的AI这种理解太表面了。Agent真正解决的是让AI从被动应答转向主动规划的问题。想象一个场景用户问今天北京天气怎么样适合出门吗。传统的AI会直接回答我无法获取实时天气然后对话就结束了。而Agent会这样思考识别需求用户需要天气信息和出行建议规划步骤先获取天气数据再分析是否适合出行执行动作调用天气API获取数据整合结果结合数据给出建议这个识别-规划-执行-整合的循环才是Agent的核心价值。它让AI不再受限于单次对话而是能够自主完成多步骤任务。1.1 为什么工具调用是Agent的基础能力工具调用Tool Use是Agent实现多步骤任务的关键桥梁。没有工具调用AI就只能停留在语言层面无法与外部世界交互。但工具调用本身并不复杂它本质上就是一个标准化的请求-响应模式。从技术角度看工具调用包含三个核心要素工具定义描述工具的功能、参数和返回值工具选择AI根据当前任务选择合适的工具工具执行调用工具并处理返回结果理解了这三点我们就能用最简洁的代码实现Agent的核心机制。2. 用30行代码构建Agent核心骨架下面这个实现虽然简单但包含了Agent最核心的组件。我们先看完整代码再逐部分解析import json class SimpleAgent: def __init__(self, tools): self.tools {tool[name]: tool for tool in tools} self.conversation [] def add_message(self, role, content): self.conversation.append({role: role, content: content}) def process_tool_call(self, tool_name, arguments): tool self.tools.get(tool_name) if not tool: return fTool {tool_name} not found return tool[function](**arguments) def run(self, user_input): self.add_message(user, user_input) # 这里应该是AI模型判断是否需要调用工具 # 简化版直接检查关键词 if weather in user_input.lower(): result self.process_tool_call(get_weather, {city: Beijing}) response fWeather info: {result}. Based on this, its a good day to go out. else: response I can help with weather queries. Ask me about weather! self.add_message(assistant, response) return response # 定义工具 tools [ { name: get_weather, function: lambda city: fSunny, 25°C in {city} } ] agent SimpleAgent(tools) print(agent.run(Whats the weather in Beijing?))2.1 代码结构解析四层核心逻辑这个简单的Agent包含了四个关键组件第一层工具注册与管理def __init__(self, tools): self.tools {tool[name]: tool for tool in tools}工具管理是Agent的基础。我们用一个字典来存储所有可用工具通过工具名快速查找。在实际的复杂Agent中工具描述会更详细包括参数说明、返回类型等。第二层对话上下文维护def add_message(self, role, content): self.conversation.append({role: role, content: content})对话历史让Agent具备连续思考能力。每次交互都会记录到conversation列表中为后续的决策提供上下文。第三层工具调用机制def process_tool_call(self, tool_name, arguments): tool self.tools.get(tool_name) if tool: return tool[function](**arguments)这是最核心的工具执行逻辑。它接收工具名和参数找到对应的工具函数并执行。错误处理虽然简单但体现了基本的健壮性。第四层任务处理流程def run(self, user_input): self.add_message(user, user_input) # 决策逻辑 response self.generate_response(user_input) self.add_message(assistant, response) return responserun方法封装了完整的处理流程记录用户输入、生成响应、记录AI回复。这是Agent的工作主循环。2.2 从简化版到生产级的演进路径这个30行版本虽然简单但已经具备了Agent的核心架构。要让它达到生产级别需要在四个方向上进行扩展智能决策用AI模型替代简单的关键词匹配错误处理增加超时、重试、降级策略状态管理支持多轮工具调用和复杂任务分解可观测性添加日志、监控和调试信息这些扩展都是在现有骨架上的自然演进而不是推翻重来。3. 理解工具调用的标准化协议在实际的AI开发中工具调用遵循着一定的标准化模式。了解这些标准有助于我们理解复杂框架的设计思路。3.1 工具描述规范完整的工具描述通常包含以下信息{ name: get_weather, description: Get current weather information for a city, parameters: { type: object, properties: { city: { type: string, description: City name } }, required: [city] } }这种结构化的描述让AI模型能够理解工具的功能和使用方法从而做出正确的调用决策。3.2 消息流模式标准的Agent消息流包含几种类型的消息块user: 用户输入assistant: AI回复tool_use: 工具调用请求tool_result: 工具调用结果这种模式确保了对话上下文的完整性和工具调用的可追溯性。3.3 实际框架中的实现对比以Claude的Tool Use为例它的消息结构是这样的messages [ {role: user, content: Whats the weather in Beijing?}, {role: assistant, content: , tool_calls: [ {id: call_1, type: function, function: { name: get_weather, arguments: {city: Beijing} }} ]}, {role: tool, content: Sunny, 25°C, tool_call_id: call_1} ]可以看到虽然实现细节不同但核心模式与我们简单版是一致的请求-调用-返回。4. 从单次调用到多步规划的实现路径真正的Agent能力体现在多步骤任务规划上。让我们扩展简单Agent支持连续的工具调用。4.1 添加任务规划能力class PlanningAgent(SimpleAgent): def plan_steps(self, user_input): 简单的任务分解逻辑 if weather and suggest in user_input.lower(): return [ {action: get_weather, args: {city: Beijing}}, {action: analyze_activity, args: {}} ] return [] def run(self, user_input): self.add_message(user, user_input) steps self.plan_steps(user_input) results [] for step in steps: result self.process_tool_call(step[action], step[args]) results.append(result) # 整合所有步骤的结果 if steps: response fIve checked: {, .join(results)}. Based on this, heres my suggestion. else: response I can help with multi-step tasks like weather analysis. self.add_message(assistant, response) return response # 扩展工具集 tools [ { name: get_weather, function: lambda city: fWeather in {city}: Sunny, 25°C }, { name: analyze_activity, function: lambda: Perfect weather for outdoor activities } ] agent PlanningAgent(tools) print(agent.run(Check Beijing weather and suggest activities))4.2 规划逻辑的复杂度分层在实际开发中任务规划可以按复杂度分为三个层次第一层固定模式规划像上面的例子基于关键词匹配固定步骤序列。适合简单、 predictable的任务。第二层规则引擎规划定义一系列规则来自动生成步骤序列rules [ { pattern: weather.*suggest, steps: [get_weather, analyze_activity] } ]第三层AI模型规划使用专门的规划模型或提示工程让AI自主生成步骤序列。这是最灵活但也最复杂的方式。4.3 状态管理与上下文传递多步骤任务需要维护任务状态和传递中间结果class StatefulAgent(PlanningAgent): def __init__(self, tools): super().__init__(tools) self.current_state {} def run(self, user_input): steps self.plan_steps(user_input) for step in steps: # 将之前步骤的结果作为当前步骤的上下文 step_args {**step[args], **self.current_state} result self.process_tool_call(step[action], step_args) self.current_state[step[action] _result] result # 清理状态准备下一个任务 final_response self.generate_final_response() self.current_state {} return final_response这种状态管理机制确保了步骤间的数据传递为复杂任务提供了基础支持。5. 生产环境中的关键考量当我们把简单Agent扩展到生产环境时需要解决一系列工程化问题。5.1 错误处理与重试机制生产级Agent需要完善的错误处理def robust_tool_call(self, tool_name, arguments, max_retries3): for attempt in range(max_retries): try: result self.process_tool_call(tool_name, arguments) return result except Exception as e: if attempt max_retries - 1: return fError after {max_retries} attempts: {str(e)} # 等待后重试 time.sleep(2 ** attempt)5.2 超时控制与资源管理工具调用必须有超时控制避免长时间阻塞import signal def timeout_handler(signum, frame): raise TimeoutError(Tool call timeout) def safe_tool_call(self, tool_name, arguments, timeout30): signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(timeout) try: result self.process_tool_call(tool_name, arguments) signal.alarm(0) # 取消定时器 return result except TimeoutError: return Tool call timed out5.3 可观测性与调试支持添加详细的日志记录方便问题排查def instrumented_run(self, user_input): logger.info(fProcessing user input: {user_input}) start_time time.time() response self.run(user_input) duration time.time() - start_time logger.info(fResponse generated in {duration:.2f}s: {response}) return response6. 从代码到架构的思维转变理解了Agent的核心机制后我们需要从代码实现转向架构设计思维。6.1 Agent系统的组件化设计一个完整的Agent系统通常包含以下组件组件职责实现复杂度对话管理器维护对话历史和上下文低工具管理器注册、发现和执行工具中规划器分解任务和生成步骤序列高执行引擎协调工具调用和状态管理中结果整合器合成最终响应低我们的30行代码已经涵盖了前两个组件为后续扩展奠定了基础。6.2 性能与扩展性考量在设计Agent系统时需要考虑并发处理支持多个用户同时使用from threading import Lock class ConcurrentAgent(SimpleAgent): def __init__(self, tools): super().__init__(tools) self.lock Lock() def thread_safe_run(self, user_input): with self.lock: return self.run(user_input)工具发现机制动态加载和注册工具缓存策略对频繁使用的工具结果进行缓存限流保护防止工具被过度调用6.3 测试策略与质量保证Agent系统的测试需要覆盖多个层面def test_agent_integration(): 端到端集成测试 agent SimpleAgent(tools) response agent.run(test query) assert weather in response.lower() def test_tool_isolation(): 工具隔离测试 # 测试一个工具的失败不影响其他工具 pass def test_performance(): 性能基准测试 start_time time.time() # 执行多次调用 assert time.time() - start_time 1.0 # 性能要求7. 实际应用场景与演进方向掌握了Agent核心机制后我们可以将其应用到各种实际场景中。7.1 常见应用模式客服助手查询订单、解答政策、转接人工customer_tools [ {name: check_order, function: lambda order_id: fOrder {order_id} status: shipped}, {name: get_policy, function: lambda topic: fPolicy on {topic}: ...} ]数据分析助手查询数据库、生成报表、解释趋势代码助手搜索文档、运行测试、调试代码7.2 技术演进趋势当前的Agent技术正在向以下几个方向发展多模态能力支持图像、音频等非文本工具长期记忆跨会话的状态保持和学习能力协作Agent多个Agent协同完成复杂任务自适应学习根据使用反馈优化工具使用策略7.3 入门到精进的学习路径基于这个30行代码的起点我建议的学习路径是理解核心彻底掌握本文的简单Agent实现扩展功能添加错误处理、状态管理等基础能力集成AI用真实的AI模型替代简单的决策逻辑学习框架研究LangChain、AutoGPT等成熟框架的设计思想实战项目构建有实际价值的Agent应用这个30行代码的Agent实现最重要的价值不是功能有多强大而是它清晰地揭示了Agent技术的本质。无论后续学习多么复杂的框架都要记得回归到这个核心模式工具注册、上下文管理、决策规划、执行整合。真正掌握Agent开发的关键不在于记忆框架API而在于理解这种思考-行动-观察的循环机制。当你能够用最简单的代码实现核心逻辑时面对任何复杂框架都能快速抓住其设计精髓。