在AI技术快速发展的今天Agent智能体已成为连接大语言模型与现实应用的重要桥梁。很多开发者一提到Agent开发就想到使用现成框架但框架往往隐藏了底层原理让初学者难以真正理解Agent的工作机制。本文将从零开始手把手带你用纯Python实现一个功能完整的Agent系统涵盖思维链、工具调用、记忆机制等核心概念让你不仅会用Agent更懂Agent的魂。1. Agent核心概念与价值1.1 什么是AgentAgent本质上是一个能够感知环境、进行决策并执行动作的智能系统。在AI语境下Agent通常由大语言模型驱动具备以下核心能力自主性无需人工干预即可完成任务反应性能够感知环境变化并做出响应目标导向有明确的完成任务目标持续性在长时间跨度内保持运作与简单的聊天机器人不同真正的Agent具备规划能力和工具使用能力能够将复杂任务分解为可执行的子步骤。1.2 为什么需要从零开始实现虽然市面上有LangChain、AutoGPT等成熟框架但从零实现有以下优势深入理解原理摆脱框架的黑箱掌握每个组件的设计思路定制化能力强根据具体需求灵活调整架构不受框架限制部署轻量避免引入不必要的依赖适合资源受限场景学习价值高通过亲手实现建立对Agent技术的直觉理解1.3 Agent的核心组件一个完整的Agent系统通常包含以下关键组件推理引擎核心的决策模块通常基于LLM工具系统Agent可调用的外部功能集合记忆机制短期和长期记忆存储规划模块任务分解和步骤规划执行监控动作执行和状态跟踪2. 环境准备与基础架构2.1 开发环境配置首先确保你的Python环境版本在3.8以上这是大多数现代AI库的最低要求。我们将使用以下核心库# 创建虚拟环境 python -m venv agent_env source agent_env/bin/activate # Linux/Mac # 或 agent_env\Scripts\activate # Windows # 安装必要依赖 pip install openai pip install requests pip install python-dotenv如果你没有OpenAI API密钥可以使用开源的本地模型如Ollama搭配Llama 3等模型。2.2 项目结构设计在开始编码前先规划好项目结构agent_project/ ├── core/ │ ├── __init__.py │ ├── agent.py # Agent主类 │ ├── memory.py # 记忆管理 │ └── planner.py # 规划模块 ├── tools/ │ ├── __init__.py │ ├── base_tool.py # 工具基类 │ └── calculator.py # 示例工具 ├── config/ │ └── settings.py # 配置文件 ├── utils/ │ └── helpers.py # 工具函数 └── main.py # 主程序入口这种模块化设计便于后续扩展和维护每个组件职责清晰。3. 核心组件实现3.1 工具系统基础实现工具是Agent与外部世界交互的桥梁。我们先定义工具基类# tools/base_tool.py from abc import ABC, abstractmethod from typing import Dict, Any class BaseTool(ABC): 工具基类所有具体工具都需要继承此类 def __init__(self, name: str, description: str): self.name name self.description description abstractmethod def execute(self, **kwargs) - str: 执行工具的核心方法 pass def get_schema(self) - Dict[str, Any]: 返回工具的调用schema用于LLM理解参数 return { name: self.name, description: self.description, parameters: self._get_parameters_schema() } abstractmethod def _get_parameters_schema(self) - Dict[str, Any]: 定义工具参数的结构 pass接下来实现一个具体的计算器工具# tools/calculator.py import math from typing import Dict, Any from .base_tool import BaseTool class CalculatorTool(BaseTool): 简单的数学计算工具 def __init__(self): super().__init__( namecalculator, description执行基本的数学运算支持加减乘除、幂运算等 ) def execute(self, operation: str, expression: str) - str: 执行数学计算 try: # 安全评估数学表达式 allowed_names {**math.__dict__, abs: abs} result eval(expression, {__builtins__: {}}, allowed_names) return f计算结果: {result} except Exception as e: return f计算错误: {str(e)} def _get_parameters_schema(self) - Dict[str, Any]: return { type: object, properties: { operation: { type: string, description: 操作类型, enum: [calculate] }, expression: { type: string, description: 数学表达式如 2 3 * 4 } }, required: [operation, expression] }3.2 记忆管理系统记忆是Agent保持连续性的关键。我们实现一个简单的记忆系统# core/memory.py from typing import List, Dict, Any from datetime import datetime class Memory: 管理Agent的记忆 def __init__(self, max_messages: int 100): self.max_messages max_messages self.messages: List[Dict[str, Any]] [] def add_message(self, role: str, content: str, metadata: Dict None): 添加消息到记忆 message { role: role, content: content, timestamp: datetime.now().isoformat(), metadata: metadata or {} } self.messages.append(message) # 保持消息数量在限制内 if len(self.messages) self.max_messages: self.messages self.messages[-self.max_messages:] def get_recent_messages(self, count: int 10) - List[Dict[str, Any]]: 获取最近的消息 return self.messages[-count:] if self.messages else [] def get_conversation_history(self) - str: 将对话历史格式化为字符串 history [] for msg in self.messages: history.append(f{msg[role]}: {msg[content]}) return \n.join(history) def clear(self): 清空记忆 self.messages.clear()3.3 规划与推理引擎这是Agent的大脑负责决策和任务分解# core/planner.py import json import re from typing import List, Dict, Any class Planner: 任务规划和推理引擎 def __init__(self, llm_client): self.llm_client llm_client self.thinking_pattern re.compile(r思考过程:\s*(.*?)\s*行动:, re.DOTALL) def plan(self, user_input: str, available_tools: List[Dict]) - Dict[str, Any]: 根据用户输入制定行动计划 # 构建系统提示词 system_prompt self._build_system_prompt(available_tools) # 调用LLM进行规划 response self.llm_client.chat_completion( messages[ {role: system, content: system_prompt}, {role: user, content: user_input} ] ) return self._parse_llm_response(response) def _build_system_prompt(self, tools: List[Dict]) - str: 构建系统提示词 tools_description \n.join([ f- {tool[name]}: {tool[description]} for tool in tools ]) return f你是一个智能助手能够使用工具解决问题。请按照以下格式响应 可用工具: {tools_description} 响应格式: 思考过程: 你的推理过程 行动: 要执行的动作JSON格式 最终答案: 直接给用户的答案 行动格式示例: {{action: tool_name, parameters: {{param1: value1}}}} 如果不需要使用工具直接在最终答案中回复。 def _parse_llm_response(self, response: str) - Dict[str, Any]: 解析LLM的响应 result { thought_process: , action: None, final_answer: } # 提取思考过程 thought_match self.thinking_pattern.search(response) if thought_match: result[thought_process] thought_match.group(1).strip() # 提取行动指令 action_match re.search(r行动:\s*(\{.*?\}), response, re.DOTALL) if action_match: try: result[action] json.loads(action_match.group(1)) except json.JSONDecodeError: result[action] None # 提取最终答案 answer_match re.search(r最终答案:\s*(.*), response, re.DOTALL) if answer_match: result[final_answer] answer_match.group(1).strip() return result4. 完整Agent实现4.1 Agent主类设计现在我们将各个组件组合成完整的Agent# core/agent.py import json from typing import List, Dict, Any from .memory import Memory from .planner import Planner class SimpleAgent: 简单的Agent实现 def __init__(self, llm_client, tools: List None): self.llm_client llm_client self.memory Memory() self.planner Planner(llm_client) self.tools {tool.name: tool for tool in (tools or [])} def add_tool(self, tool): 添加工具 self.tools[tool.name] tool def process_message(self, user_input: str) - str: 处理用户输入并返回响应 # 添加到记忆 self.memory.add_message(user, user_input) # 获取可用工具信息 available_tools [tool.get_schema() for tool in self.tools.values()] # 进行规划 plan self.planner.plan(user_input, available_tools) # 执行行动 if plan[action]: result self._execute_action(plan[action]) # 将行动结果添加到记忆 self.memory.add_message(system, f执行动作: {plan[action]} - 结果: {result}) # 如果需要基于行动结果生成最终响应 if not plan[final_answer]: final_response self._generate_final_response(user_input, result, plan) else: final_response plan[final_answer] else: final_response plan[final_answer] # 添加助手响应到记忆 self.memory.add_message(assistant, final_response, {thought_process: plan[thought_process]}) return final_response def _execute_action(self, action: Dict[str, Any]) - str: 执行具体的工具调用 tool_name action.get(action) parameters action.get(parameters, {}) if tool_name not in self.tools: return f错误: 工具 {tool_name} 不存在 try: tool self.tools[tool_name] return tool.execute(**parameters) except Exception as e: return f工具执行错误: {str(e)} def _generate_final_response(self, user_input: str, action_result: str, plan: Dict) - str: 基于行动结果生成最终响应 messages [ {role: system, content: 基于用户的原始问题和工具执行结果生成友好的最终响应。}, {role: user, content: f用户问题: {user_input}}, {role: system, content: f工具执行结果: {action_result}}, {role: system, content: f思考过程: {plan[thought_process]}} ] response self.llm_client.chat_completion(messages) return response4.2 LLM客户端封装为了适配不同的LLM提供商我们创建一个统一的客户端接口# utils/llm_client.py import openai import os from typing import List, Dict from dotenv import load_dotenv load_dotenv() class LLMClient: 统一的LLM客户端 def __init__(self, model: str gpt-3.5-turbo, api_key: str None): self.model model self.api_key api_key or os.getenv(OPENAI_API_KEY) openai.api_key self.api_key def chat_completion(self, messages: List[Dict], temperature: float 0.7) - str: 调用聊天补全API try: response openai.ChatCompletion.create( modelself.model, messagesmessages, temperaturetemperature, max_tokens1000 ) return response.choices[0].message.content except Exception as e: return fLLM调用错误: {str(e)} # 本地模型客户端示例使用Ollama class LocalLLMClient: 本地LLM客户端使用Ollama def __init__(self, model: str llama3, base_url: str http://localhost:11434): self.model model self.base_url base_url def chat_completion(self, messages: List[Dict], temperature: float 0.7) - str: 调用本地LLM import requests # 将消息格式转换为Ollama格式 prompt self._format_messages(messages) try: response requests.post( f{self.base_url}/api/generate, json{ model: self.model, prompt: prompt, temperature: temperature, stream: False } ) return response.json()[response] except Exception as e: return f本地LLM调用错误: {str(e)} def _format_messages(self, messages: List[Dict]) - str: 将消息列表格式化为提示词文本 formatted [] for msg in messages: if msg[role] system: formatted.append(f系统: {msg[content]}) elif msg[role] user: formatted.append(f用户: {msg[content]}) elif msg[role] assistant: formatted.append(f助手: {msg[content]}) return \n.join(formatted) \n助手:5. 完整实战演示5.1 创建并配置Agent现在让我们创建一个完整的示例# main.py from core.agent import SimpleAgent from tools.calculator import CalculatorTool from utils.llm_client import LLMClient, LocalLLMClient def main(): # 初始化LLM客户端选择一种 # llm_client LLMClient(gpt-3.5-turbo) # 使用OpenAI llm_client LocalLLMClient(llama3) # 使用本地模型 # 创建工具实例 calculator CalculatorTool() # 创建Agent实例 agent SimpleAgent(llm_client, [calculator]) print( 简单Agent演示 ) print(输入quit退出程序) while True: user_input input(\n用户: ).strip() if user_input.lower() in [quit, exit, 退出]: print(再见) break if not user_input: continue # 处理用户输入 response agent.process_message(user_input) print(f助手: {response}) if __name__ __main__: main()5.2 运行示例对话让我们测试几个典型场景用户: 计算一下 25的平方加上18的三次方是多少 助手: 思考过程: 用户需要计算一个复合数学表达式。我需要先计算25的平方然后计算18的三次方最后将两个结果相加。 行动: {action: calculator, parameters: {operation: calculate, expression: 25**2 18**3}} 最终答案: 计算结果: 25的平方是62518的三次方是5832两者相加得到6457。 用户: 那这个结果除以5是多少 助手: 思考过程: 用户想要将上一个结果6457除以5。我可以直接使用计算器工具。 行动: {action: calculator, parameters: {operation: calculate, expression: 6457 / 5}} 最终答案: 计算结果: 6457除以5等于1291.4。5.3 添加更多工具示例让我们扩展Agent的能力添加网络搜索和文件操作工具# tools/web_search.py import requests from .base_tool import BaseTool class WebSearchTool(BaseTool): 简单的网络搜索工具 def __init__(self): super().__init__( nameweb_search, description搜索网络信息模拟实现 ) def execute(self, query: str) - str: 执行网络搜索模拟 # 实际项目中可以集成真正的搜索API return f搜索 {query} 的结果: 这是模拟的搜索结果实际项目中需要集成搜索API。 def _get_parameters_schema(self) - Dict[str, Any]: return { type: object, properties: { query: { type: string, description: 搜索查询关键词 } }, required: [query] } # tools/file_operator.py import os from .base_tool import BaseTool class FileReadTool(BaseTool): 文件读取工具 def __init__(self, base_path: str ./data): super().__init__( nameread_file, description读取指定文件的内容 ) self.base_path base_path os.makedirs(base_path, exist_okTrue) def execute(self, filename: str) - str: 读取文件内容 filepath os.path.join(self.base_path, filename) try: with open(filepath, r, encodingutf-8) as f: return f.read() except FileNotFoundError: return f错误: 文件 {filename} 不存在 except Exception as e: return f读取文件错误: {str(e)} def _get_parameters_schema(self) - Dict[str, Any]: return { type: object, properties: { filename: { type: string, description: 要读取的文件名 } }, required: [filename] }6. 高级特性实现6.1 思维链Chain of Thought增强为了让Agent的推理过程更加透明我们增强思维链功能# core/advanced_planner.py import re import json from typing import List, Dict, Any class AdvancedPlanner(Planner): 增强的规划器支持多步推理 def __init__(self, llm_client, max_steps: int 5): super().__init__(llm_client) self.max_steps max_steps def multi_step_plan(self, user_input: str, available_tools: List[Dict], conversation_history: str ) - List[Dict[str, Any]]: 多步任务规划 steps [] current_step 1 while current_step self.max_steps: # 构建包含历史信息的提示词 system_prompt self._build_multi_step_prompt( available_tools, conversation_history, steps, current_step ) response self.llm_client.chat_completion([ {role: system, content: system_prompt}, {role: user, content: user_input} ]) plan self._parse_advanced_response(response, current_step) steps.append(plan) # 检查是否完成任务 if plan.get(is_final, False): break current_step 1 return steps def _build_multi_step_prompt(self, tools: List[Dict], history: str, previous_steps: List[Dict], current_step: int) - str: 构建多步规划提示词 tools_desc \n.join([f- {t[name]}: {t[description]} for t in tools]) steps_desc self._format_previous_steps(previous_steps) return f你是一个高级任务规划助手。请将复杂任务分解为多个步骤。 可用工具: {tools_desc} 对话历史: {history} 已完成步骤: {steps_desc} 当前步骤: {current_step} 响应格式: 步骤{current_step}思考: 详细推理 步骤{current_step}行动: JSON动作或无 步骤{current_step}结果预期: 期望得到什么信息 是否最终步骤: 是/否6.2 自我反思与错误处理实现Agent的自我监控和错误恢复机制# core/reflective_agent.py class ReflectiveAgent(SimpleAgent): 具备自我反思能力的Agent def __init__(self, llm_client, tools: List None): super().__init__(llm_client, tools) self.error_history [] def process_message_with_reflection(self, user_input: str) - str: 带反思的消息处理 initial_response self.process_message(user_input) # 检查响应质量 if self._needs_reflection(initial_response): refined_response self._reflect_and_refine(user_input, initial_response) return refined_response return initial_response def _needs_reflection(self, response: str) - bool: 判断是否需要反思 reflection_triggers [ 错误:, 无法, 不明白, 我不确定, 抱歉, 对不起 ] return any(trigger in response.lower() for trigger in reflection_triggers) def _reflect_and_refine(self, user_input: str, initial_response: str) - str: 反思并优化响应 reflection_prompt f 初始响应: {initial_response} 用户问题: {user_input} 请分析为什么初始响应不够好并提出改进方案。然后生成更好的最终响应。 reflection self.llm_client.chat_completion([ {role: system, content: 你是一个质量检查助手负责改进AI助手的响应。}, {role: user, content: reflection_prompt} ]) # 记录错误用于学习 self.error_history.append({ input: user_input, initial_response: initial_response, reflection: reflection }) return reflection7. 性能优化与最佳实践7.1 记忆优化策略当对话历史变长时我们需要优化记忆管理# core/optimized_memory.py class OptimizedMemory(Memory): 优化后的记忆系统 def __init__(self, max_tokens: int 4000, model_name: str gpt-3.5-turbo): super().__init__() self.max_tokens max_tokens self.model_name model_name def get_compressed_history(self) - str: 获取压缩后的对话历史 if len(self.messages) 10: # 消息少时直接返回 return self.get_conversation_history() # 使用LLM压缩历史 compression_prompt f请将以下对话历史压缩为关键信息摘要保留重要事实和决策 {self.get_conversation_history()} 压缩摘要: compressed self.llm_client.chat_completion([ {role: user, content: compression_prompt} ]) return f历史摘要: {compressed}\n最近对话: {self.get_recent_messages(3)} def estimate_tokens(self, text: str) - int: 估算文本的token数量简化版 return len(text) // 4 # 近似估算7.2 工具调用安全规范确保工具调用的安全性# utils/security.py import re import ast class SecurityValidator: 安全验证器 staticmethod def validate_math_expression(expression: str) - bool: 验证数学表达式的安全性 # 允许的数学函数和常量 allowed_names { abs, round, min, max, sum, pow, e, pi, sin, cos, tan, log, sqrt } try: # 解析表达式 tree ast.parse(expression, modeeval) # 检查所有使用的名称 for node in ast.walk(tree): if isinstance(node, ast.Name): if node.id not in allowed_names: return False elif isinstance(node, ast.Call): if not isinstance(node.func, ast.Name): return False if node.func.id not in allowed_names: return False return True except: return False staticmethod def sanitize_filename(filename: str) - str: 文件名安全处理 # 移除危险字符 filename re.sub(r[^\w\.-], _, filename) # 防止路径遍历 filename filename.replace(.., _) return filename8. 常见问题与解决方案8.1 Agent响应质量问题问题现象Agent响应不准确或偏离主题解决方案优化系统提示词明确角色和职责添加响应格式约束和示例实现响应质量检查机制使用更高质量的LLM模型# 改进的系统提示词示例 improved_system_prompt 你是一个专业、准确的助手。请遵循以下原则 1. 如果不确定承认不知道而不是猜测 2. 使用工具前先确认需求 3. 保持响应简洁专业 4. 对于数学问题确保计算准确8.2 工具调用失败处理问题现象工具执行出错或返回意外结果解决方案添加完善的错误处理机制实现工具调用重试逻辑提供有意义的错误信息记录工具调用日志用于调试def safe_tool_execution(self, tool_name: str, parameters: Dict, max_retries: int 3): 安全的工具执行 for attempt in range(max_retries): try: result self.tools[tool_name].execute(**parameters) return result except Exception as e: if attempt max_retries - 1: # 最后一次尝试 return f工具执行失败: {str(e)} # 等待后重试 time.sleep(1 * (attempt 1))8.3 记忆管理挑战问题现象长对话中记忆混乱或丢失重要信息解决方案实现分层记忆系统短期/长期添加关键信息提取和摘要功能定期清理无关记忆使用向量数据库存储长期记忆9. 生产环境部署建议9.1 性能优化配置对于生产环境需要考虑以下优化# config/production.py PRODUCTION_CONFIG { llm: { timeout: 30, # API调用超时 retry_attempts: 3, # 重试次数 rate_limit_delay: 0.1, # 速率限制延迟 }, memory: { max_messages: 500, # 最大消息数 compression_threshold: 50, # 压缩阈值 persistence_interval: 300, # 持久化间隔(秒) }, security: { input_validation: True, # 输入验证 tool_sandboxing: True, # 工具沙箱 audit_logging: True, # 审计日志 } }9.2 监控与日志实现完整的监控体系# utils/monitoring.py import logging import time from datetime import datetime class AgentMonitor: Agent性能监控 def __init__(self): self.logger logging.getLogger(agent_monitor) self.metrics { total_requests: 0, successful_responses: 0, tool_calls: 0, average_response_time: 0 } def log_interaction(self, user_input: str, response: str, processing_time: float, success: bool): 记录交互日志 self.metrics[total_requests] 1 if success: self.metrics[successful_responses] 1 log_entry { timestamp: datetime.now().isoformat(), user_input: user_input[:100], # 截断长输入 response_length: len(response), processing_time: processing_time, success: success } self.logger.info(fInteraction: {log_entry})9.3 扩展性考虑为未来扩展预留接口# core/extensible_agent.py class ExtensibleAgent(SimpleAgent): 支持插件扩展的Agent def __init__(self, llm_client, tools: List None): super().__init__(llm_client, tools) self.plugins [] self.hooks { pre_process: [], post_process: [], pre_tool_call: [], post_tool_call: [] } def register_plugin(self, plugin): 注册插件 self.plugins.append(plugin) # 注册插件钩子 for hook_name in self.hooks: if hasattr(plugin, hook_name): self.hooks[hook_name].append(getattr(plugin, hook_name)) def process_message(self, user_input: str) - str: 带插件钩子的消息处理 # 执行预处理钩子 for hook in self.hooks[pre_process]: user_input hook(user_input) or user_input # 原始处理逻辑 response super().process_message(user_input) # 执行后处理钩子 for hook in self.hooks[post_process]: response hook(response) or response return response通过从零开始实现Agent我们不仅掌握了核心技术原理还建立了可扩展的基础架构。这种深度理解将帮助你在实际项目中更好地选择和使用Agent框架或者在需要高度定制化的场景中游刃有余。记住优秀的Agent开发不仅仅是技术实现更是对用户体验和业务需求的深刻理解。
从零实现Python智能体：思维链、工具调用与记忆机制详解
发布时间：2026/7/30 15:25:51
在AI技术快速发展的今天Agent智能体已成为连接大语言模型与现实应用的重要桥梁。很多开发者一提到Agent开发就想到使用现成框架但框架往往隐藏了底层原理让初学者难以真正理解Agent的工作机制。本文将从零开始手把手带你用纯Python实现一个功能完整的Agent系统涵盖思维链、工具调用、记忆机制等核心概念让你不仅会用Agent更懂Agent的魂。1. Agent核心概念与价值1.1 什么是AgentAgent本质上是一个能够感知环境、进行决策并执行动作的智能系统。在AI语境下Agent通常由大语言模型驱动具备以下核心能力自主性无需人工干预即可完成任务反应性能够感知环境变化并做出响应目标导向有明确的完成任务目标持续性在长时间跨度内保持运作与简单的聊天机器人不同真正的Agent具备规划能力和工具使用能力能够将复杂任务分解为可执行的子步骤。1.2 为什么需要从零开始实现虽然市面上有LangChain、AutoGPT等成熟框架但从零实现有以下优势深入理解原理摆脱框架的黑箱掌握每个组件的设计思路定制化能力强根据具体需求灵活调整架构不受框架限制部署轻量避免引入不必要的依赖适合资源受限场景学习价值高通过亲手实现建立对Agent技术的直觉理解1.3 Agent的核心组件一个完整的Agent系统通常包含以下关键组件推理引擎核心的决策模块通常基于LLM工具系统Agent可调用的外部功能集合记忆机制短期和长期记忆存储规划模块任务分解和步骤规划执行监控动作执行和状态跟踪2. 环境准备与基础架构2.1 开发环境配置首先确保你的Python环境版本在3.8以上这是大多数现代AI库的最低要求。我们将使用以下核心库# 创建虚拟环境 python -m venv agent_env source agent_env/bin/activate # Linux/Mac # 或 agent_env\Scripts\activate # Windows # 安装必要依赖 pip install openai pip install requests pip install python-dotenv如果你没有OpenAI API密钥可以使用开源的本地模型如Ollama搭配Llama 3等模型。2.2 项目结构设计在开始编码前先规划好项目结构agent_project/ ├── core/ │ ├── __init__.py │ ├── agent.py # Agent主类 │ ├── memory.py # 记忆管理 │ └── planner.py # 规划模块 ├── tools/ │ ├── __init__.py │ ├── base_tool.py # 工具基类 │ └── calculator.py # 示例工具 ├── config/ │ └── settings.py # 配置文件 ├── utils/ │ └── helpers.py # 工具函数 └── main.py # 主程序入口这种模块化设计便于后续扩展和维护每个组件职责清晰。3. 核心组件实现3.1 工具系统基础实现工具是Agent与外部世界交互的桥梁。我们先定义工具基类# tools/base_tool.py from abc import ABC, abstractmethod from typing import Dict, Any class BaseTool(ABC): 工具基类所有具体工具都需要继承此类 def __init__(self, name: str, description: str): self.name name self.description description abstractmethod def execute(self, **kwargs) - str: 执行工具的核心方法 pass def get_schema(self) - Dict[str, Any]: 返回工具的调用schema用于LLM理解参数 return { name: self.name, description: self.description, parameters: self._get_parameters_schema() } abstractmethod def _get_parameters_schema(self) - Dict[str, Any]: 定义工具参数的结构 pass接下来实现一个具体的计算器工具# tools/calculator.py import math from typing import Dict, Any from .base_tool import BaseTool class CalculatorTool(BaseTool): 简单的数学计算工具 def __init__(self): super().__init__( namecalculator, description执行基本的数学运算支持加减乘除、幂运算等 ) def execute(self, operation: str, expression: str) - str: 执行数学计算 try: # 安全评估数学表达式 allowed_names {**math.__dict__, abs: abs} result eval(expression, {__builtins__: {}}, allowed_names) return f计算结果: {result} except Exception as e: return f计算错误: {str(e)} def _get_parameters_schema(self) - Dict[str, Any]: return { type: object, properties: { operation: { type: string, description: 操作类型, enum: [calculate] }, expression: { type: string, description: 数学表达式如 2 3 * 4 } }, required: [operation, expression] }3.2 记忆管理系统记忆是Agent保持连续性的关键。我们实现一个简单的记忆系统# core/memory.py from typing import List, Dict, Any from datetime import datetime class Memory: 管理Agent的记忆 def __init__(self, max_messages: int 100): self.max_messages max_messages self.messages: List[Dict[str, Any]] [] def add_message(self, role: str, content: str, metadata: Dict None): 添加消息到记忆 message { role: role, content: content, timestamp: datetime.now().isoformat(), metadata: metadata or {} } self.messages.append(message) # 保持消息数量在限制内 if len(self.messages) self.max_messages: self.messages self.messages[-self.max_messages:] def get_recent_messages(self, count: int 10) - List[Dict[str, Any]]: 获取最近的消息 return self.messages[-count:] if self.messages else [] def get_conversation_history(self) - str: 将对话历史格式化为字符串 history [] for msg in self.messages: history.append(f{msg[role]}: {msg[content]}) return \n.join(history) def clear(self): 清空记忆 self.messages.clear()3.3 规划与推理引擎这是Agent的大脑负责决策和任务分解# core/planner.py import json import re from typing import List, Dict, Any class Planner: 任务规划和推理引擎 def __init__(self, llm_client): self.llm_client llm_client self.thinking_pattern re.compile(r思考过程:\s*(.*?)\s*行动:, re.DOTALL) def plan(self, user_input: str, available_tools: List[Dict]) - Dict[str, Any]: 根据用户输入制定行动计划 # 构建系统提示词 system_prompt self._build_system_prompt(available_tools) # 调用LLM进行规划 response self.llm_client.chat_completion( messages[ {role: system, content: system_prompt}, {role: user, content: user_input} ] ) return self._parse_llm_response(response) def _build_system_prompt(self, tools: List[Dict]) - str: 构建系统提示词 tools_description \n.join([ f- {tool[name]}: {tool[description]} for tool in tools ]) return f你是一个智能助手能够使用工具解决问题。请按照以下格式响应 可用工具: {tools_description} 响应格式: 思考过程: 你的推理过程 行动: 要执行的动作JSON格式 最终答案: 直接给用户的答案 行动格式示例: {{action: tool_name, parameters: {{param1: value1}}}} 如果不需要使用工具直接在最终答案中回复。 def _parse_llm_response(self, response: str) - Dict[str, Any]: 解析LLM的响应 result { thought_process: , action: None, final_answer: } # 提取思考过程 thought_match self.thinking_pattern.search(response) if thought_match: result[thought_process] thought_match.group(1).strip() # 提取行动指令 action_match re.search(r行动:\s*(\{.*?\}), response, re.DOTALL) if action_match: try: result[action] json.loads(action_match.group(1)) except json.JSONDecodeError: result[action] None # 提取最终答案 answer_match re.search(r最终答案:\s*(.*), response, re.DOTALL) if answer_match: result[final_answer] answer_match.group(1).strip() return result4. 完整Agent实现4.1 Agent主类设计现在我们将各个组件组合成完整的Agent# core/agent.py import json from typing import List, Dict, Any from .memory import Memory from .planner import Planner class SimpleAgent: 简单的Agent实现 def __init__(self, llm_client, tools: List None): self.llm_client llm_client self.memory Memory() self.planner Planner(llm_client) self.tools {tool.name: tool for tool in (tools or [])} def add_tool(self, tool): 添加工具 self.tools[tool.name] tool def process_message(self, user_input: str) - str: 处理用户输入并返回响应 # 添加到记忆 self.memory.add_message(user, user_input) # 获取可用工具信息 available_tools [tool.get_schema() for tool in self.tools.values()] # 进行规划 plan self.planner.plan(user_input, available_tools) # 执行行动 if plan[action]: result self._execute_action(plan[action]) # 将行动结果添加到记忆 self.memory.add_message(system, f执行动作: {plan[action]} - 结果: {result}) # 如果需要基于行动结果生成最终响应 if not plan[final_answer]: final_response self._generate_final_response(user_input, result, plan) else: final_response plan[final_answer] else: final_response plan[final_answer] # 添加助手响应到记忆 self.memory.add_message(assistant, final_response, {thought_process: plan[thought_process]}) return final_response def _execute_action(self, action: Dict[str, Any]) - str: 执行具体的工具调用 tool_name action.get(action) parameters action.get(parameters, {}) if tool_name not in self.tools: return f错误: 工具 {tool_name} 不存在 try: tool self.tools[tool_name] return tool.execute(**parameters) except Exception as e: return f工具执行错误: {str(e)} def _generate_final_response(self, user_input: str, action_result: str, plan: Dict) - str: 基于行动结果生成最终响应 messages [ {role: system, content: 基于用户的原始问题和工具执行结果生成友好的最终响应。}, {role: user, content: f用户问题: {user_input}}, {role: system, content: f工具执行结果: {action_result}}, {role: system, content: f思考过程: {plan[thought_process]}} ] response self.llm_client.chat_completion(messages) return response4.2 LLM客户端封装为了适配不同的LLM提供商我们创建一个统一的客户端接口# utils/llm_client.py import openai import os from typing import List, Dict from dotenv import load_dotenv load_dotenv() class LLMClient: 统一的LLM客户端 def __init__(self, model: str gpt-3.5-turbo, api_key: str None): self.model model self.api_key api_key or os.getenv(OPENAI_API_KEY) openai.api_key self.api_key def chat_completion(self, messages: List[Dict], temperature: float 0.7) - str: 调用聊天补全API try: response openai.ChatCompletion.create( modelself.model, messagesmessages, temperaturetemperature, max_tokens1000 ) return response.choices[0].message.content except Exception as e: return fLLM调用错误: {str(e)} # 本地模型客户端示例使用Ollama class LocalLLMClient: 本地LLM客户端使用Ollama def __init__(self, model: str llama3, base_url: str http://localhost:11434): self.model model self.base_url base_url def chat_completion(self, messages: List[Dict], temperature: float 0.7) - str: 调用本地LLM import requests # 将消息格式转换为Ollama格式 prompt self._format_messages(messages) try: response requests.post( f{self.base_url}/api/generate, json{ model: self.model, prompt: prompt, temperature: temperature, stream: False } ) return response.json()[response] except Exception as e: return f本地LLM调用错误: {str(e)} def _format_messages(self, messages: List[Dict]) - str: 将消息列表格式化为提示词文本 formatted [] for msg in messages: if msg[role] system: formatted.append(f系统: {msg[content]}) elif msg[role] user: formatted.append(f用户: {msg[content]}) elif msg[role] assistant: formatted.append(f助手: {msg[content]}) return \n.join(formatted) \n助手:5. 完整实战演示5.1 创建并配置Agent现在让我们创建一个完整的示例# main.py from core.agent import SimpleAgent from tools.calculator import CalculatorTool from utils.llm_client import LLMClient, LocalLLMClient def main(): # 初始化LLM客户端选择一种 # llm_client LLMClient(gpt-3.5-turbo) # 使用OpenAI llm_client LocalLLMClient(llama3) # 使用本地模型 # 创建工具实例 calculator CalculatorTool() # 创建Agent实例 agent SimpleAgent(llm_client, [calculator]) print( 简单Agent演示 ) print(输入quit退出程序) while True: user_input input(\n用户: ).strip() if user_input.lower() in [quit, exit, 退出]: print(再见) break if not user_input: continue # 处理用户输入 response agent.process_message(user_input) print(f助手: {response}) if __name__ __main__: main()5.2 运行示例对话让我们测试几个典型场景用户: 计算一下 25的平方加上18的三次方是多少 助手: 思考过程: 用户需要计算一个复合数学表达式。我需要先计算25的平方然后计算18的三次方最后将两个结果相加。 行动: {action: calculator, parameters: {operation: calculate, expression: 25**2 18**3}} 最终答案: 计算结果: 25的平方是62518的三次方是5832两者相加得到6457。 用户: 那这个结果除以5是多少 助手: 思考过程: 用户想要将上一个结果6457除以5。我可以直接使用计算器工具。 行动: {action: calculator, parameters: {operation: calculate, expression: 6457 / 5}} 最终答案: 计算结果: 6457除以5等于1291.4。5.3 添加更多工具示例让我们扩展Agent的能力添加网络搜索和文件操作工具# tools/web_search.py import requests from .base_tool import BaseTool class WebSearchTool(BaseTool): 简单的网络搜索工具 def __init__(self): super().__init__( nameweb_search, description搜索网络信息模拟实现 ) def execute(self, query: str) - str: 执行网络搜索模拟 # 实际项目中可以集成真正的搜索API return f搜索 {query} 的结果: 这是模拟的搜索结果实际项目中需要集成搜索API。 def _get_parameters_schema(self) - Dict[str, Any]: return { type: object, properties: { query: { type: string, description: 搜索查询关键词 } }, required: [query] } # tools/file_operator.py import os from .base_tool import BaseTool class FileReadTool(BaseTool): 文件读取工具 def __init__(self, base_path: str ./data): super().__init__( nameread_file, description读取指定文件的内容 ) self.base_path base_path os.makedirs(base_path, exist_okTrue) def execute(self, filename: str) - str: 读取文件内容 filepath os.path.join(self.base_path, filename) try: with open(filepath, r, encodingutf-8) as f: return f.read() except FileNotFoundError: return f错误: 文件 {filename} 不存在 except Exception as e: return f读取文件错误: {str(e)} def _get_parameters_schema(self) - Dict[str, Any]: return { type: object, properties: { filename: { type: string, description: 要读取的文件名 } }, required: [filename] }6. 高级特性实现6.1 思维链Chain of Thought增强为了让Agent的推理过程更加透明我们增强思维链功能# core/advanced_planner.py import re import json from typing import List, Dict, Any class AdvancedPlanner(Planner): 增强的规划器支持多步推理 def __init__(self, llm_client, max_steps: int 5): super().__init__(llm_client) self.max_steps max_steps def multi_step_plan(self, user_input: str, available_tools: List[Dict], conversation_history: str ) - List[Dict[str, Any]]: 多步任务规划 steps [] current_step 1 while current_step self.max_steps: # 构建包含历史信息的提示词 system_prompt self._build_multi_step_prompt( available_tools, conversation_history, steps, current_step ) response self.llm_client.chat_completion([ {role: system, content: system_prompt}, {role: user, content: user_input} ]) plan self._parse_advanced_response(response, current_step) steps.append(plan) # 检查是否完成任务 if plan.get(is_final, False): break current_step 1 return steps def _build_multi_step_prompt(self, tools: List[Dict], history: str, previous_steps: List[Dict], current_step: int) - str: 构建多步规划提示词 tools_desc \n.join([f- {t[name]}: {t[description]} for t in tools]) steps_desc self._format_previous_steps(previous_steps) return f你是一个高级任务规划助手。请将复杂任务分解为多个步骤。 可用工具: {tools_desc} 对话历史: {history} 已完成步骤: {steps_desc} 当前步骤: {current_step} 响应格式: 步骤{current_step}思考: 详细推理 步骤{current_step}行动: JSON动作或无 步骤{current_step}结果预期: 期望得到什么信息 是否最终步骤: 是/否6.2 自我反思与错误处理实现Agent的自我监控和错误恢复机制# core/reflective_agent.py class ReflectiveAgent(SimpleAgent): 具备自我反思能力的Agent def __init__(self, llm_client, tools: List None): super().__init__(llm_client, tools) self.error_history [] def process_message_with_reflection(self, user_input: str) - str: 带反思的消息处理 initial_response self.process_message(user_input) # 检查响应质量 if self._needs_reflection(initial_response): refined_response self._reflect_and_refine(user_input, initial_response) return refined_response return initial_response def _needs_reflection(self, response: str) - bool: 判断是否需要反思 reflection_triggers [ 错误:, 无法, 不明白, 我不确定, 抱歉, 对不起 ] return any(trigger in response.lower() for trigger in reflection_triggers) def _reflect_and_refine(self, user_input: str, initial_response: str) - str: 反思并优化响应 reflection_prompt f 初始响应: {initial_response} 用户问题: {user_input} 请分析为什么初始响应不够好并提出改进方案。然后生成更好的最终响应。 reflection self.llm_client.chat_completion([ {role: system, content: 你是一个质量检查助手负责改进AI助手的响应。}, {role: user, content: reflection_prompt} ]) # 记录错误用于学习 self.error_history.append({ input: user_input, initial_response: initial_response, reflection: reflection }) return reflection7. 性能优化与最佳实践7.1 记忆优化策略当对话历史变长时我们需要优化记忆管理# core/optimized_memory.py class OptimizedMemory(Memory): 优化后的记忆系统 def __init__(self, max_tokens: int 4000, model_name: str gpt-3.5-turbo): super().__init__() self.max_tokens max_tokens self.model_name model_name def get_compressed_history(self) - str: 获取压缩后的对话历史 if len(self.messages) 10: # 消息少时直接返回 return self.get_conversation_history() # 使用LLM压缩历史 compression_prompt f请将以下对话历史压缩为关键信息摘要保留重要事实和决策 {self.get_conversation_history()} 压缩摘要: compressed self.llm_client.chat_completion([ {role: user, content: compression_prompt} ]) return f历史摘要: {compressed}\n最近对话: {self.get_recent_messages(3)} def estimate_tokens(self, text: str) - int: 估算文本的token数量简化版 return len(text) // 4 # 近似估算7.2 工具调用安全规范确保工具调用的安全性# utils/security.py import re import ast class SecurityValidator: 安全验证器 staticmethod def validate_math_expression(expression: str) - bool: 验证数学表达式的安全性 # 允许的数学函数和常量 allowed_names { abs, round, min, max, sum, pow, e, pi, sin, cos, tan, log, sqrt } try: # 解析表达式 tree ast.parse(expression, modeeval) # 检查所有使用的名称 for node in ast.walk(tree): if isinstance(node, ast.Name): if node.id not in allowed_names: return False elif isinstance(node, ast.Call): if not isinstance(node.func, ast.Name): return False if node.func.id not in allowed_names: return False return True except: return False staticmethod def sanitize_filename(filename: str) - str: 文件名安全处理 # 移除危险字符 filename re.sub(r[^\w\.-], _, filename) # 防止路径遍历 filename filename.replace(.., _) return filename8. 常见问题与解决方案8.1 Agent响应质量问题问题现象Agent响应不准确或偏离主题解决方案优化系统提示词明确角色和职责添加响应格式约束和示例实现响应质量检查机制使用更高质量的LLM模型# 改进的系统提示词示例 improved_system_prompt 你是一个专业、准确的助手。请遵循以下原则 1. 如果不确定承认不知道而不是猜测 2. 使用工具前先确认需求 3. 保持响应简洁专业 4. 对于数学问题确保计算准确8.2 工具调用失败处理问题现象工具执行出错或返回意外结果解决方案添加完善的错误处理机制实现工具调用重试逻辑提供有意义的错误信息记录工具调用日志用于调试def safe_tool_execution(self, tool_name: str, parameters: Dict, max_retries: int 3): 安全的工具执行 for attempt in range(max_retries): try: result self.tools[tool_name].execute(**parameters) return result except Exception as e: if attempt max_retries - 1: # 最后一次尝试 return f工具执行失败: {str(e)} # 等待后重试 time.sleep(1 * (attempt 1))8.3 记忆管理挑战问题现象长对话中记忆混乱或丢失重要信息解决方案实现分层记忆系统短期/长期添加关键信息提取和摘要功能定期清理无关记忆使用向量数据库存储长期记忆9. 生产环境部署建议9.1 性能优化配置对于生产环境需要考虑以下优化# config/production.py PRODUCTION_CONFIG { llm: { timeout: 30, # API调用超时 retry_attempts: 3, # 重试次数 rate_limit_delay: 0.1, # 速率限制延迟 }, memory: { max_messages: 500, # 最大消息数 compression_threshold: 50, # 压缩阈值 persistence_interval: 300, # 持久化间隔(秒) }, security: { input_validation: True, # 输入验证 tool_sandboxing: True, # 工具沙箱 audit_logging: True, # 审计日志 } }9.2 监控与日志实现完整的监控体系# utils/monitoring.py import logging import time from datetime import datetime class AgentMonitor: Agent性能监控 def __init__(self): self.logger logging.getLogger(agent_monitor) self.metrics { total_requests: 0, successful_responses: 0, tool_calls: 0, average_response_time: 0 } def log_interaction(self, user_input: str, response: str, processing_time: float, success: bool): 记录交互日志 self.metrics[total_requests] 1 if success: self.metrics[successful_responses] 1 log_entry { timestamp: datetime.now().isoformat(), user_input: user_input[:100], # 截断长输入 response_length: len(response), processing_time: processing_time, success: success } self.logger.info(fInteraction: {log_entry})9.3 扩展性考虑为未来扩展预留接口# core/extensible_agent.py class ExtensibleAgent(SimpleAgent): 支持插件扩展的Agent def __init__(self, llm_client, tools: List None): super().__init__(llm_client, tools) self.plugins [] self.hooks { pre_process: [], post_process: [], pre_tool_call: [], post_tool_call: [] } def register_plugin(self, plugin): 注册插件 self.plugins.append(plugin) # 注册插件钩子 for hook_name in self.hooks: if hasattr(plugin, hook_name): self.hooks[hook_name].append(getattr(plugin, hook_name)) def process_message(self, user_input: str) - str: 带插件钩子的消息处理 # 执行预处理钩子 for hook in self.hooks[pre_process]: user_input hook(user_input) or user_input # 原始处理逻辑 response super().process_message(user_input) # 执行后处理钩子 for hook in self.hooks[post_process]: response hook(response) or response return response通过从零开始实现Agent我们不仅掌握了核心技术原理还建立了可扩展的基础架构。这种深度理解将帮助你在实际项目中更好地选择和使用Agent框架或者在需要高度定制化的场景中游刃有余。记住优秀的Agent开发不仅仅是技术实现更是对用户体验和业务需求的深刻理解。
 函数详解：就地构造原理、性能优势与实战应用)
