最近在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通常包含以下四个核心组件：

1. LLM（大语言模型）：Agent的"大脑"，负责思考、规划和决策
2. 工具集（Tools）：Agent可以调用的函数或API，如计算器、网络搜索、数据库查询等
3. 记忆系统（Memory）：存储对话历史、执行状态和中间结果
4. 控制循环（Control Loop）：协调各个组件的工作流程

2.3 为什么需要从零开始实现？

使用现成框架确实方便，但存在几个问题：

• 过度抽象：框架隐藏了关键细节，让你难以理解底层机制
• 灵活性不足：当需要定制特殊逻辑时，框架的限制会变得明显
• 调试困难：出现问题时，你需要在框架的复杂层级中寻找根源

从零实现让你能够：

• 完全控制Agent的决策逻辑
• 轻松添加自定义工具和验证逻辑
• 更好地理解性能瓶颈和优化方向

3. 环境准备与前置条件

在开始编码之前，我们需要准备开发环境。本文使用Python作为开发语言，因为它在AI领域有丰富的生态支持。

3.1 环境要求

• Python版本：3.8或更高版本
• 操作系统：Windows、macOS或Linux均可
• 必要的Python包：openai、requests、python-dotenv

3.2 安装依赖

创建并激活虚拟环境（推荐）：

# 创建虚拟环境 python -m venv agent_env # 激活虚拟环境（Windows） agent_env\Scripts\activate # 激活虚拟环境（macOS/Linux） source agent_env/bin/activate # 安装必要包 pip install openai requests python-dotenv

3.3 API密钥配置

由于我们的Agent需要调用LLM，你需要准备一个OpenAI API密钥。创建.env文件存储密钥：

# 创建.env文件 touch .env

在.env文件中添加：

OPENAI_API_KEY=你的API密钥

重要安全提醒：

• 永远不要将API密钥提交到版本控制系统
• 使用环境变量或配置文件管理敏感信息
• 为不同的项目使用不同的API密钥以便管理权限

4. 核心架构设计

在我们开始写代码之前，先来设计Agent的整体架构。这将帮助我们理解各个组件如何协同工作。

4.1 Agent系统架构

我们的Agent采用经典的ReAct（Reasoning + 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__( name="calculator", 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__( name="budget_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__( name="distance_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( model=self.model, messages=[{"role": "user", "content": prompt}], temperature=0.1 # 低温度保证稳定性 ) return response.choices[0].message.content.strip() except Exception as e: return f"LLM调用错误: {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"=== 第{iteration+1}次迭代 ===") 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_prompt

7.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 启动程序

运行我们的旅行规划Agent：

python main.py

8.2 测试用例

让我们测试几个典型的旅行规划场景：

测试1：预算估算

你的旅行需求: 我们3个人计划去北京玩5天，中等消费水平，请帮我估算一下预算

预期输出流程：

1. Agent思考：用户需要预算估算，使用budget_estimator工具
2. 调用工具：budget_estimator(days=5, people=3, cost_level="medium")
3. 返回结果：预算估算: 3人5天medium消费水平，总预算: 7500元

测试2：多步骤问题

你的旅行需求: 我想从北京去上海，然后去广州，请计算总距离和大概的旅行预算

预期输出流程：

1. 首先计算北京到上海的距离
2. 然后计算上海到广州的距离
3. 最后估算预算（可能需要询问具体天数）
4. 综合所有信息给出答案

8.3 验证Agent的思考过程

运行程序时，你会看到详细的思考过程：

=== 第1次迭代 === 思考: 用户需要计算从北京到上海再到广州的总距离和预算。我需要先计算各段距离，然后询问具体天数来估算预算。 行动: distance_calculator 工具结果: 北京到上海的距离约为1200公里 === 第2次迭代 === 思考: 已经得到北京到上海的距离，现在需要计算上海到广州的距离。 行动: distance_calculator 工具结果: 上海到广州的距离约为1300公里 === 第3次迭代 === 思考: 现在有了总距离(1200+1300=2500公里)，但需要知道旅行天数才能估算预算。 行动: final_answer 最终答案: 北京→上海→广州总距离约2500公里。请告诉我计划旅行多少天，我可以为您估算预算。

9. 常见问题与排查思路

在开发和使用Agent过程中，你可能会遇到以下常见问题：

9.1 LLM相关问题

9.2 工具执行问题

9.3 系统架构问题

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_length=10): """管理对话历史，保持相关上下文""" 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 history

11. 扩展性与自定义

我们的基础架构支持多种扩展方式：

11.1 添加新工具

class WeatherTool(Tool): """天气查询工具""" def __init__(self): super().__init__( name="weather", 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实现，你不仅掌握了构建智能体的核心技术原理，还具备了根据实际需求定制和扩展的能力。这种深度理解将帮助你在使用高级框架时做出更明智的架构决策，并在遇到复杂问题时能够进行有效调试和优化。

建议将本文代码作为学习基础，然后尝试添加更多工具、优化提示词策略，或者集成到实际项目中。真正的掌握来自于实践和迭代，现在你已经有了一套完整的起点。