在人工智能应用开发中Agent智能体正从理论研究快速走向工程实践。很多开发者习惯直接使用 LangChain、AutoGPT 等成熟框架来构建 Agent但框架封装了大量底层细节导致开发者难以真正理解 Agent 的核心工作机制。当遇到复杂业务逻辑或需要深度定制时这种黑盒使用方式会成为瓶颈。本文将通过一个完整的可运行案例从零开始构建一个具备工具调用、状态管理和决策能力的 Agent 系统。我们将使用纯 Python 实现不依赖任何第三方 Agent 框架重点揭示 Agent 内部的消息循环、工具调度和状态管理机制。学完后你将能自主设计适合特定业务的 Agent 架构并在框架选择时做出更明智的技术决策。1. 理解 Agent 的核心组件与工作循环Agent 的本质是一个自主决策系统它通过感知环境、分析状态、执行动作的循环来完成任务。与普通程序的最大区别在于Agent 具备根据环境反馈动态调整行为的能力。1.1 Agent 的四个基本组成部分一个最小可用的 Agent 必须包含以下组件状态管理器维护 Agent 的当前认知状态包括任务目标、执行历史、环境信息等工具集Agent 可以调用的外部能力如计算器、搜索引擎、API 调用等决策引擎基于当前状态决定下一步行动的核心逻辑执行器负责具体执行决策结果并处理执行过程中的异常1.2 Agent 的工作循环流程Agent 的典型工作循环遵循感知-思考-行动模式# Agent 工作循环伪代码 def agent_loop(initial_state): state initial_state while not is_task_complete(state): # 感知获取环境信息 observation perceive_environment(state) # 思考基于状态做出决策 action decision_engine.think(state, observation) # 行动执行决策并更新状态 state executor.execute(action, state) return state这个循环的核心在于状态如何传递和更新以及决策引擎如何根据不断变化的状态做出合理决策。2. 环境准备与项目结构设计我们将构建一个数学问题求解 Agent它能够理解自然语言描述的数字问题调用合适的工具进行计算并给出推理过程。2.1 开发环境要求确保你的 Python 环境满足以下要求组件版本要求说明Python3.8需要类型提示和最新语法特性核心库无额外依赖仅使用 Python 标准库创建项目目录结构math_agent/ ├── core/ │ ├── __init__.py │ ├── agent.py # Agent 核心类 │ ├── state_manager.py # 状态管理 │ └── decision_engine.py # 决策引擎 ├── tools/ │ ├── __init__.py │ ├── base_tool.py # 工具基类 │ └── math_tools.py # 数学工具实现 ├── examples/ │ └── demo.py # 使用示例 └── requirements.txt # 项目依赖当前为空2.2 定义工具接口规范工具是 Agent 能力的扩展点需要统一的接口规范# tools/base_tool.py from abc import ABC, abstractmethod from typing import Any, Dict, Tuple class BaseTool(ABC): 工具基类所有工具必须继承此类 property abstractmethod def name(self) - str: 工具的唯一标识名称 pass property abstractmethod def description(self) - str: 工具的功能描述用于决策引擎判断何时使用 pass abstractmethod def execute(self, **kwargs) - Tuple[bool, Any]: 执行工具操作 Returns: Tuple[成功标志, 执行结果] pass def validate_args(self, **kwargs) - bool: 验证参数是否合法子类可重写 return True这种设计确保了工具接口的一致性便于 Agent 统一管理和调用。3. 实现数学工具集我们的数学 Agent 需要具备基本的计算能力下面实现几个核心工具。3.1 基础算术工具# tools/math_tools.py import re from typing import Tuple, Any from .base_tool import BaseTool class ArithmeticTool(BaseTool): 四则运算工具 property def name(self) - str: return arithmetic_calculator property def description(self) - str: return 执行加减乘除四则运算输入应包含数字和运算符 def execute(self, expression: str) - Tuple[bool, Any]: try: # 安全验证只允许数字和基本运算符 if not re.match(r^[\d\-*/().\s]$, expression): return False, 表达式包含不安全字符 # 使用 eval 但要限制作用域 allowed_globals {__builtins__: {}} allowed_locals {} result eval(expression, allowed_globals, allowed_locals) return True, f{expression} {result} except Exception as e: return False, f计算错误: {str(e)} class ComparisonTool(BaseTool): 数值比较工具 property def name(self) - str: return number_comparison property def description(self) - str: return 比较两个数字的大小关系 def execute(self, a: float, b: float) - Tuple[bool, Any]: try: if a b: result f{a} 大于 {b} elif a b: result f{a} 小于 {b} else: result f{a} 等于 {b} return True, result except Exception as e: return False, f比较错误: {str(e)}3.2 工具管理器工具管理器负责维护工具注册和查找# tools/__init__.py from typing import Dict, List from .base_tool import BaseTool class ToolManager: 工具管理器负责工具的注册和查找 def __init__(self): self._tools: Dict[str, BaseTool] {} def register_tool(self, tool: BaseTool) - None: 注册工具 if tool.name in self._tools: raise ValueError(f工具 {tool.name} 已存在) self._tools[tool.name] tool def get_tool(self, name: str) - BaseTool: 根据名称获取工具 if name not in self._tools: raise KeyError(f工具 {name} 未注册) return self._tools[name] def list_tools(self) - List[Dict[str, str]]: 列出所有可用工具的信息 return [ {name: tool.name, description: tool.description} for tool in self._tools.values() ]4. 构建状态管理系统Agent 的状态管理是其具备持续对话和能力的关键。4.1 定义状态数据结构# core/state_manager.py from typing import List, Dict, Any, Optional from dataclasses import dataclass, field import time dataclass class ActionRecord: 动作执行记录 tool_name: str parameters: Dict[str, Any] success: bool result: Any timestamp: float field(default_factorytime.time) dataclass class AgentState: Agent 的完整状态 session_id: str original_query: str current_goal: str action_history: List[ActionRecord] field(default_factorylist) context: Dict[str, Any] field(default_factorydict) start_time: float field(default_factorytime.time) def add_action_record(self, record: ActionRecord) - None: 添加动作记录 self.action_history.append(record) def get_recent_actions(self, count: int 5) - List[ActionRecord]: 获取最近的动作记录 return self.action_history[-count:] def is_goal_achieved(self) - bool: 判断当前目标是否已完成 # 简单的目标完成判断最近一次动作成功且与目标相关 if not self.action_history: return False last_action self.action_history[-1] return (last_action.success and self.current_goal in str(last_action.result))4.2 状态管理器实现# core/state_manager.py class StateManager: 状态管理器 def __init__(self): self._sessions: Dict[str, AgentState] {} def create_session(self, query: str, session_id: Optional[str] None) - str: 创建新会话 if session_id is None: session_id fsession_{int(time.time()*1000)} state AgentState( session_idsession_id, original_queryquery, current_goalquery ) self._sessions[session_id] state return session_id def get_state(self, session_id: str) - AgentState: 获取会话状态 if session_id not in self._sessions: raise KeyError(f会话 {session_id} 不存在) return self._sessions[session_id] def update_goal(self, session_id: str, new_goal: str) - None: 更新当前目标 state self.get_state(session_id) state.current_goal new_goal5. 开发决策引擎决策引擎是 Agent 的大脑负责分析状态并决定下一步行动。5.1 基础决策逻辑# core/decision_engine.py import re from typing import List, Dict, Any, Optional, Tuple from .state_manager import AgentState from tools import ToolManager class DecisionEngine: 决策引擎分析状态并决定下一步行动 def __init__(self, tool_manager: ToolManager): self.tool_manager tool_manager def analyze_query(self, query: str) - Dict[str, Any]: 分析查询意图 analysis { contains_math: bool(re.search(r[\d\-*/()], query)), contains_comparison: bool(re.search(r(大于|小于|等于|比较), query)), question_words: bool(re.search(r(多少|几|怎么|如何|为什么), query)) } return analysis def select_tool(self, state: AgentState) - Optional[Tuple[str, Dict[str, Any]]]: 根据当前状态选择合适的工具和参数 analysis self.analyze_query(state.current_goal) # 简单的规则匹配策略 if analysis[contains_math] and not analysis[contains_comparison]: # 提取数学表达式 expression self._extract_math_expression(state.current_goal) if expression: return arithmetic_calculator, {expression: expression} elif analysis[contains_comparison]: # 提取比较的数字 numbers self._extract_numbers(state.current_goal) if len(numbers) 2: return number_comparison, {a: numbers[0], b: numbers[1]} return None def _extract_math_expression(self, text: str) - Optional[str]: 从文本中提取数学表达式 # 简单的表达式提取逻辑 match re.search(r(\d\.?\d*[\s]*[\-*/][\s]*\d\.?\d*), text) return match.group(1) if match else None def _extract_numbers(self, text: str) - List[float]: 从文本中提取所有数字 numbers [] for match in re.finditer(r\d\.?\d*, text): try: numbers.append(float(match.group())) except ValueError: continue return numbers5.2 增强的决策策略在实际项目中决策引擎需要更复杂的策略# core/decision_engine.py class EnhancedDecisionEngine(DecisionEngine): 增强的决策引擎考虑历史记录和上下文 def select_tool(self, state: AgentState) - Optional[Tuple[str, Dict[str, Any]]]: # 先尝试基础匹配 basic_result super().select_tool(state) if basic_result: return basic_result # 基于历史记录的分析 return self._strategy_based_on_history(state) def _strategy_based_on_history(self, state: AgentState) - Optional[Tuple[str, Dict[str, Any]]]: 基于历史记录的决策策略 if not state.action_history: return None # 分析最近的成功模式 recent_success [a for a in state.action_history[-3:] if a.success] if recent_success: # 如果最近有成功记录尝试类似的工具 last_success recent_success[-1] return last_success.tool_name, last_success.parameters return None6. 实现 Agent 核心类现在我们将各个组件整合成完整的 Agent。6.1 Agent 基础实现# core/agent.py import time from typing import Any, Dict, Optional from .state_manager import StateManager, ActionRecord from .decision_engine import EnhancedDecisionEngine from tools import ToolManager class MathAgent: 数学问题求解 Agent def __init__(self): self.tool_manager ToolManager() self.state_manager StateManager() self.decision_engine EnhancedDecisionEngine(self.tool_manager) self._setup_tools() def _setup_tools(self) - None: 注册所有可用工具 from tools.math_tools import ArithmeticTool, ComparisonTool self.tool_manager.register_tool(ArithmeticTool()) self.tool_manager.register_tool(ComparisonTool()) def process_query(self, query: str, session_id: Optional[str] None) - Dict[str, Any]: 处理用户查询 # 创建或获取会话 if session_id is None or session_id not in self.state_manager._sessions: session_id self.state_manager.create_session(query, session_id) state self.state_manager.get_state(session_id) # 决策-执行循环 max_steps 5 # 防止无限循环 for step in range(max_steps): # 决策阶段 tool_selection self.decision_engine.select_tool(state) if tool_selection is None: return self._format_response(state, 无法处理该问题, False) tool_name, parameters tool_selection # 执行阶段 try: tool self.tool_manager.get_tool(tool_name) success, result tool.execute(**parameters) # 记录执行结果 action_record ActionRecord( tool_nametool_name, parametersparameters, successsuccess, resultresult ) state.add_action_record(action_record) # 检查目标是否达成 if state.is_goal_achieved() or step max_steps - 1: return self._format_response(state, result, success) except Exception as e: action_record ActionRecord( tool_nametool_name, parametersparameters, successFalse, resultstr(e) ) state.add_action_record(action_record) return self._format_response(state, f执行错误: {str(e)}, False) def _format_response(self, state: AgentState, result: Any, success: bool) - Dict[str, Any]: 格式化响应 return { session_id: state.session_id, success: success, result: result, action_history: [ { tool: record.tool_name, parameters: record.parameters, success: record.success, result: record.result } for record in state.action_history ], processing_time: time.time() - state.start_time }7. 运行验证与结果分析7.1 基础功能测试创建测试脚本来验证 Agent 功能# examples/demo.py from core.agent import MathAgent def test_basic_operations(): 测试基础数学运算 agent MathAgent() test_cases [ 计算 25 37 等于多少, 比较 15.5 和 20.3 的大小, 请问 100 除以 4 的结果是什么 ] for i, query in enumerate(test_cases, 1): print(f\n 测试案例 {i} ) print(f问题: {query}) response agent.process_query(query) print(f成功: {response[success]}) print(f结果: {response[result]}) print(f处理时间: {response[processing_time]:.2f}秒) if response[action_history]: print(执行历史:) for action in response[action_history]: print(f - 工具: {action[tool]}) print(f 参数: {action[parameters]}) print(f 成功: {action[success]}) if __name__ __main__: test_basic_operations()7.2 预期输出示例运行测试脚本应该看到类似输出 测试案例 1 问题: 计算 25 37 等于多少 成功: True 结果: 25 37 62 处理时间: 0.05秒 执行历史: - 工具: arithmetic_calculator 参数: {expression: 25 37} 成功: True 测试案例 2 问题: 比较 15.5 和 20.3 的大小 成功: True 结果: 15.5 小于 20.3 处理时间: 0.03秒 执行历史: - 工具: number_comparison 参数: {a: 15.5, b: 20.3} 成功: True7.3 会话连续性测试验证 Agent 在多轮对话中的表现# examples/conversation_test.py from core.agent import MathAgent def test_conversation(): 测试多轮对话能力 agent MathAgent() session_id None conversations [ 25 37 等于多少, 那再乘以 2 呢, 比较这个结果和 100 的大小 ] for query in conversations: print(f\n用户: {query}) response agent.process_query(query, session_id) session_id response[session_id] print(fAgent: {response[result]}) print(f会话ID: {session_id}) if __name__ __main__: test_conversation()8. 常见问题排查与调试8.1 工具执行失败诊断当工具执行失败时需要系统化的排查方法问题现象可能原因检查方式解决方案工具找不到工具未正确注册检查tool_manager.list_tools()确保工具在_setup_tools()中注册参数错误参数类型或格式不匹配打印决策引擎输出的参数在工具中增加参数验证逻辑计算异常数学表达式不合法查看具体的异常信息在工具执行中添加异常捕获会话丢失session_id 管理错误检查状态管理器的会话存储确保每次对话使用相同的 session_id8.2 决策逻辑调试决策引擎是复杂性的主要来源需要有效的调试手段# 在 DecisionEngine 类中添加调试方法 def debug_decision(self, state: AgentState) - Dict[str, Any]: 决策过程调试信息 analysis self.analyze_query(state.current_goal) tool_selection self.select_tool(state) return { query_analysis: analysis, selected_tool: tool_selection[0] if tool_selection else None, tool_parameters: tool_selection[1] if tool_selection else None, available_tools: self.tool_manager.list_tools() }8.3 性能监控点在生产环境中需要监控以下关键指标单次决策耗时工具执行成功率会话平均步数内存使用情况9. 生产环境最佳实践9.1 安全性增强当前实现中的eval使用存在安全风险生产环境需要替换# 安全的表达式计算替代方案 import operator class SafeArithmeticTool(BaseTool): 安全的算术工具避免使用 eval def execute(self, expression: str) - Tuple[bool, Any]: try: # 使用 AST 解析或自定义解析器 result self._safe_eval(expression) return True, f{expression} {result} except Exception as e: return False, f计算错误: {str(e)} def _safe_eval(self, expression: str) - float: 安全的表达式求值 # 实现安全的表达式解析逻辑 # 这里可以使用第三方库如 simpleeval pass9.2 可扩展性设计为支持更复杂的应用场景可以考虑以下扩展点工具热插拔支持运行时动态加载和卸载工具决策策略配置化通过配置文件调整决策逻辑状态持久化支持会话状态的保存和恢复性能监控集成指标收集和性能分析9.3 错误处理与降级策略健壮的 Agent 需要完善的错误处理机制class RobustMathAgent(MathAgent): 增强错误处理的 Agent def process_query(self, query: str, session_id: Optional[str] None) - Dict[str, Any]: try: return super().process_query(query, session_id) except Exception as e: # 记录详细错误日志 self._log_error(e, query, session_id) # 返回友好的错误信息 return { success: False, result: 系统暂时无法处理您的请求, error_type: type(e).__name__ } def _log_error(self, error: Exception, query: str, session_id: Optional[str]) - None: 记录错误日志 # 实现日志记录逻辑 pass10. 扩展方向与进阶学习基于这个基础 Agent 框架你可以向多个方向扩展10.1 集成大型语言模型将决策引擎与 LLM 结合实现更自然语言理解class LLMEnhancedDecisionEngine(DecisionEngine): LLM 增强的决策引擎 def select_tool(self, state: AgentState) - Optional[Tuple[str, Dict[str, Any]]]: # 使用 LLM 分析用户意图 intent_analysis self.llm_analyze(state.current_goal) return self._map_intent_to_tool(intent_analysis)10.2 多 Agent 协作实现多个 Agent 之间的协作机制class MultiAgentSystem: 多 Agent 协作系统 def __init__(self): self.agents: Dict[str, MathAgent] {} self.coordination_engine CoordinationEngine() def solve_complex_problem(self, problem: str) - Dict[str, Any]: 使用多个 Agent 协作解决复杂问题 # 问题分解、任务分配、结果整合 pass10.3 可视化与监控开发管理界面来监控 Agent 的运行状态实时显示决策过程工具使用统计性能指标仪表盘错误日志分析这个从零开始的 Agent 实现展示了智能体系统的核心机制。虽然功能相对基础但包含了状态管理、工具调用、决策循环等关键概念。在实际项目中你可以基于这个框架逐步添加更复杂的特性如学习能力、长期记忆、多模态处理等最终构建出适合特定业务场景的智能体系统。理解这些底层机制的最大价值在于当使用高级框架遇到复杂问题时你能够快速定位到问题根源并具备定制化解决方案的能力。建议在掌握基础原理后再对比学习 LangChain、AutoGPT 等框架的设计思路这样能更深刻地理解框架所做的取舍和优化方向。