在实际 AI 工程实践中构建一个能够自主思考、规划并执行任务的智能体Agent是迈向高级应用的关键一步。许多开发者通过跑通一个简单的 Demo 来理解 Agent 的概念但若想将这项技能转化为简历上的亮点尤其是应对大厂对系统设计和工程化能力的高要求就必须深入理解其核心机制——Agent Loop并完成一个结构清晰、可维护、可扩展的工程实践项目。Triple-pi 作为一个在 GitHub 上获得近 8 万星标的知名项目其设计理念和实现为我们提供了一个绝佳的学习范本。它不仅仅是一个玩具项目其模块化设计、清晰的职责分离以及对不同规模模型LLM的适配策略都体现了生产级 AI 应用的思考。本文将带你超越 Demo 阶段以 Triple-pi 项目为蓝本手把手构建一个具备完整 Agent Loop 的系统。我们将从零开始剖析 Agent 的核心工作循环设计项目结构实现关键模块并最终整合成一个可以处理复杂任务的智能体。整个过程将严格遵循工程化实践注重代码结构、配置管理、错误处理和可观测性确保你构建出的不是一个“一次性”脚本而是一个值得写进简历的、具备工业水准的项目。1. 理解 Agent Loop智能体的工作心脏在开始编码之前必须透彻理解驱动智能体的核心引擎——Agent Loop。它决定了 Agent 如何感知、思考、行动并学习是区分简单提示词调用与真正自主系统的关键。1.1 Agent Loop 的基本构成一个典型的 Agent Loop 包含四个核心阶段它们循环往复直至任务完成或达到终止条件观察ObservationAgent 从环境中获取信息。这可能是用户的输入、传感器的数据、数据库的查询结果或者是上一个工具Tool执行的输出。观察为决策提供了上下文。思考ThoughtAgent 的核心处理阶段。它利用大型语言模型LLM分析观察到的信息结合自身的目标和记忆规划下一步行动。思考的输出通常是一个“推理过程”和一个明确的“行动指令”。行动ActionAgent 根据思考阶段产生的指令调用一个特定的工具Tool来执行具体操作。例如执行一段代码、调用一个 API、查询数据库或操作文件系统。反馈Observation’行动执行后环境会给出一个结果成功、失败、返回数据等。这个结果作为新的“观察”被输入到下一个循环的起点。这个观察 - 思考 - 行动 - 反馈的循环会持续进行直到 LLM 在思考阶段判断任务已经完成输出一个最终答案或者触发了预设的停止条件如循环次数超限。1.2 Triple-pi 对 Agent Loop 的工程化抽象Triple-pi 等项目之所以受到推崇是因为它们没有将上述循环写成一个庞杂的while循环而是进行了清晰的职责抽象通常包含以下核心组件Agent 核心Agent Core持有 LLM 的配置和调用逻辑是“思考”的发起者。它接收包含历史记录和当前观察的提示Prompt并生成包含推理和行动指令的响应。工具集Tools一组可被 Agent 调用的函数或接口是“行动”的执行者。每个工具都有明确的名称、描述和参数定义以便 LLM 理解其用途。记忆Memory存储 Agent 与环境的交互历史即过去的观察、思考、行动、反馈。短期记忆如对话轮次和长期记忆如向量数据库共同帮助 Agent 保持上下文连贯性。执行器Executor协调整个循环的组件。它负责组装提示、调用 Agent 核心、解析 LLM 响应、匹配并执行工具、处理工具结果、更新记忆并决定循环是否继续。提示词模板Prompt Templates将当前任务、历史记忆、可用工具列表等信息结构化成 LLM 能高效理解的文本格式。良好的模板设计直接决定 Agent 的推理质量。理解了这个架构我们就知道代码不是从一个庞大的main函数开始而是需要先定义这些组件之间的接口和通信协议。2. 项目环境搭建与核心依赖配置我们选择 Python 作为实现语言因为它拥有最丰富的 AI 开发生态。项目将使用poetry进行依赖管理确保环境隔离和可复现性。如果你使用pip和venv原理是相通的。2.1 初始化项目结构首先创建一个清晰的项目目录结构。好的结构是工程实践的第一步。mkdir triple-pi-agent-project cd triple-pi-agent-project使用poetry初始化项目如果没有安装poetry可以使用pip install poetry。poetry new . # 这会在当前目录生成 pyproject.toml 等文件但我们需要调整结构。 # 我们可以手动创建更符合我们需求的结构。推荐的项目结构如下triple-pi-agent-project/ ├── pyproject.toml # 项目依赖和配置 ├── README.md ├── .env.example # 环境变量示例 ├── src/ │ └── triple_pi_agent/ │ ├── __init__.py │ ├── agent/ # Agent核心模块 │ │ ├── __init__.py │ │ ├── core.py # Agent核心类 │ │ └── prompts.py # 提示词模板 │ ├── tools/ # 工具模块 │ │ ├── __init__.py │ │ ├── base.py # 工具基类 │ │ ├── calculator.py │ │ ├── web_search.py │ │ └── ... │ ├── memory/ # 记忆模块 │ │ ├── __init__.py │ │ └── buffer.py # 对话缓冲记忆 │ ├── executor/ # 执行器模块 │ │ ├── __init__.py │ │ └── sequential.py # 顺序执行器 │ ├── schemas/ # Pydantic数据模型 │ │ ├── __init__.py │ │ └── agent.py # 定义Agent输入输出格式 │ └── cli.py # 命令行入口 ├── tests/ # 单元测试 │ ├── __init__.py │ ├── test_agent.py │ └── test_tools.py └── scripts/ # 辅助脚本 └── run_agent.py2.2 配置核心依赖编辑pyproject.toml文件添加必要的依赖。我们将使用langchain社区的核心抽象作为参考但会强调自己的实现以加深理解。同时我们会集成OpenAI的 API 作为 LLM 后端。[tool.poetry] name triple-pi-agent version 0.1.0 description A practical agent system with full agent loop implementation. authors [Your Name your.emailexample.com] [tool.poetry.dependencies] python ^3.9 openai ^1.3.0 # 用于调用GPT系列模型 pydantic ^2.0.0 # 用于数据验证和设置管理 tenacity ^8.2.0 # 用于重试逻辑 requests ^2.31.0 # 用于工具中调用HTTP API python-dotenv ^1.0.0 # 用于加载环境变量 # 可选如果你计划使用本地模型可以添加 transformers, torch 等 [tool.poetry.group.dev.dependencies] pytest ^7.4.0 black ^23.0.0 isort ^5.12.0 mypy ^1.5.0 [build-system] requires [poetry-core] build-backend poetry.core.masonry.api安装依赖poetry install2.3 设置环境变量创建.env文件请确保将其加入.gitignore用于存储敏感信息如 API Key。# .env OPENAI_API_KEYsk-your-openai-api-key-here OPENAI_BASE_URLhttps://api.openai.com/v1 # 如果使用Azure或代理可修改 MODEL_NAMEgpt-3.5-turbo # 或 gpt-4, gpt-4-turbo-preview在代码中我们使用python-dotenv来加载这些配置。3. 实现核心组件从工具到执行器现在我们开始自底向上地实现 Agent 系统的各个组件。我们将遵循“定义接口再实现具体类”的原则。3.1 定义工具基类与具体工具在src/triple_pi_agent/tools/base.py中我们定义一个所有工具都必须遵守的协议基类。from abc import ABC, abstractmethod from typing import Any, Dict, Optional from pydantic import BaseModel, Field class ToolInput(BaseModel): 工具调用的输入参数模型。 # 这是一个示例具体参数由工具定义 argument: str Field(descriptionThe input argument for the tool.) class BaseTool(ABC): 工具基类。每个工具必须有名称、描述和调用方法。 name: str description: str args_schema: Optional[type[BaseModel]] None def __init__(self, name: str, description: str): self.name name self.description description abstractmethod async def _arun(self, **kwargs: Any) - str: 异步执行工具的核心逻辑。 pass async def run(self, tool_input: str, **kwargs: Any) - str: 对外暴露的run方法。可以在这里添加输入验证、错误处理、日志等。 tool_input: 通常是LLM解析出的参数字符串需要进一步处理。 # 这里可以添加输入解析逻辑例如将JSON字符串解析为字典 # 为了简化我们假设工具输入已经是合适的格式或由子类处理 try: result await self._arun(**self._parse_input(tool_input)) return result except Exception as e: return fError executing tool {self.name}: {str(e)} def _parse_input(self, tool_input: str) - Dict[str, Any]: 简单的输入解析子类可重写。默认尝试解析JSON否则作为单个参数。 import json try: return json.loads(tool_input) except json.JSONDecodeError: # 如果不是JSON假设整个字符串是单个参数 return {input: tool_input} def __str__(self): return f{self.name}: {self.description}接下来实现一个具体的计算器工具src/triple_pi_agent/tools/calculator.py。import math import ast import operator as op from typing import Dict, Any from .base import BaseTool, ToolInput # 支持的安全运算符 _SAFE_OPERATORS { ast.Add: op.add, ast.Sub: op.sub, ast.Mult: op.mul, ast.Div: op.truediv, ast.Pow: op.pow, ast.USub: op.neg, ast.Mod: op.mod, } class CalculatorToolInput(ToolInput): expression: str Field(descriptionA mathematical expression to evaluate, e.g., 2 3 * 4.) class CalculatorTool(BaseTool): A tool to evaluate mathematical expressions. Uses ast for safety. def __init__(self): super().__init__( namecalculator, descriptionUseful for performing arithmetic calculations. Input should be a valid mathematical expression. ) self.args_schema CalculatorToolInput async def _arun(self, expression: str, **kwargs: Any) - str: 安全地计算数学表达式。 try: # 使用抽象语法树进行安全评估避免eval的直接风险 node ast.parse(expression, modeeval).body result self._eval_node(node) return str(result) except (SyntaxError, TypeError, KeyError, ZeroDivisionError) as e: return fCalculation error: {str(e)}. Please check your expression: {expression} def _eval_node(self, node): 递归评估AST节点。 if isinstance(node, ast.Num): # Python 3.8 return node.n elif isinstance(node, ast.Constant): # Python 3.8 return node.value elif isinstance(node, ast.BinOp): left_val self._eval_node(node.left) right_val self._eval_node(node.right) operator_func _SAFE_OPERATORS.get(type(node.op)) if operator_func is None: raise TypeError(fUnsupported operator: {type(node.op)}) return operator_func(left_val, right_val) elif isinstance(node, ast.UnaryOp): operand_val self._eval_node(node.operand) operator_func _SAFE_OPERATORS.get(type(node.op)) if operator_func is None: raise TypeError(fUnsupported unary operator: {type(node.op)}) return operator_func(operand_val) else: raise TypeError(fUnsupported AST node type: {type(node)}) def _parse_input(self, tool_input: str) - Dict[str, Any]: # 计算器工具期望一个表达式字符串 return {expression: tool_input.strip()}3.2 构建记忆模块我们实现一个简单的对话缓冲记忆用于保存最近的交互历史。更复杂的记忆如向量存储可以在此基础上扩展。src/triple_pi_agent/memory/buffer.py:from typing import List, Dict, Any from pydantic import BaseModel class Message(BaseModel): 代表Agent循环中的一条消息。 role: str # 例如user, assistant, tool content: str class ConversationBufferMemory: 存储最近的对话历史。 def __init__(self, max_turns: int 10): self.messages: List[Message] [] self.max_turns max_turns def add_message(self, role: str, content: str): 添加一条消息。 self.messages.append(Message(rolerole, contentcontent)) # 如果超过最大轮次移除最早的消息可以根据策略调整 if len(self.messages) self.max_turns * 2: # 粗略估计一轮包含user和assistant self.messages self.messages[-self.max_turns * 2:] def get_messages(self) - List[Dict[str, str]]: 获取格式化后的消息列表用于构造LLM提示。 return [{role: msg.role, content: msg.content} for msg in self.messages] def clear(self): 清空记忆。 self.messages []3.3 设计 Agent 核心与提示词模板Agent 核心负责与 LLM 交互。我们使用 OpenAI 的官方客户端。src/triple_pi_agent/agent/core.py:import os from typing import List, Dict, Any, Optional from openai import AsyncOpenAI from dotenv import load_dotenv from ..schemas.agent import AgentResponse, ToolCall import json load_dotenv() class OpenAIAgentCore: 基于OpenAI API的Agent核心。 def __init__( self, model: Optional[str] None, api_key: Optional[str] None, base_url: Optional[str] None, temperature: float 0.1, # 低温度使输出更确定适合工具调用 ): self.model model or os.getenv(MODEL_NAME, gpt-3.5-turbo) self.client AsyncOpenAI( api_keyapi_key or os.getenv(OPENAI_API_KEY), base_urlbase_url or os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1), ) self.temperature temperature async def generate( self, messages: List[Dict[str, str]], tools: List[Dict[str, Any]], # OpenAI格式的工具描述列表 max_tokens: int 1500, ) - AgentResponse: 调用LLM生成包含思考和工具调用的响应。 返回结构化的AgentResponse。 try: response await self.client.chat.completions.create( modelself.model, messagesmessages, toolstools, tool_choiceauto, # 让模型决定是否调用工具 temperatureself.temperature, max_tokensmax_tokens, ) message response.choices[0].message # 解析响应 final_answer None tool_calls [] if message.content: # 模型直接给出了最终答案 final_answer message.content.strip() elif message.tool_calls: # 模型要求调用工具 for tc in message.tool_calls: tool_calls.append( ToolCall( idtc.id, nametc.function.name, argumentstc.function.arguments ) ) return AgentResponse( thoughtmessage.content or fDecided to call tool(s): {[tc.name for tc in tool_calls]}, final_answerfinal_answer, tool_callstool_calls, ) except Exception as e: # 在实际项目中这里应该有更细致的错误处理和重试逻辑 return AgentResponse( thoughtfError during LLM call: {str(e)}, final_answerNone, tool_calls[], )对应的数据模型定义在src/triple_pi_agent/schemas/agent.py:from pydantic import BaseModel from typing import Optional, List class ToolCall(BaseModel): id: str name: str arguments: str # JSON格式的参数字符串 class AgentResponse(BaseModel): thought: str # 模型的推理过程或直接输出 final_answer: Optional[str] None # 如果是最终答案则存在 tool_calls: List[ToolCall] [] # 如果需要调用工具则存在提示词模板对于引导 LLM 按照 Agent 格式思考至关重要。src/triple_pi_agent/agent/prompts.py:def get_system_prompt(tools_descriptions: str) - str: 生成系统提示词定义Agent的角色和行为规范。 return fYou are a helpful and precise assistant. You have access to the following tools: {tools_descriptions} To solve a users task, you must ALWAYS follow this format: Thought: You should always think about what to do. Explain your reasoning step by step. If you need to use a tool, state which one and why. Action: The action to take, must be one of the tool names listed above, or Final Answer if you have the answer. Action Input: The input to the action, formatted as a JSON string if the tool requires it. If action is Final Answer, this is your final response to the user. After an Action that is a tool call, you will receive an Observation, which is the result of that tool. Then you must continue with another Thought, Action, Action Input cycle until the task is solved. Begin! def format_tool_for_prompt(tool) - str: 将工具对象格式化为LLM可理解的描述文本。 args_desc if tool.args_schema: # 这里简化处理实际可以解析Pydantic模型的schema args_desc f It takes input that should be a JSON object with the required arguments. return f- {tool.name}: {tool.description}{args_desc}3.4 实现执行器串联整个循环执行器是 Agent Loop 的调度中心。我们实现一个顺序执行器。src/triple_pi_agent/executor/sequential.py:import asyncio import json from typing import List, Optional from ..agent.core import OpenAIAgentCore from ..tools.base import BaseTool from ..memory.buffer import ConversationBufferMemory from ..agent.prompts import get_system_prompt, format_tool_for_prompt from ..schemas.agent import AgentResponse class SequentialExecutor: 顺序执行Agent循环。 def __init__( self, agent_core: OpenAIAgentCore, tools: List[BaseTool], memory: Optional[ConversationBufferMemory] None, max_iterations: int 10, ): self.agent_core agent_core self.tools {tool.name: tool for tool in tools} self.memory memory or ConversationBufferMemory() self.max_iterations max_iterations # 初始化系统提示 tools_description \n.join([format_tool_for_prompt(tool) for tool in tools]) self.system_prompt get_system_prompt(tools_description) async def run(self, user_input: str) - str: 执行Agent循环处理用户输入返回最终答案。 self.memory.clear() self.memory.add_message(system, self.system_prompt) self.memory.add_message(user, user_input) iteration 0 final_answer None while iteration self.max_iterations and final_answer is None: iteration 1 print(f\n--- Iteration {iteration} ---) # 1. 观察从记忆构建消息列表 messages_for_llm self.memory.get_messages() # 2. 思考 决策调用Agent核心 openai_tools_format [ { type: function, function: { name: tool.name, description: tool.description, parameters: tool.args_schema.schema() if tool.args_schema else {type: object, properties: {}}, } } for tool in self.tools.values() ] agent_response: AgentResponse await self.agent_core.generate( messagesmessages_for_llm, toolsopenai_tools_format, ) # 将模型的“思考”加入记忆 self.memory.add_message(assistant, fThought: {agent_response.thought}) # 3. 行动判断是最终答案还是工具调用 if agent_response.final_answer: final_answer agent_response.final_answer self.memory.add_message(assistant, fFinal Answer: {final_answer}) break elif agent_response.tool_calls: for tool_call in agent_response.tool_calls: tool_name tool_call.name tool_args tool_call.arguments # 记录行动 action_message fAction: {tool_name}\nAction Input: {tool_args} self.memory.add_message(assistant, action_message) print(fExecuting Tool: {tool_name} with args: {tool_args}) # 执行工具 tool self.tools.get(tool_name) if not tool: observation fError: Tool {tool_name} not found. else: try: observation await tool.run(tool_args) except Exception as e: observation fError executing {tool_name}: {str(e)} # 4. 反馈将观察结果加入记忆 print(fTool Result: {observation[:100]}...) # 打印部分结果 self.memory.add_message(tool, fObservation: {observation}) else: # 模型既没有给出答案也没有调用工具可能出错了 observation Error: Model response did not contain a final answer or tool calls. self.memory.add_message(system, observation) final_answer I encountered an error in processing. Please try again. break if final_answer is None: final_answer fReached maximum iterations ({self.max_iterations}) without a final answer. return final_answer4. 组装与运行构建你的第一个智能体所有核心组件都已就绪现在我们将它们组装起来并创建一个命令行接口来交互。4.1 创建主程序入口src/triple_pi_agent/cli.py:import asyncio import sys from .agent.core import OpenAIAgentCore from .tools.calculator import CalculatorTool from .executor.sequential import SequentialExecutor async def main(): 主函数初始化并运行Agent。 print(Initializing Triple-pi Style Agent...) # 1. 初始化Agent核心 agent_core OpenAIAgentCore() # 2. 注册工具 tools [ CalculatorTool(), # 未来可以在这里添加更多工具如 WebSearchTool(), DBTool() 等 ] print(fLoaded tools: {[tool.name for tool in tools]}) # 3. 初始化执行器 executor SequentialExecutor(agent_coreagent_core, toolstools, max_iterations8) # 4. 交互循环 print(\nAgent is ready. Type quit or exit to end.) while True: try: user_input input(\nYou: ).strip() if user_input.lower() in [quit, exit, q]: print(Goodbye!) break if not user_input: continue print(\nAgent is thinking...) final_answer await executor.run(user_input) print(f\nAgent: {final_answer}) except KeyboardInterrupt: print(\n\nInterrupted by user.) break except Exception as e: print(f\nAn unexpected error occurred: {e}) if __name__ __main__: asyncio.run(main())4.2 运行与验证首先确保你的.env文件已正确配置 OpenAI API Key。通过 Poetry 运行你的 Agent# 在项目根目录下执行 poetry run python -m src.triple_pi_agent.cli你将看到类似以下的输出Initializing Triple-pi Style Agent... Loaded tools: [calculator] Agent is ready. Type quit or exit to end. You: 请计算 (15 7) * 3 的值。 Agent is thinking... --- Iteration 1 --- Executing Tool: calculator with args: {expression: (15 7) * 3} Tool Result: 66 Agent: 66更复杂的交互示例You: 我有一个问题先计算 2 的 10 次方然后加上 100最后告诉我结果。 Agent is thinking... --- Iteration 1 --- Executing Tool: calculator with args: {expression: 2 ** 10} Tool Result: 1024 --- Iteration 2 --- Executing Tool: calculator with args: {expression: 1024 100} Tool Result: 1124 Agent: 2 的 10 次方是 1024加上 100 后结果是 1124。这个例子展示了 Agent 如何通过多轮思考虽然这里思考过程被简化打印和工具调用来解决一个分步骤的问题。它首先识别出需要计算2 ** 10得到结果后在下一轮循环中将此结果作为上下文再次调用计算器完成加法。5. 工程化深化从可运行到可面试一个能跑的 Demo 距离“写进大厂简历”还差关键的工程化步骤。以下是需要重点加强的方向。5.1 错误处理与鲁棒性当前的实现错误处理比较基础。生产级系统需要LLM 调用重试与退避使用tenacity库为 OpenAI 调用添加指数退避重试处理网络抖动和速率限制。工具执行超时为每个工具执行设置超时防止某个工具挂起整个 Agent。输入验证与清理对所有用户输入和工具参数进行严格的验证和清理防止注入攻击。结构化错误响应定义统一的错误码和错误信息格式便于上游处理。# 示例使用tenacity进行重试 from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from openai import RateLimitError, APIError class RobustOpenAIAgentCore(OpenAIAgentCore): retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10), retryretry_if_exception_type((RateLimitError, APIError)), ) async def generate(self, messages, tools, max_tokens1500): # 在父类方法基础上增加了重试装饰器 return await super().generate(messages, tools, max_tokens)5.2 可观测性与日志大厂项目非常重视可观测性。你需要记录 Agent 的完整决策轨迹。结构化日志使用structlog或logging的 JSON Formatter记录每个循环的输入、输出、工具调用、耗时和错误。追踪Tracing为每个用户会话或任务生成唯一trace_id串联所有相关日志。关键指标记录循环次数、工具调用成功率、LLM 响应时间、Token 消耗等。import logging import uuid from contextvars import ContextVar request_id: ContextVar[str] ContextVar(request_id, default) class AgentExecutorWithLogging(SequentialExecutor): async def run(self, user_input: str) - str: rid str(uuid.uuid4())[:8] request_id.set(rid) logging.info(f[{rid}] Starting agent loop for input: {user_input[:50]}...) start_time asyncio.get_event_loop().time() try: result await super().run(user_input) end_time asyncio.get_event_loop().time() logging.info(f[{rid}] Agent loop finished in {end_time-start_time:.2f}s. Result: {result[:100]}...) return result except Exception as e: logging.error(f[{rid}] Agent loop failed with error: {e}, exc_infoTrue) raise5.3 配置化管理将所有可配置项模型参数、工具列表、循环次数、提示词模板等外置到配置文件如config.yaml或环境变量中。# config.yaml agent: model: gpt-4 temperature: 0.1 max_tokens: 2000 max_iterations: 15 tools: enabled: - calculator - web_search - sql_query web_search: api_key: ${WEB_SEARCH_API_KEY} endpoint: https://api.serper.dev/search logging: level: INFO format: json使用pydantic-settings来管理配置并支持环境变量覆盖。5.4 测试策略编写全面的单元测试和集成测试是工程能力的体现。单元测试测试每个工具类、记忆模块、提示词格式化函数。集成测试模拟 LLM 响应测试整个SequentialExecutor对不同输入的处理流程。端到端测试针对关键用户旅程使用真实但成本低的模型如gpt-3.5-turbo进行测试。# tests/test_calculator.py import pytest from src.triple_pi_agent.tools.calculator import CalculatorTool pytest.mark.asyncio async def test_calculator_simple(): tool CalculatorTool() result await tool.run(2 2) assert result 4 pytest.mark.asyncio async def test_calculator_error(): tool CalculatorTool() result await tool.run(2 / 0) assert error in result.lower()5.5 性能与扩展性考虑异步与并发我们的基础实现已经是异步的。对于需要同时调用多个独立工具的场景可以使用asyncio.gather并发执行。缓存对 LLM 的相同提示词或工具的相同查询结果进行缓存减少开销和延迟。流式输出支持 LLM 思考过程的流式输出提升用户体验。插件化架构使工具和记忆模块可以通过配置文件动态加载无需修改核心代码。6. 常见问题排查与优化在开发和运行过程中你可能会遇到以下典型问题。问题现象可能原因检查与解决思路Agent 陷入无限循环或达到最大迭代次数1. 工具执行结果未能让 LLM 得出最终结论。2. 提示词未明确要求输出Final Answer。3. 工具描述不清LLM 无法正确使用。1. 检查每次工具调用后的Observation是否清晰、完整。2. 强化系统提示词明确要求“当你有足够信息时必须使用Final Answer动作”。3. 优化工具的描述使其目的和输入格式极其明确。LLM 不调用工具直接回答1. 工具描述不够有吸引力或必要性。2. 用户问题太简单LLM 认为自己能直接回答。3. 模型温度 (temperature) 设置过高导致输出随机。1. 在提示词中强调“你必须使用工具来计算或查询信息”。2. 可以接受这是 Agent 的合理行为。如果强制使用工具可在提示词中设置约束。3. 将temperature调低如 0.1。工具调用参数格式错误1. LLM 生成的参数字符串不是有效的 JSON。2. 参数名或类型与工具期望的不匹配。1. 在工具基类的run方法或具体工具的_parse_input中加强解析和容错逻辑。2. 使用args_schema(Pydantic Model) 并让 LLM 知道OpenAI 的tools参数会帮助模型生成合规的 JSON。API 调用超时或报错1. 网络问题。2. OpenAI API 速率限制或服务异常。3. 工具依赖的外部 API 不可用。1. 实现重试机制见 5.1。2. 监控 API 状态设置合理的超时和退避策略。3. 为工具调用添加超时和降级逻辑返回一个友好的错误信息作为 Observation。记忆混乱上下文丢失1. 记忆缓冲区 (max_turns) 设置太小。2. 长对话导致 Token 超限被模型截断。1. 根据任务复杂度调整max_turns。2. 实现更智能的记忆管理如总结之前的对话内容或将重要信息存入长期记忆向量数据库。7. 下一步与简历亮点提炼完成以上步骤后你不仅拥有了一个可运行的 Agent 系统更掌握了一套构建生产级 AI 应用的工程方法论。在简历中你可以这样描述这个项目项目名称基于 Agent Loop 的模块化智能体系统Triple-pi 风格核心职责独立设计并实现了一个具备完整思考-行动循环Agent Loop的智能体系统。架构设计采用清晰的职责分离架构包含 Agent Core、Tools、Memory、Executor 和 Prompt Templates 五大模块支持高内聚低耦合的扩展。工程实践使用 Pydantic 进行强类型数据验证通过 Poetry 管理依赖与环境实现异步执行器以提升 I/O 效率集成结构化日志与请求追踪提升系统可观测性。问题解决设计了安全的工具调用机制如基于 AST 的计算器实现了针对 LLM API 的指数退避重试并制定了防止 Agent 循环失控的策略。效果系统可准确理解多步骤用户指令自主规划并调用相应工具如计算、查询完成任务展示了将大语言模型与确定性逻辑结合解决复杂问题的能力。为了进一步深化你可以尝试以下方向集成更多工具添加网络搜索、数据库查询、文件读写、代码执行等工具构建更强大的 Agent。实现 ReAct 模式显式地在提示词和响应中要求 LLM 输出Thought:、Action:、Observation:使推理过程更透明可控。接入长期记忆使用 ChromaDB、Pinecone 等向量数据库让 Agent 能够记住跨会话的信息。构建 Web 界面使用 FastAPI 或 Gradio 为你的 Agent 提供一个交互式前端。探索多 Agent 协作设计多个具有不同专长的 Agent并通过一个协调器Orchestrator让它们共同解决复杂问题。通过这个从零到一的实践你真正理解了 Agent 如何工作并具备了将其工程化的能力。这远比仅仅调用一个高级框架的 API 更有价值也是面试官希望看到的扎实功底。
基于Agent Loop构建生产级智能体：从Triple-pi项目学习AI工程实践
在实际 AI 工程实践中构建一个能够自主思考、规划并执行任务的智能体Agent是迈向高级应用的关键一步。许多开发者通过跑通一个简单的 Demo 来理解 Agent 的概念但若想将这项技能转化为简历上的亮点尤其是应对大厂对系统设计和工程化能力的高要求就必须深入理解其核心机制——Agent Loop并完成一个结构清晰、可维护、可扩展的工程实践项目。Triple-pi 作为一个在 GitHub 上获得近 8 万星标的知名项目其设计理念和实现为我们提供了一个绝佳的学习范本。它不仅仅是一个玩具项目其模块化设计、清晰的职责分离以及对不同规模模型LLM的适配策略都体现了生产级 AI 应用的思考。本文将带你超越 Demo 阶段以 Triple-pi 项目为蓝本手把手构建一个具备完整 Agent Loop 的系统。我们将从零开始剖析 Agent 的核心工作循环设计项目结构实现关键模块并最终整合成一个可以处理复杂任务的智能体。整个过程将严格遵循工程化实践注重代码结构、配置管理、错误处理和可观测性确保你构建出的不是一个“一次性”脚本而是一个值得写进简历的、具备工业水准的项目。1. 理解 Agent Loop智能体的工作心脏在开始编码之前必须透彻理解驱动智能体的核心引擎——Agent Loop。它决定了 Agent 如何感知、思考、行动并学习是区分简单提示词调用与真正自主系统的关键。1.1 Agent Loop 的基本构成一个典型的 Agent Loop 包含四个核心阶段它们循环往复直至任务完成或达到终止条件观察ObservationAgent 从环境中获取信息。这可能是用户的输入、传感器的数据、数据库的查询结果或者是上一个工具Tool执行的输出。观察为决策提供了上下文。思考ThoughtAgent 的核心处理阶段。它利用大型语言模型LLM分析观察到的信息结合自身的目标和记忆规划下一步行动。思考的输出通常是一个“推理过程”和一个明确的“行动指令”。行动ActionAgent 根据思考阶段产生的指令调用一个特定的工具Tool来执行具体操作。例如执行一段代码、调用一个 API、查询数据库或操作文件系统。反馈Observation’行动执行后环境会给出一个结果成功、失败、返回数据等。这个结果作为新的“观察”被输入到下一个循环的起点。这个观察 - 思考 - 行动 - 反馈的循环会持续进行直到 LLM 在思考阶段判断任务已经完成输出一个最终答案或者触发了预设的停止条件如循环次数超限。1.2 Triple-pi 对 Agent Loop 的工程化抽象Triple-pi 等项目之所以受到推崇是因为它们没有将上述循环写成一个庞杂的while循环而是进行了清晰的职责抽象通常包含以下核心组件Agent 核心Agent Core持有 LLM 的配置和调用逻辑是“思考”的发起者。它接收包含历史记录和当前观察的提示Prompt并生成包含推理和行动指令的响应。工具集Tools一组可被 Agent 调用的函数或接口是“行动”的执行者。每个工具都有明确的名称、描述和参数定义以便 LLM 理解其用途。记忆Memory存储 Agent 与环境的交互历史即过去的观察、思考、行动、反馈。短期记忆如对话轮次和长期记忆如向量数据库共同帮助 Agent 保持上下文连贯性。执行器Executor协调整个循环的组件。它负责组装提示、调用 Agent 核心、解析 LLM 响应、匹配并执行工具、处理工具结果、更新记忆并决定循环是否继续。提示词模板Prompt Templates将当前任务、历史记忆、可用工具列表等信息结构化成 LLM 能高效理解的文本格式。良好的模板设计直接决定 Agent 的推理质量。理解了这个架构我们就知道代码不是从一个庞大的main函数开始而是需要先定义这些组件之间的接口和通信协议。2. 项目环境搭建与核心依赖配置我们选择 Python 作为实现语言因为它拥有最丰富的 AI 开发生态。项目将使用poetry进行依赖管理确保环境隔离和可复现性。如果你使用pip和venv原理是相通的。2.1 初始化项目结构首先创建一个清晰的项目目录结构。好的结构是工程实践的第一步。mkdir triple-pi-agent-project cd triple-pi-agent-project使用poetry初始化项目如果没有安装poetry可以使用pip install poetry。poetry new . # 这会在当前目录生成 pyproject.toml 等文件但我们需要调整结构。 # 我们可以手动创建更符合我们需求的结构。推荐的项目结构如下triple-pi-agent-project/ ├── pyproject.toml # 项目依赖和配置 ├── README.md ├── .env.example # 环境变量示例 ├── src/ │ └── triple_pi_agent/ │ ├── __init__.py │ ├── agent/ # Agent核心模块 │ │ ├── __init__.py │ │ ├── core.py # Agent核心类 │ │ └── prompts.py # 提示词模板 │ ├── tools/ # 工具模块 │ │ ├── __init__.py │ │ ├── base.py # 工具基类 │ │ ├── calculator.py │ │ ├── web_search.py │ │ └── ... │ ├── memory/ # 记忆模块 │ │ ├── __init__.py │ │ └── buffer.py # 对话缓冲记忆 │ ├── executor/ # 执行器模块 │ │ ├── __init__.py │ │ └── sequential.py # 顺序执行器 │ ├── schemas/ # Pydantic数据模型 │ │ ├── __init__.py │ │ └── agent.py # 定义Agent输入输出格式 │ └── cli.py # 命令行入口 ├── tests/ # 单元测试 │ ├── __init__.py │ ├── test_agent.py │ └── test_tools.py └── scripts/ # 辅助脚本 └── run_agent.py2.2 配置核心依赖编辑pyproject.toml文件添加必要的依赖。我们将使用langchain社区的核心抽象作为参考但会强调自己的实现以加深理解。同时我们会集成OpenAI的 API 作为 LLM 后端。[tool.poetry] name triple-pi-agent version 0.1.0 description A practical agent system with full agent loop implementation. authors [Your Name your.emailexample.com] [tool.poetry.dependencies] python ^3.9 openai ^1.3.0 # 用于调用GPT系列模型 pydantic ^2.0.0 # 用于数据验证和设置管理 tenacity ^8.2.0 # 用于重试逻辑 requests ^2.31.0 # 用于工具中调用HTTP API python-dotenv ^1.0.0 # 用于加载环境变量 # 可选如果你计划使用本地模型可以添加 transformers, torch 等 [tool.poetry.group.dev.dependencies] pytest ^7.4.0 black ^23.0.0 isort ^5.12.0 mypy ^1.5.0 [build-system] requires [poetry-core] build-backend poetry.core.masonry.api安装依赖poetry install2.3 设置环境变量创建.env文件请确保将其加入.gitignore用于存储敏感信息如 API Key。# .env OPENAI_API_KEYsk-your-openai-api-key-here OPENAI_BASE_URLhttps://api.openai.com/v1 # 如果使用Azure或代理可修改 MODEL_NAMEgpt-3.5-turbo # 或 gpt-4, gpt-4-turbo-preview在代码中我们使用python-dotenv来加载这些配置。3. 实现核心组件从工具到执行器现在我们开始自底向上地实现 Agent 系统的各个组件。我们将遵循“定义接口再实现具体类”的原则。3.1 定义工具基类与具体工具在src/triple_pi_agent/tools/base.py中我们定义一个所有工具都必须遵守的协议基类。from abc import ABC, abstractmethod from typing import Any, Dict, Optional from pydantic import BaseModel, Field class ToolInput(BaseModel): 工具调用的输入参数模型。 # 这是一个示例具体参数由工具定义 argument: str Field(descriptionThe input argument for the tool.) class BaseTool(ABC): 工具基类。每个工具必须有名称、描述和调用方法。 name: str description: str args_schema: Optional[type[BaseModel]] None def __init__(self, name: str, description: str): self.name name self.description description abstractmethod async def _arun(self, **kwargs: Any) - str: 异步执行工具的核心逻辑。 pass async def run(self, tool_input: str, **kwargs: Any) - str: 对外暴露的run方法。可以在这里添加输入验证、错误处理、日志等。 tool_input: 通常是LLM解析出的参数字符串需要进一步处理。 # 这里可以添加输入解析逻辑例如将JSON字符串解析为字典 # 为了简化我们假设工具输入已经是合适的格式或由子类处理 try: result await self._arun(**self._parse_input(tool_input)) return result except Exception as e: return fError executing tool {self.name}: {str(e)} def _parse_input(self, tool_input: str) - Dict[str, Any]: 简单的输入解析子类可重写。默认尝试解析JSON否则作为单个参数。 import json try: return json.loads(tool_input) except json.JSONDecodeError: # 如果不是JSON假设整个字符串是单个参数 return {input: tool_input} def __str__(self): return f{self.name}: {self.description}接下来实现一个具体的计算器工具src/triple_pi_agent/tools/calculator.py。import math import ast import operator as op from typing import Dict, Any from .base import BaseTool, ToolInput # 支持的安全运算符 _SAFE_OPERATORS { ast.Add: op.add, ast.Sub: op.sub, ast.Mult: op.mul, ast.Div: op.truediv, ast.Pow: op.pow, ast.USub: op.neg, ast.Mod: op.mod, } class CalculatorToolInput(ToolInput): expression: str Field(descriptionA mathematical expression to evaluate, e.g., 2 3 * 4.) class CalculatorTool(BaseTool): A tool to evaluate mathematical expressions. Uses ast for safety. def __init__(self): super().__init__( namecalculator, descriptionUseful for performing arithmetic calculations. Input should be a valid mathematical expression. ) self.args_schema CalculatorToolInput async def _arun(self, expression: str, **kwargs: Any) - str: 安全地计算数学表达式。 try: # 使用抽象语法树进行安全评估避免eval的直接风险 node ast.parse(expression, modeeval).body result self._eval_node(node) return str(result) except (SyntaxError, TypeError, KeyError, ZeroDivisionError) as e: return fCalculation error: {str(e)}. Please check your expression: {expression} def _eval_node(self, node): 递归评估AST节点。 if isinstance(node, ast.Num): # Python 3.8 return node.n elif isinstance(node, ast.Constant): # Python 3.8 return node.value elif isinstance(node, ast.BinOp): left_val self._eval_node(node.left) right_val self._eval_node(node.right) operator_func _SAFE_OPERATORS.get(type(node.op)) if operator_func is None: raise TypeError(fUnsupported operator: {type(node.op)}) return operator_func(left_val, right_val) elif isinstance(node, ast.UnaryOp): operand_val self._eval_node(node.operand) operator_func _SAFE_OPERATORS.get(type(node.op)) if operator_func is None: raise TypeError(fUnsupported unary operator: {type(node.op)}) return operator_func(operand_val) else: raise TypeError(fUnsupported AST node type: {type(node)}) def _parse_input(self, tool_input: str) - Dict[str, Any]: # 计算器工具期望一个表达式字符串 return {expression: tool_input.strip()}3.2 构建记忆模块我们实现一个简单的对话缓冲记忆用于保存最近的交互历史。更复杂的记忆如向量存储可以在此基础上扩展。src/triple_pi_agent/memory/buffer.py:from typing import List, Dict, Any from pydantic import BaseModel class Message(BaseModel): 代表Agent循环中的一条消息。 role: str # 例如user, assistant, tool content: str class ConversationBufferMemory: 存储最近的对话历史。 def __init__(self, max_turns: int 10): self.messages: List[Message] [] self.max_turns max_turns def add_message(self, role: str, content: str): 添加一条消息。 self.messages.append(Message(rolerole, contentcontent)) # 如果超过最大轮次移除最早的消息可以根据策略调整 if len(self.messages) self.max_turns * 2: # 粗略估计一轮包含user和assistant self.messages self.messages[-self.max_turns * 2:] def get_messages(self) - List[Dict[str, str]]: 获取格式化后的消息列表用于构造LLM提示。 return [{role: msg.role, content: msg.content} for msg in self.messages] def clear(self): 清空记忆。 self.messages []3.3 设计 Agent 核心与提示词模板Agent 核心负责与 LLM 交互。我们使用 OpenAI 的官方客户端。src/triple_pi_agent/agent/core.py:import os from typing import List, Dict, Any, Optional from openai import AsyncOpenAI from dotenv import load_dotenv from ..schemas.agent import AgentResponse, ToolCall import json load_dotenv() class OpenAIAgentCore: 基于OpenAI API的Agent核心。 def __init__( self, model: Optional[str] None, api_key: Optional[str] None, base_url: Optional[str] None, temperature: float 0.1, # 低温度使输出更确定适合工具调用 ): self.model model or os.getenv(MODEL_NAME, gpt-3.5-turbo) self.client AsyncOpenAI( api_keyapi_key or os.getenv(OPENAI_API_KEY), base_urlbase_url or os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1), ) self.temperature temperature async def generate( self, messages: List[Dict[str, str]], tools: List[Dict[str, Any]], # OpenAI格式的工具描述列表 max_tokens: int 1500, ) - AgentResponse: 调用LLM生成包含思考和工具调用的响应。 返回结构化的AgentResponse。 try: response await self.client.chat.completions.create( modelself.model, messagesmessages, toolstools, tool_choiceauto, # 让模型决定是否调用工具 temperatureself.temperature, max_tokensmax_tokens, ) message response.choices[0].message # 解析响应 final_answer None tool_calls [] if message.content: # 模型直接给出了最终答案 final_answer message.content.strip() elif message.tool_calls: # 模型要求调用工具 for tc in message.tool_calls: tool_calls.append( ToolCall( idtc.id, nametc.function.name, argumentstc.function.arguments ) ) return AgentResponse( thoughtmessage.content or fDecided to call tool(s): {[tc.name for tc in tool_calls]}, final_answerfinal_answer, tool_callstool_calls, ) except Exception as e: # 在实际项目中这里应该有更细致的错误处理和重试逻辑 return AgentResponse( thoughtfError during LLM call: {str(e)}, final_answerNone, tool_calls[], )对应的数据模型定义在src/triple_pi_agent/schemas/agent.py:from pydantic import BaseModel from typing import Optional, List class ToolCall(BaseModel): id: str name: str arguments: str # JSON格式的参数字符串 class AgentResponse(BaseModel): thought: str # 模型的推理过程或直接输出 final_answer: Optional[str] None # 如果是最终答案则存在 tool_calls: List[ToolCall] [] # 如果需要调用工具则存在提示词模板对于引导 LLM 按照 Agent 格式思考至关重要。src/triple_pi_agent/agent/prompts.py:def get_system_prompt(tools_descriptions: str) - str: 生成系统提示词定义Agent的角色和行为规范。 return fYou are a helpful and precise assistant. You have access to the following tools: {tools_descriptions} To solve a users task, you must ALWAYS follow this format: Thought: You should always think about what to do. Explain your reasoning step by step. If you need to use a tool, state which one and why. Action: The action to take, must be one of the tool names listed above, or Final Answer if you have the answer. Action Input: The input to the action, formatted as a JSON string if the tool requires it. If action is Final Answer, this is your final response to the user. After an Action that is a tool call, you will receive an Observation, which is the result of that tool. Then you must continue with another Thought, Action, Action Input cycle until the task is solved. Begin! def format_tool_for_prompt(tool) - str: 将工具对象格式化为LLM可理解的描述文本。 args_desc if tool.args_schema: # 这里简化处理实际可以解析Pydantic模型的schema args_desc f It takes input that should be a JSON object with the required arguments. return f- {tool.name}: {tool.description}{args_desc}3.4 实现执行器串联整个循环执行器是 Agent Loop 的调度中心。我们实现一个顺序执行器。src/triple_pi_agent/executor/sequential.py:import asyncio import json from typing import List, Optional from ..agent.core import OpenAIAgentCore from ..tools.base import BaseTool from ..memory.buffer import ConversationBufferMemory from ..agent.prompts import get_system_prompt, format_tool_for_prompt from ..schemas.agent import AgentResponse class SequentialExecutor: 顺序执行Agent循环。 def __init__( self, agent_core: OpenAIAgentCore, tools: List[BaseTool], memory: Optional[ConversationBufferMemory] None, max_iterations: int 10, ): self.agent_core agent_core self.tools {tool.name: tool for tool in tools} self.memory memory or ConversationBufferMemory() self.max_iterations max_iterations # 初始化系统提示 tools_description \n.join([format_tool_for_prompt(tool) for tool in tools]) self.system_prompt get_system_prompt(tools_description) async def run(self, user_input: str) - str: 执行Agent循环处理用户输入返回最终答案。 self.memory.clear() self.memory.add_message(system, self.system_prompt) self.memory.add_message(user, user_input) iteration 0 final_answer None while iteration self.max_iterations and final_answer is None: iteration 1 print(f\n--- Iteration {iteration} ---) # 1. 观察从记忆构建消息列表 messages_for_llm self.memory.get_messages() # 2. 思考 决策调用Agent核心 openai_tools_format [ { type: function, function: { name: tool.name, description: tool.description, parameters: tool.args_schema.schema() if tool.args_schema else {type: object, properties: {}}, } } for tool in self.tools.values() ] agent_response: AgentResponse await self.agent_core.generate( messagesmessages_for_llm, toolsopenai_tools_format, ) # 将模型的“思考”加入记忆 self.memory.add_message(assistant, fThought: {agent_response.thought}) # 3. 行动判断是最终答案还是工具调用 if agent_response.final_answer: final_answer agent_response.final_answer self.memory.add_message(assistant, fFinal Answer: {final_answer}) break elif agent_response.tool_calls: for tool_call in agent_response.tool_calls: tool_name tool_call.name tool_args tool_call.arguments # 记录行动 action_message fAction: {tool_name}\nAction Input: {tool_args} self.memory.add_message(assistant, action_message) print(fExecuting Tool: {tool_name} with args: {tool_args}) # 执行工具 tool self.tools.get(tool_name) if not tool: observation fError: Tool {tool_name} not found. else: try: observation await tool.run(tool_args) except Exception as e: observation fError executing {tool_name}: {str(e)} # 4. 反馈将观察结果加入记忆 print(fTool Result: {observation[:100]}...) # 打印部分结果 self.memory.add_message(tool, fObservation: {observation}) else: # 模型既没有给出答案也没有调用工具可能出错了 observation Error: Model response did not contain a final answer or tool calls. self.memory.add_message(system, observation) final_answer I encountered an error in processing. Please try again. break if final_answer is None: final_answer fReached maximum iterations ({self.max_iterations}) without a final answer. return final_answer4. 组装与运行构建你的第一个智能体所有核心组件都已就绪现在我们将它们组装起来并创建一个命令行接口来交互。4.1 创建主程序入口src/triple_pi_agent/cli.py:import asyncio import sys from .agent.core import OpenAIAgentCore from .tools.calculator import CalculatorTool from .executor.sequential import SequentialExecutor async def main(): 主函数初始化并运行Agent。 print(Initializing Triple-pi Style Agent...) # 1. 初始化Agent核心 agent_core OpenAIAgentCore() # 2. 注册工具 tools [ CalculatorTool(), # 未来可以在这里添加更多工具如 WebSearchTool(), DBTool() 等 ] print(fLoaded tools: {[tool.name for tool in tools]}) # 3. 初始化执行器 executor SequentialExecutor(agent_coreagent_core, toolstools, max_iterations8) # 4. 交互循环 print(\nAgent is ready. Type quit or exit to end.) while True: try: user_input input(\nYou: ).strip() if user_input.lower() in [quit, exit, q]: print(Goodbye!) break if not user_input: continue print(\nAgent is thinking...) final_answer await executor.run(user_input) print(f\nAgent: {final_answer}) except KeyboardInterrupt: print(\n\nInterrupted by user.) break except Exception as e: print(f\nAn unexpected error occurred: {e}) if __name__ __main__: asyncio.run(main())4.2 运行与验证首先确保你的.env文件已正确配置 OpenAI API Key。通过 Poetry 运行你的 Agent# 在项目根目录下执行 poetry run python -m src.triple_pi_agent.cli你将看到类似以下的输出Initializing Triple-pi Style Agent... Loaded tools: [calculator] Agent is ready. Type quit or exit to end. You: 请计算 (15 7) * 3 的值。 Agent is thinking... --- Iteration 1 --- Executing Tool: calculator with args: {expression: (15 7) * 3} Tool Result: 66 Agent: 66更复杂的交互示例You: 我有一个问题先计算 2 的 10 次方然后加上 100最后告诉我结果。 Agent is thinking... --- Iteration 1 --- Executing Tool: calculator with args: {expression: 2 ** 10} Tool Result: 1024 --- Iteration 2 --- Executing Tool: calculator with args: {expression: 1024 100} Tool Result: 1124 Agent: 2 的 10 次方是 1024加上 100 后结果是 1124。这个例子展示了 Agent 如何通过多轮思考虽然这里思考过程被简化打印和工具调用来解决一个分步骤的问题。它首先识别出需要计算2 ** 10得到结果后在下一轮循环中将此结果作为上下文再次调用计算器完成加法。5. 工程化深化从可运行到可面试一个能跑的 Demo 距离“写进大厂简历”还差关键的工程化步骤。以下是需要重点加强的方向。5.1 错误处理与鲁棒性当前的实现错误处理比较基础。生产级系统需要LLM 调用重试与退避使用tenacity库为 OpenAI 调用添加指数退避重试处理网络抖动和速率限制。工具执行超时为每个工具执行设置超时防止某个工具挂起整个 Agent。输入验证与清理对所有用户输入和工具参数进行严格的验证和清理防止注入攻击。结构化错误响应定义统一的错误码和错误信息格式便于上游处理。# 示例使用tenacity进行重试 from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from openai import RateLimitError, APIError class RobustOpenAIAgentCore(OpenAIAgentCore): retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10), retryretry_if_exception_type((RateLimitError, APIError)), ) async def generate(self, messages, tools, max_tokens1500): # 在父类方法基础上增加了重试装饰器 return await super().generate(messages, tools, max_tokens)5.2 可观测性与日志大厂项目非常重视可观测性。你需要记录 Agent 的完整决策轨迹。结构化日志使用structlog或logging的 JSON Formatter记录每个循环的输入、输出、工具调用、耗时和错误。追踪Tracing为每个用户会话或任务生成唯一trace_id串联所有相关日志。关键指标记录循环次数、工具调用成功率、LLM 响应时间、Token 消耗等。import logging import uuid from contextvars import ContextVar request_id: ContextVar[str] ContextVar(request_id, default) class AgentExecutorWithLogging(SequentialExecutor): async def run(self, user_input: str) - str: rid str(uuid.uuid4())[:8] request_id.set(rid) logging.info(f[{rid}] Starting agent loop for input: {user_input[:50]}...) start_time asyncio.get_event_loop().time() try: result await super().run(user_input) end_time asyncio.get_event_loop().time() logging.info(f[{rid}] Agent loop finished in {end_time-start_time:.2f}s. Result: {result[:100]}...) return result except Exception as e: logging.error(f[{rid}] Agent loop failed with error: {e}, exc_infoTrue) raise5.3 配置化管理将所有可配置项模型参数、工具列表、循环次数、提示词模板等外置到配置文件如config.yaml或环境变量中。# config.yaml agent: model: gpt-4 temperature: 0.1 max_tokens: 2000 max_iterations: 15 tools: enabled: - calculator - web_search - sql_query web_search: api_key: ${WEB_SEARCH_API_KEY} endpoint: https://api.serper.dev/search logging: level: INFO format: json使用pydantic-settings来管理配置并支持环境变量覆盖。5.4 测试策略编写全面的单元测试和集成测试是工程能力的体现。单元测试测试每个工具类、记忆模块、提示词格式化函数。集成测试模拟 LLM 响应测试整个SequentialExecutor对不同输入的处理流程。端到端测试针对关键用户旅程使用真实但成本低的模型如gpt-3.5-turbo进行测试。# tests/test_calculator.py import pytest from src.triple_pi_agent.tools.calculator import CalculatorTool pytest.mark.asyncio async def test_calculator_simple(): tool CalculatorTool() result await tool.run(2 2) assert result 4 pytest.mark.asyncio async def test_calculator_error(): tool CalculatorTool() result await tool.run(2 / 0) assert error in result.lower()5.5 性能与扩展性考虑异步与并发我们的基础实现已经是异步的。对于需要同时调用多个独立工具的场景可以使用asyncio.gather并发执行。缓存对 LLM 的相同提示词或工具的相同查询结果进行缓存减少开销和延迟。流式输出支持 LLM 思考过程的流式输出提升用户体验。插件化架构使工具和记忆模块可以通过配置文件动态加载无需修改核心代码。6. 常见问题排查与优化在开发和运行过程中你可能会遇到以下典型问题。问题现象可能原因检查与解决思路Agent 陷入无限循环或达到最大迭代次数1. 工具执行结果未能让 LLM 得出最终结论。2. 提示词未明确要求输出Final Answer。3. 工具描述不清LLM 无法正确使用。1. 检查每次工具调用后的Observation是否清晰、完整。2. 强化系统提示词明确要求“当你有足够信息时必须使用Final Answer动作”。3. 优化工具的描述使其目的和输入格式极其明确。LLM 不调用工具直接回答1. 工具描述不够有吸引力或必要性。2. 用户问题太简单LLM 认为自己能直接回答。3. 模型温度 (temperature) 设置过高导致输出随机。1. 在提示词中强调“你必须使用工具来计算或查询信息”。2. 可以接受这是 Agent 的合理行为。如果强制使用工具可在提示词中设置约束。3. 将temperature调低如 0.1。工具调用参数格式错误1. LLM 生成的参数字符串不是有效的 JSON。2. 参数名或类型与工具期望的不匹配。1. 在工具基类的run方法或具体工具的_parse_input中加强解析和容错逻辑。2. 使用args_schema(Pydantic Model) 并让 LLM 知道OpenAI 的tools参数会帮助模型生成合规的 JSON。API 调用超时或报错1. 网络问题。2. OpenAI API 速率限制或服务异常。3. 工具依赖的外部 API 不可用。1. 实现重试机制见 5.1。2. 监控 API 状态设置合理的超时和退避策略。3. 为工具调用添加超时和降级逻辑返回一个友好的错误信息作为 Observation。记忆混乱上下文丢失1. 记忆缓冲区 (max_turns) 设置太小。2. 长对话导致 Token 超限被模型截断。1. 根据任务复杂度调整max_turns。2. 实现更智能的记忆管理如总结之前的对话内容或将重要信息存入长期记忆向量数据库。7. 下一步与简历亮点提炼完成以上步骤后你不仅拥有了一个可运行的 Agent 系统更掌握了一套构建生产级 AI 应用的工程方法论。在简历中你可以这样描述这个项目项目名称基于 Agent Loop 的模块化智能体系统Triple-pi 风格核心职责独立设计并实现了一个具备完整思考-行动循环Agent Loop的智能体系统。架构设计采用清晰的职责分离架构包含 Agent Core、Tools、Memory、Executor 和 Prompt Templates 五大模块支持高内聚低耦合的扩展。工程实践使用 Pydantic 进行强类型数据验证通过 Poetry 管理依赖与环境实现异步执行器以提升 I/O 效率集成结构化日志与请求追踪提升系统可观测性。问题解决设计了安全的工具调用机制如基于 AST 的计算器实现了针对 LLM API 的指数退避重试并制定了防止 Agent 循环失控的策略。效果系统可准确理解多步骤用户指令自主规划并调用相应工具如计算、查询完成任务展示了将大语言模型与确定性逻辑结合解决复杂问题的能力。为了进一步深化你可以尝试以下方向集成更多工具添加网络搜索、数据库查询、文件读写、代码执行等工具构建更强大的 Agent。实现 ReAct 模式显式地在提示词和响应中要求 LLM 输出Thought:、Action:、Observation:使推理过程更透明可控。接入长期记忆使用 ChromaDB、Pinecone 等向量数据库让 Agent 能够记住跨会话的信息。构建 Web 界面使用 FastAPI 或 Gradio 为你的 Agent 提供一个交互式前端。探索多 Agent 协作设计多个具有不同专长的 Agent并通过一个协调器Orchestrator让它们共同解决复杂问题。通过这个从零到一的实践你真正理解了 Agent 如何工作并具备了将其工程化的能力。这远比仅仅调用一个高级框架的 API 更有价值也是面试官希望看到的扎实功底。
