30行Python代码实现AI Agent核心:从工具调用到任务规划
2026/7/25 20:04:00 网站建设 项目流程

最近在尝试各种AI Agent框架时,我发现了一个有趣的现象:很多开发者把Agent想得太复杂了。他们一上来就研究LangChain、AutoGPT这些重型框架,却忽略了Agent最核心的机制其实可以用几十行代码说清楚。

今天我就用30行Python代码,带你从零实现一个具备工具调用能力的AI Agent核心。这不是玩具代码,而是真正能工作的Agent骨架,理解了它,你就能看清所有复杂框架背后的共同逻辑。

1. 先搞清楚Agent到底在解决什么问题

很多人误以为Agent就是"能调用工具的AI",这种理解太表面了。Agent真正解决的是让AI从被动应答转向主动规划的问题。

想象一个场景:用户问"今天北京天气怎么样?适合出门吗?"。传统的AI会直接回答"我无法获取实时天气",然后对话就结束了。而Agent会这样思考:

  1. 识别需求:用户需要天气信息和出行建议
  2. 规划步骤:先获取天气数据,再分析是否适合出行
  3. 执行动作:调用天气API获取数据
  4. 整合结果:结合数据给出建议

这个"识别-规划-执行-整合"的循环,才是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 f"Tool {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 = f"Weather info: {result}. Based on this, it's 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: f"Sunny, 25°C in {city}" } ] agent = SimpleAgent(tools) print(agent.run("What's 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 response

run方法封装了完整的处理流程:记录用户输入、生成响应、记录AI回复。这是Agent的工作主循环。

2.2 从简化版到生产级的演进路径

这个30行版本虽然简单,但已经具备了Agent的核心架构。要让它达到生产级别,需要在四个方向上进行扩展:

  1. 智能决策:用AI模型替代简单的关键词匹配
  2. 错误处理:增加超时、重试、降级策略
  3. 状态管理:支持多轮工具调用和复杂任务分解
  4. 可观测性:添加日志、监控和调试信息

这些扩展都是在现有骨架上的自然演进,而不是推翻重来。

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": "What's 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 = f"I've checked: {', '.join(results)}. Based on this, here's 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: f"Weather 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_retries=3): 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 f"Error 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, timeout=30): 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 out"

5.3 可观测性与调试支持

添加详细的日志记录,方便问题排查:

def instrumented_run(self, user_input): logger.info(f"Processing user input: {user_input}") start_time = time.time() response = self.run(user_input) duration = time.time() - start_time logger.info(f"Response generated in {duration:.2f}s: {response}") return response

6. 从代码到架构的思维转变

理解了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: f"Order {order_id} status: shipped"}, {'name': 'get_policy', 'function': lambda topic: f"Policy on {topic}: ..."} ]

数据分析助手:查询数据库、生成报表、解释趋势代码助手:搜索文档、运行测试、调试代码

7.2 技术演进趋势

当前的Agent技术正在向以下几个方向发展:

多模态能力:支持图像、音频等非文本工具长期记忆:跨会话的状态保持和学习能力协作Agent:多个Agent协同完成复杂任务自适应学习:根据使用反馈优化工具使用策略

7.3 入门到精进的学习路径

基于这个30行代码的起点,我建议的学习路径是:

  1. 理解核心:彻底掌握本文的简单Agent实现
  2. 扩展功能:添加错误处理、状态管理等基础能力
  3. 集成AI:用真实的AI模型替代简单的决策逻辑
  4. 学习框架:研究LangChain、AutoGPT等成熟框架的设计思想
  5. 实战项目:构建有实际价值的Agent应用

这个30行代码的Agent实现,最重要的价值不是功能有多强大,而是它清晰地揭示了Agent技术的本质。无论后续学习多么复杂的框架,都要记得回归到这个核心模式:工具注册、上下文管理、决策规划、执行整合。

真正掌握Agent开发的关键,不在于记忆框架API,而在于理解这种"思考-行动-观察"的循环机制。当你能够用最简单的代码实现核心逻辑时,面对任何复杂框架都能快速抓住其设计精髓。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询