从零构建AI Agent：200行Python代码实现智能体思考回路

最近在AI圈里Agent这个词的热度持续攀升。各种Agent框架层出不穷从LangChain到AutoGPT每个都宣称能帮你快速构建智能体。但当你真正想理解Agent到底是怎么工作时这些框架反而成了黑箱——它们封装了太多细节让你很难看清底层的运行逻辑。如果你也有这样的困惑Agent到底是如何思考、规划和执行任务的为什么有些Agent能处理复杂问题而有些只能做简单问答那么这篇文章就是为你准备的。我们将抛开所有现成框架从零开始构建一个真正的Agent让你真正掌握智能体的核心原理。通过本文你将学会如何用不到200行Python代码实现一个具备思考能力的Agent理解LLM调用、工具执行、状态管理的完整流程并能够根据实际需求定制自己的智能体。1. 这篇文章真正要解决的问题很多开发者学习Agent时容易陷入一个误区过早依赖框架而忽略了底层原理。这导致当遇到复杂业务需求时不知道如何调整框架的默认行为或者无法诊断Agent为什么犯傻。本文要解决的核心问题是理解Agent的思考回路reasoning loop是如何工作的。这个回路包括如何让LLM分析当前状态并决定下一步行动如何将自然语言指令转换为具体的函数调用如何管理执行过程中的状态和上下文如何处理失败和异常情况我们将通过构建一个旅行规划Agent来演示这个过程。这个Agent能够帮用户规划旅行路线、查询天气、估算预算等虽然功能简单但包含了完整Agent的所有核心组件。2. Agent基础概念与核心原理2.1 什么是Agent在AI语境中Agent不是指代理服务器而是一个能够感知环境、做出决策并执行行动的智能体。与传统程序不同Agent的核心特点是具备自主性和适应性。自主性意味着Agent能够基于目标自主决定要做什么而不需要每一步都由人类明确指令。适应性体现在Agent能够根据环境反馈调整策略比如当某个方法失败时会尝试替代方案。2.2 Agent的核心组件一个完整的Agent通常包含以下四个核心组件LLM大语言模型Agent的大脑负责思考、规划和决策工具集ToolsAgent可以调用的函数或API如计算器、网络搜索、数据库查询等记忆系统Memory存储对话历史、执行状态和中间结果控制循环Control Loop协调各个组件的工作流程2.3 为什么需要从零开始实现使用现成框架确实方便但存在几个问题过度抽象框架隐藏了关键细节让你难以理解底层机制灵活性不足当需要定制特殊逻辑时框架的限制会变得明显调试困难出现问题时你需要在框架的复杂层级中寻找根源从零实现让你能够完全控制Agent的决策逻辑轻松添加自定义工具和验证逻辑更好地理解性能瓶颈和优化方向3. 环境准备与前置条件在开始编码之前我们需要准备开发环境。本文使用Python作为开发语言因为它在AI领域有丰富的生态支持。3.1 环境要求Python版本3.8或更高版本操作系统Windows、macOS或Linux均可必要的Python包openai、requests、python-dotenv3.2 安装依赖创建并激活虚拟环境推荐# 创建虚拟环境 python -m venv agent_env # 激活虚拟环境Windows agent_env\Scripts\activate # 激活虚拟环境macOS/Linux source agent_env/bin/activate # 安装必要包 pip install openai requests python-dotenv3.3 API密钥配置由于我们的Agent需要调用LLM你需要准备一个OpenAI API密钥。创建.env文件存储密钥# 创建.env文件 touch .env在.env文件中添加OPENAI_API_KEY你的API密钥重要安全提醒永远不要将API密钥提交到版本控制系统使用环境变量或配置文件管理敏感信息为不同的项目使用不同的API密钥以便管理权限4. 核心架构设计在我们开始写代码之前先来设计Agent的整体架构。这将帮助我们理解各个组件如何协同工作。4.1 Agent系统架构我们的Agent采用经典的ReActReasoning Acting架构用户输入 → Agent思考 → 选择工具 → 执行工具 → 观察结果 → 继续思考/返回结果这个循环会持续进行直到Agent认为任务已经完成或达到最大迭代次数。4.2 核心类设计我们将创建三个核心类Tool类定义工具的基本接口Agent类核心的智能体包含思考逻辑TravelPlanner类具体的旅行规划Agent实现5. 工具Tool实现工具是Agent能够调用的具体功能。我们先实现几个基础工具然后定义工具的管理机制。5.1 基础工具类# tool.py from abc import ABC, abstractmethod from typing import Dict, Any class Tool(ABC): 工具基类所有工具都需要继承这个类 def __init__(self, name: str, description: str): self.name name self.description description abstractmethod def execute(self, **kwargs) - str: 执行工具的具体逻辑 pass def __str__(self) - str: return f{self.name}: {self.description}5.2 具体工具实现让我们实现几个旅行规划中常用的工具# tools.py import math from typing import Dict, Any class CalculatorTool(Tool): 计算器工具用于数学计算 def __init__(self): super().__init__( namecalculator, description执行数学计算支持加减乘除、指数等运算 ) def execute(self, expression: str) - str: try: # 安全评估数学表达式 result eval(expression, {__builtins__: None}, { abs: abs, round: round, min: min, max: max, pow: pow, sqrt: math.sqrt }) return f计算结果: {expression} {result} except Exception as e: return f计算错误: {str(e)} class BudgetEstimatorTool(Tool): 旅行预算估算工具 def __init__(self): super().__init__( namebudget_estimator, description根据天数、人数、消费水平估算旅行预算 ) def execute(self, days: int, people: int, cost_level: str medium) - str: # 定义不同消费水平的每日人均费用单位元 cost_levels { low: 300, medium: 500, high: 800 } if cost_level not in cost_levels: return f错误的消费水平: {cost_level}可选: low, medium, high daily_cost cost_levels[cost_level] total_cost daily_cost * days * people return (f预算估算: {people}人{days}天{cost_level}消费水平\n f总预算: {total_cost}元) class DistanceCalculatorTool(Tool): 简单距离计算工具模拟 def __init__(self): super().__init__( namedistance_calculator, description计算两个城市之间的距离模拟数据 ) def execute(self, city1: str, city2: str) - str: # 模拟城市间距离数据 distances { (北京, 上海): 1200, (北京, 广州): 1900, (上海, 广州): 1300, (北京, 深圳): 1950, (上海, 深圳): 1350, } key (city1, city2) reverse_key (city2, city1) if key in distances: distance distances[key] elif reverse_key in distances: distance distances[reverse_key] else: return f未找到{city1}到{city2}的距离数据 return f{city1}到{city2}的距离约为{distance}公里5.3 工具管理器# tool_manager.py from typing import Dict, List from tool import Tool class ToolManager: 管理所有可用工具 def __init__(self): self.tools: Dict[str, Tool] {} def register_tool(self, tool: Tool) - None: 注册工具 self.tools[tool.name] tool def get_tool(self, name: str) - Tool: 获取指定工具 if name not in self.tools: raise ValueError(f工具不存在: {name}) return self.tools[name] def list_tools(self) - List[Tool]: 列出所有可用工具 return list(self.tools.values()) def get_tools_description(self) - str: 获取所有工具的详细描述用于LLM提示词 descriptions [] for tool in self.tools.values(): descriptions.append(f- {tool.name}: {tool.description}) return \n.join(descriptions)6. Agent核心实现现在我们来实现最核心的Agent类这是整个系统的大脑。6.1 Agent基类实现# agent.py import json import openai from typing import Dict, List, Any from tool_manager import ToolManager class Agent: Agent基类 def __init__(self, model: str gpt-3.5-turbo): self.model model self.tool_manager ToolManager() self.conversation_history: List[Dict[str, str]] [] self.max_iterations 5 # 最大思考迭代次数 def add_tool(self, tool) - None: 添加工具 self.tool_manager.register_tool(tool) def _call_llm(self, prompt: str) - str: 调用LLM获取响应 try: response openai.ChatCompletion.create( modelself.model, messages[{role: user, content: prompt}], temperature0.1 # 低温度保证稳定性 ) return response.choices[0].message.content.strip() except Exception as e: return fLLM调用错误: {str(e)} def _parse_action(self, text: str) - Dict[str, Any]: 解析LLM返回的行动指令 try: # 尝试解析JSON格式的响应 if json in text: json_str text.split(json)[1].split()[0].strip() elif { in text and } in text: json_str text[text.find({):text.rfind(})1] else: # 如果不是JSON格式返回思考结果 return {action: final_answer, answer: text} action_data json.loads(json_str) return action_data except json.JSONDecodeError: return {action: final_answer, answer: text} def _build_system_prompt(self) - str: 构建系统提示词 tools_description self.tool_manager.get_tools_description() return f你是一个智能助手可以调用工具来帮助用户解决问题。 可用工具 {tools_description} 请按照以下格式思考 1. 分析用户的问题和目标 2. 决定是否需要使用工具以及使用哪个工具 3. 如果使用工具提供正确的参数 4. 根据工具结果继续思考或给出最终答案 响应格式 json {{ thought: 你的思考过程, action: 工具名称 或 final_answer, action_input: {{工具参数}} 或 最终答案 }}重要规则只有在必要时才使用工具确保工具参数正确如果任务完成使用final_answer返回结果保持思考逻辑清晰def run(self, user_input: str) - str: 运行Agent处理用户输入 self.conversation_history.append({role: user, content: user_input})for iteration in range(self.max_iterations): # 构建完整的对话上下文 context self._build_system_prompt() \n\n对话历史:\n for msg in self.conversation_history[-3:]: # 最近3条消息 context f{msg[role]}: {msg[content]}\n context f\n当前问题: {user_input}\n请分析并响应: # 调用LLM获取响应 response self._call_llm(context) # 解析行动指令 action_data self._parse_action(response) print(f 第{iteration1}次迭代 ) print(f思考: {action_data.get(thought, 无)}) print(f行动: {action_data.get(action, 无)}) # 执行行动 if action_data.get(action) final_answer: final_answer action_data.get(action_input, action_data.get(answer, 无法提供答案)) self.conversation_history.append({role: assistant, content: final_answer}) return final_answer elif action_data.get(action) in self.tool_manager.tools: # 执行工具 tool_name action_data[action] tool_input action_data.get(action_input, {}) try: tool self.tool_manager.get_tool(tool_name) result tool.execute(**tool_input) print(f工具结果: {result}) # 将工具结果添加到对话历史 self.conversation_history.append({ role: assistant, content: f使用了工具 {tool_name}结果: {result} }) # 如果这是最后一次迭代直接返回结果 if iteration self.max_iterations - 1: return f经过{self.max_iterations}次尝试最终结果: {result} except Exception as e: error_msg f工具执行错误: {str(e)} self.conversation_history.append({ role: assistant, content: error_msg }) return error_msg else: # 未知行动返回思考内容 return action_data.get(thought, 无法处理该请求) return f达到最大迭代次数({self.max_iterations})未能完成请求## 7. 旅行规划Agent实现 现在我们将基础的Agent类特化为旅行规划Agent。 ### 7.1 specialized Agent实现 python # travel_agent.py from agent import Agent from tools import CalculatorTool, BudgetEstimatorTool, DistanceCalculatorTool class TravelPlannerAgent(Agent): 专门的旅行规划Agent def __init__(self): super().__init__() self._setup_tools() self._enhance_system_prompt() def _setup_tools(self): 设置旅行规划专用工具 self.add_tool(CalculatorTool()) self.add_tool(BudgetEstimatorTool()) self.add_tool(DistanceCalculatorTool()) def _enhance_system_prompt(self): 增强系统提示词加入旅行领域知识 # 我们将在run方法中动态构建提示词 pass def run(self, user_input: str) - str: 重写run方法加入旅行领域特定的逻辑 # 在系统提示词中加入旅行规划指导 original_system_prompt self._build_system_prompt() travel_guidance 旅行规划专项指导 - 预算估算时考虑交通、住宿、餐饮、门票等费用 - 距离计算时提醒用户实际交通时间可能因交通方式而异 - 多城市行程要合理安排顺序以节省时间和费用 - 考虑季节和天气对旅行体验的影响 # 临时替换系统提示词 original_prompt self._build_system_prompt self._build_system_prompt lambda: original_system_prompt travel_guidance try: result super().run(user_input) return result finally: # 恢复原始方法 self._build_system_prompt original_prompt7.2 主程序入口# main.py import os from dotenv import load_dotenv from travel_agent import TravelPlannerAgent def main(): # 加载环境变量 load_dotenv() # 检查API密钥 if not os.getenv(OPENAI_API_KEY): print(错误: 请设置OPENAI_API_KEY环境变量) return # 创建Agent实例 agent TravelPlannerAgent() print( 旅行规划Agent已启动 ) print(输入quit退出程序) print(- * 40) while True: try: user_input input(\n你的旅行需求: ).strip() if user_input.lower() in [quit, exit, 退出]: print(感谢使用旅行规划Agent) break if not user_input: continue print(\n *50) result agent.run(user_input) print(f\n最终答案: {result}) print(*50) except KeyboardInterrupt: print(\n\n程序被用户中断) break except Exception as e: print(f程序错误: {str(e)}) if __name__ __main__: main()8. 运行示例与效果验证8.1 启动程序运行我们的旅行规划Agentpython main.py8.2 测试用例让我们测试几个典型的旅行规划场景测试1预算估算你的旅行需求: 我们3个人计划去北京玩5天中等消费水平请帮我估算一下预算预期输出流程Agent思考用户需要预算估算使用budget_estimator工具调用工具budget_estimator(days5, people3, cost_levelmedium)返回结果预算估算: 3人5天medium消费水平总预算: 7500元测试2多步骤问题你的旅行需求: 我想从北京去上海然后去广州请计算总距离和大概的旅行预算预期输出流程首先计算北京到上海的距离然后计算上海到广州的距离最后估算预算可能需要询问具体天数综合所有信息给出答案8.3 验证Agent的思考过程运行程序时你会看到详细的思考过程 第1次迭代 思考: 用户需要计算从北京到上海再到广州的总距离和预算。我需要先计算各段距离然后询问具体天数来估算预算。 行动: distance_calculator 工具结果: 北京到上海的距离约为1200公里 第2次迭代 思考: 已经得到北京到上海的距离现在需要计算上海到广州的距离。 行动: distance_calculator 工具结果: 上海到广州的距离约为1300公里 第3次迭代 思考: 现在有了总距离(120013002500公里)但需要知道旅行天数才能估算预算。 行动: final_answer 最终答案: 北京→上海→广州总距离约2500公里。请告诉我计划旅行多少天我可以为您估算预算。9. 常见问题与排查思路在开发和使用Agent过程中你可能会遇到以下常见问题9.1 LLM相关问题问题现象可能原因排查方式解决方案LLM返回格式错误提示词不够清晰检查系统提示词中的格式要求加强格式约束提供更明确的示例LLM不调用工具工具描述不清晰检查工具的描述是否准确改进工具描述强调工具的能力响应内容无关temperature设置过高检查temperature参数降低temperature值如0.19.2 工具执行问题问题现象可能原因排查方式解决方案工具参数错误LLM不理解参数格式打印LLM的原始响应在提示词中提供参数示例工具执行异常参数类型不匹配检查工具函数的参数类型添加参数验证和类型转换工具返回结果无法理解结果格式复杂检查工具返回的数据结构简化工具返回格式9.3 系统架构问题问题现象可能原因排查方式解决方案无限循环停止条件不明确检查迭代逻辑和停止条件添加最大迭代次数限制内存泄漏对话历史无限增长检查conversation_history管理限制历史记录长度性能低下LLM调用频繁分析思考迭代次数优化提示词减少不必要的迭代10. 性能优化与最佳实践基于我们的实现经验以下是构建高效Agent的最佳实践10.1 提示词工程优化清晰的工具描述# 好的工具描述 description计算两个城市间的距离参数: city1(字符串), city2(字符串) # 差的工具描述 description计算距离明确的格式要求在系统提示词中提供完整的JSON格式示例强调必须遵守指定格式提供错误格式的示例和正确格式的对比10.2 错误处理与容错多层错误处理def safe_tool_execution(tool_name, parameters): try: # 参数验证 validated_params validate_parameters(parameters) # 工具执行 result execute_tool(tool_name, validated_params) return result except ValidationError as e: return f参数错误: {str(e)} except ToolExecutionError as e: return f工具执行失败: {str(e)} except Exception as e: return f未知错误: {str(e)}10.3 状态管理优化智能历史管理def manage_conversation_history(history, max_length10): 管理对话历史保持相关上下文 if len(history) max_length: # 保留系统提示词和最近对话 important_messages [h for h in history if h.get(important, False)] recent_messages history[-max_length//2:] return important_messages recent_messages return history11. 扩展性与自定义我们的基础架构支持多种扩展方式11.1 添加新工具class WeatherTool(Tool): 天气查询工具 def __init__(self): super().__init__( nameweather, description查询城市天气情况参数: city(字符串) ) def execute(self, city: str) - str: # 实现天气API调用 return f{city}天气: 晴25°C # 注册新工具 agent.add_tool(WeatherTool())11.2 支持多模态class ImageAnalysisTool(Tool): 图像分析工具 def execute(self, image_path: str) - str: # 调用视觉模型分析图片 return 图片分析结果: 包含山脉和湖泊11.3 记忆系统增强class EnhancedMemory: 增强记忆系统支持长期记忆和关键信息提取 def __init__(self): self.long_term_memory {} self.conversation_history [] def save_important_info(self, key, value): 保存重要信息到长期记忆 self.long_term_memory[key] value def get_relevant_memories(self, query): 根据查询检索相关记忆 # 实现简单的相关性匹配 relevant [] for key, value in self.long_term_memory.items(): if query.lower() in key.lower(): relevant.append(value) return relevant通过这个从零开始的Agent实现你不仅掌握了构建智能体的核心技术原理还具备了根据实际需求定制和扩展的能力。这种深度理解将帮助你在使用高级框架时做出更明智的架构决策并在遇到复杂问题时能够进行有效调试和优化。建议将本文代码作为学习基础然后尝试添加更多工具、优化提示词策略或者集成到实际项目中。真正的掌握来自于实践和迭代现在你已经有了一套完整的起点。