从零搭建AI Agent工具链：掌握智能体核心架构与工程实践

在业务迭代中引入AI能力时，直接调用大模型API往往只能完成简单的问答。当我们需要一个能理解复杂意图、自主调用工具、并持续完成多步任务的“智能助手”时，就触及了AI Agent（智能体）的领域。然而，市面上的Agent平台虽多，但要么封装过深难以定制，要么过于零散不成体系。想要真正掌握其核心并灵活应用于自身业务，从零开始搭建一套属于自己的智能体工具链，是理解其运作机制、掌控其能力边界的最佳路径。

本文将带你从第一性原理出发，亲手搭建一套可运行、可扩展的智能体工具链。我们将从最基础的概念拆解开始，逐步完成环境搭建、核心组件开发、工具链集成、任务编排与持久化，最终构建一个能处理“查询天气并总结”的完整智能体。无论你是想深入AI应用开发的工程师，还是希望将Agent能力融入现有系统的架构师，这套从零开始的实践指南都能为你提供清晰的路线图和可复用的代码。

1. 智能体（Agent）核心概念与架构拆解

在开始动手之前，我们必须厘清几个关键概念，这有助于理解我们即将构建的每一个组件。

1.1 什么是AI Agent？

AI Agent（智能体）不是一个单一模型，而是一个系统。它通常由大型语言模型（LLM）作为“大脑”，辅以任务规划、工具调用、记忆管理和执行控制等模块构成。其核心目标是接收用户以自然语言表述的复杂指令，通过感知（理解指令）、规划（拆解步骤）、行动（调用工具）、观察（评估结果）的循环，最终自主完成目标。

与单纯的大模型对话相比，Agent的核心特征在于：

• 自主性：能根据目标自发规划步骤，而非仅回答单轮问题。
• 工具使用：可以调用外部API、数据库、函数等扩展能力。
• 持续性：拥有短期或长期记忆，能在多轮对话中保持上下文和目标。

1.2 智能体工具链的组成部分

一套完整的智能体工具链，通常包含以下核心层，这也是我们本文的构建蓝图：

1. 编排层（Orchestrator）：智能体的调度中枢。负责理解用户意图，管理任务执行流程（规划-行动-观察循环），并协调各个模块工作。
2. 模型层（Model Layer）：提供核心认知能力。即我们所使用的大语言模型（如GPT、Claude、国产大模型等），通过API或本地部署进行调用。
3. 工具层（Tools Layer）：智能体的“手脚”。是一系列可供Agent调用的函数或API，例如：搜索引擎、计算器、数据库查询、代码执行器等。工具需要被标准化描述，以便Agent理解其功能。
4. 记忆层（Memory Layer）：智能体的“经验”。负责存储和检索对话历史、任务上下文、执行结果等，分为短期记忆（当前会话）和长期记忆（向量数据库等）。
5. 评估与持久化层（Evaluation & Persistence）：监控Agent运行状态，记录执行日志，并将关键数据（如对话、工具调用记录）持久化到数据库，便于分析和复盘。

1.3 主流框架与自建的意义

当前社区有诸多优秀的Agent框架，如LangChain、LlamaIndex、Semantic Kernel等。它们提供了高层次的抽象，能快速搭建原型。然而，自建工具链的意义在于：

• 深度掌控：理解每个环节的数据流与控制逻辑，便于深度定制和优化。
• 轻量灵活：避免引入庞大框架的冗余依赖，更适合嵌入现有系统或资源受限场景。
• 学习价值：是理解Agent技术本质的最佳实践。

接下来，我们将使用Python作为主要语言，从零开始实现上述每一层。

2. 环境准备与项目初始化

我们选择Python生态，因为它拥有丰富的AI库和灵活的胶水能力。请确保你的开发环境已就绪。

2.1 基础环境要求

• 操作系统：Windows 10/11, macOS, 或 Linux (Ubuntu 20.04+ 推荐)
• Python版本：Python 3.9 或 3.10（3.11+也可，注意某些库的兼容性）
• 包管理工具：pip (建议使用虚拟环境)

2.2 创建项目与虚拟环境

首先，创建一个干净的项目目录并初始化虚拟环境。

# 创建项目目录 mkdir hardcore-agent-toolchain cd hardcore-agent-toolchain # 创建虚拟环境 (以venv为例) python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/macOS: source venv/bin/activate

2.3 安装核心依赖

我们将分步安装所需的库。创建一个requirements.txt文件，内容如下：

# 核心HTTP请求与JSON处理 requests>=2.28.0 httpx>=0.24.0 # 大模型API调用 (以OpenAI格式API为例，可适配其他模型) openai>=1.0.0 # 向量数据库与嵌入 (用于高级记忆功能) chromadb>=0.4.0 sentence-transformers>=2.2.0 # 用于本地生成嵌入 # 任务队列与异步处理 (可选，用于复杂任务流) celery>=5.3.0 # 或使用 asyncio # 数据持久化 sqlalchemy>=2.0.0 alembic>=1.12.0 # 开发与工具 pydantic>=2.0.0 # 数据验证与设置管理 python-dotenv>=1.0.0 # 环境变量管理 loguru>=0.7.0 # 日志记录

使用pip安装所有依赖：

pip install -r requirements.txt

如果你的网络环境访问OpenAI服务存在困难，可以选择支持OpenAI API兼容接口的国内大模型平台（如DeepSeek、智谱AI、月之暗面等），只需调整API Base URL和API Key即可，代码结构基本不变。本文示例将采用OpenAI API格式进行演示，并会说明适配点。

2.4 项目结构设计

一个清晰的项目结构是工程化的开端。我们的项目目录将如下组织：

hardcore-agent-toolchain/ ├── .env # 环境变量配置文件（需加入.gitignore） ├── requirements.txt # 项目依赖 ├── README.md # 项目说明 ├── main.py # 应用主入口 ├── config/ # 配置模块 │ ├── __init__.py │ └── settings.py # 应用配置（从环境变量加载） ├── core/ # 核心Agent逻辑 │ ├── __init__.py │ ├── agent.py # Agent编排器主类 │ ├── models.py # 数据模型（消息、工具等） │ └── memory/ # 记忆模块 │ ├── __init__.py │ ├── base.py # 记忆基类 │ └── simple.py # 简单内存实现 ├── tools/ # 工具层 │ ├── __init__.py │ ├── base.py # 工具基类 │ ├── calculator.py # 计算器工具示例 │ ├── weather.py # 天气查询工具示例 │ └── registry.py # 工具注册与管理 ├── llm/ # 大模型层 │ ├── __init__.py │ └── client.py # 统一的大模型客户端 ├── persistence/ # 持久化层 │ ├── __init__.py │ ├── database.py # 数据库连接与模型定义 │ └── models.py # SQLAlchemy ORM 模型 └── utils/ # 工具函数 ├── __init__.py └── logging.py # 日志配置

现在，基础环境与项目骨架已准备完毕。让我们开始编写第一个核心模块。

3. 核心模块开发：从模型层到工具层

我们将采用自底向上的方式，先构建稳定的基础组件，再组装成完整的Agent。

3.1 统一大模型客户端 (llm/client.py)

为了兼容不同的大模型提供商，我们设计一个统一的客户端。它负责封装API调用、处理错误、格式化消息。

# llm/client.py import os from typing import List, Dict, Any, Optional from openai import OpenAI from pydantic import BaseModel, Field from loguru import logger from config.settings import settings class Message(BaseModel): """对话消息模型""" role: str # 'system', 'user', 'assistant', 'tool' content: str class LLMClient: """统一的大语言模型客户端""" def __init__(self, api_key: Optional[str] = None, base_url: Optional[str] = None): self.api_key = api_key or settings.LLM_API_KEY self.base_url = base_url or settings.LLM_BASE_URL self.model = settings.LLM_MODEL # 初始化OpenAI客户端（兼容其他提供者） self.client = OpenAI( api_key=self.api_key, base_url=self.base_url # 例如: "https://api.openai.com/v1" 或国内平台兼容地址 ) logger.info(f"LLM Client initialized with model: {self.model}") def chat_completion(self, messages: List[Dict[str, str]], **kwargs) -> str: """ 调用聊天补全API Args: messages: 消息列表，格式 [{"role": "user", "content": "Hello"}] **kwargs: 其他API参数，如temperature, max_tokens Returns: 模型返回的文本内容 """ try: response = self.client.chat.completions.create( model=self.model, messages=messages, temperature=kwargs.get('temperature', 0.7), max_tokens=kwargs.get('max_tokens', 1000), **{k: v for k, v in kwargs.items() if k not in ['temperature', 'max_tokens']} ) content = response.choices[0].message.content logger.debug(f"LLM Response: {content[:200]}...") # 日志截断 return content.strip() except Exception as e: logger.error(f"LLM API call failed: {e}") # 在实际项目中，这里应实现重试、降级等策略 raise def structured_output(self, messages: List[Dict], response_format: Any) -> Any: """ 获取结构化输出（如果模型支持，如GPT-4o）。 这是一个高级功能示例，基础版可先使用文本解析。 """ # 简化实现：先获取文本，再尝试解析（例如JSON） text_response = self.chat_completion(messages) # 这里可以添加JSON解析逻辑，或使用模型的function calling/JSON mode return text_response

对应的配置文件config/settings.py用于管理所有环境变量：

# config/settings.py import os from pydantic_settings import BaseSettings from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 class Settings(BaseSettings): """应用配置""" # LLM 配置 LLM_API_KEY: str = os.getenv("LLM_API_KEY", "") LLM_BASE_URL: str = os.getenv("LLM_BASE_URL", "https://api.openai.com/v1") LLM_MODEL: str = os.getenv("LLM_MODEL", "gpt-3.5-turbo") # 数据库配置 (示例，用于持久化) DATABASE_URL: str = os.getenv("DATABASE_URL", "sqlite:///./agent.db") # 日志级别 LOG_LEVEL: str = os.getenv("LOG_LEVEL", "INFO") class Config: env_file = ".env" settings = Settings()

在项目根目录创建.env文件（切勿提交到版本控制）：

# .env LLM_API_KEY=your_api_key_here # 如果使用国内兼容API的平台，修改以下两项 # LLM_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1 # LLM_MODEL=qwen-plus LLM_MODEL=gpt-3.5-turbo LOG_LEVEL=DEBUG

3.2 工具层设计与实现 (tools/)

工具是Agent能力的扩展。每个工具都需要一个清晰的描述，以便LLM理解何时以及如何调用它。

首先，定义工具基类：

# tools/base.py from abc import ABC, abstractmethod from typing import Any, Dict from pydantic import BaseModel, Field class ToolSchema(BaseModel): """工具的模式定义，用于描述工具""" name: str = Field(..., description="工具的唯一名称") description: str = Field(..., description="工具功能的详细描述") parameters: Dict[str, Any] = Field(default_factory=dict, description="工具参数的JSON Schema") class BaseTool(ABC): """所有工具的抽象基类""" def __init__(self): self.schema = self._define_schema() @abstractmethod def _define_schema(self) -> ToolSchema: """定义工具的schema，必须由子类实现""" pass @abstractmethod def execute(self, **kwargs) -> str: """执行工具的核心逻辑，必须由子类实现""" pass def __call__(self, **kwargs) -> str: """使工具可调用""" return self.execute(**kwargs) @property def name(self) -> str: return self.schema.name @property def description(self) -> str: return self.schema.description

接着，实现两个具体的工具示例：计算器和天气查询。

# tools/calculator.py import math from typing import Dict, Any from tools.base import BaseTool, ToolSchema class CalculatorTool(BaseTool): """一个简单的计算器工具，能执行基础数学运算""" def _define_schema(self) -> ToolSchema: return ToolSchema( name="calculator", description="执行数学计算。支持加(+)、减(-)、乘(*)、除(/)、乘方(**)等运算。", parameters={ "type": "object", "properties": { "expression": { "type": "string", "description": "数学表达式，例如：'3 + 5 * 2' 或 'sqrt(16)'" } }, "required": ["expression"] } ) def execute(self, **kwargs) -> str: expression = kwargs.get("expression", "") if not expression: return "错误：未提供表达式。" # 安全警告：在生产环境中，直接eval是危险的！ # 这里仅为演示，实际应使用安全的表达式解析库（如 ast.literal_eval, pyparsing） # 或严格限制允许的运算符和函数。 try: # 替换一些常用数学函数和常量 safe_dict = {"__builtins__": None} safe_dict.update(math.__dict__) # 仅允许部分安全的数学函数 allowed_names = {k: v for k, v in math.__dict__.items() if not k.startswith('_')} safe_dict.update(allowed_names) # 非常简化的安全评估 - 仅用于演示，不适用于生产！ # 生产环境请使用：ast.literal_eval 或 自定义解析器 result = eval(expression, {"__builtins__": {}}, safe_dict) return f"计算结果：{expression} = {result}" except Exception as e: return f"计算错误：无法解析表达式 '{expression}'。错误信息：{e}"

# tools/weather.py import requests from typing import Dict, Any from tools.base import BaseTool, ToolSchema from config.settings import settings class WeatherTool(BaseTool): """查询城市天气的工具（使用模拟API）""" def _define_schema(self) -> ToolSchema: return ToolSchema( name="get_weather", description="获取指定城市的当前天气信息。", parameters={ "type": "object", "properties": { "city": { "type": "string", "description": "城市名称，例如：'北京'、'Shanghai'" } }, "required": ["city"] } ) def execute(self, **kwargs) -> str: city = kwargs.get("city", "") if not city: return "错误：未提供城市名称。" # 注意：这里使用一个免费的模拟天气API作为示例。 # 在实际项目中，你需要替换为真实的天气API（如和风天气、OpenWeatherMap等）。 # 并妥善处理API Key和请求限制。 try: # 示例：使用一个公开的模拟API（仅用于演示，可能不稳定） url = f"https://wttr.in/{city}?format=%C+%t" response = requests.get(url, timeout=10) if response.status_code == 200: weather_info = response.text.strip() return f"{city}的天气：{weather_info}" else: return f"无法获取{city}的天气信息。API返回状态码：{response.status_code}" except requests.exceptions.RequestException as e: return f"天气查询请求失败：{e}"

最后，我们需要一个工具注册中心来管理所有可用工具：

# tools/registry.py from typing import Dict, List from tools.base import BaseTool class ToolRegistry: """工具注册表，集中管理所有可用工具""" def __init__(self): self._tools: Dict[str, BaseTool] = {} def register(self, tool: BaseTool) -> None: """注册一个工具""" if tool.name in self._tools: raise ValueError(f"工具 '{tool.name}' 已注册。") self._tools[tool.name] = tool print(f"工具已注册: {tool.name} - {tool.description[:50]}...") def get_tool(self, name: str) -> BaseTool: """根据名称获取工具""" tool = self._tools.get(name) if not tool: raise KeyError(f"未找到工具: {name}") return tool def list_tools(self) -> List[Dict]: """列出所有已注册工具的信息""" return [ { "name": tool.name, "description": tool.description, "parameters": tool.schema.parameters } for tool in self._tools.values() ] @property def tool_descriptions_for_prompt(self) -> str: """生成用于提示词的工具描述文本""" descriptions = [] for tool in self._tools.values(): desc = f"- {tool.name}: {tool.description}" # 可以添加参数schema的简要说明 # desc += f"\n 参数: {tool.schema.parameters}" descriptions.append(desc) return "\n".join(descriptions) # 创建全局工具注册表实例 registry = ToolRegistry()

3.3 记忆层实现 (core/memory/)

记忆使Agent能记住对话历史。我们先实现一个简单的基于列表的短期记忆。

# core/memory/base.py from abc import ABC, abstractmethod from typing import List, Dict, Any from core.models import Message class BaseMemory(ABC): """记忆基类""" @abstractmethod def add(self, message: Message) -> None: """添加一条消息到记忆""" pass @abstractmethod def get_context(self, limit: int = 10) -> List[Message]: """获取最近的对话上下文""" pass @abstractmethod def clear(self) -> None: """清空记忆""" pass

# core/memory/simple.py from typing import List from core.memory.base import BaseMemory from core.models import Message class SimpleMemory(BaseMemory): """简单的对话记忆，使用列表存储""" def __init__(self, max_messages: int = 20): self.messages: List[Message] = [] self.max_messages = max_messages def add(self, message: Message) -> None: self.messages.append(message) # 限制记忆长度，移除最早的消息 if len(self.messages) > self.max_messages: self.messages = self.messages[-self.max_messages:] def get_context(self, limit: int = 10) -> List[Message]: return self.messages[-limit:] if self.messages else [] def clear(self) -> None: self.messages.clear() def __len__(self) -> int: return len(self.messages)

数据模型core/models.py定义了消息等基础结构：

# core/models.py from pydantic import BaseModel, Field from typing import Optional, Dict, Any from datetime import datetime class Message(BaseModel): """对话消息""" role: str # 'system', 'user', 'assistant', 'tool' content: str timestamp: datetime = Field(default_factory=datetime.now) def to_dict(self) -> Dict[str, str]: """转换为LLM API所需的格式""" return {"role": self.role, "content": self.content} class ToolCall(BaseModel): """工具调用记录""" tool_name: str parameters: Dict[str, Any] result: Optional[str] = None timestamp: datetime = Field(default_factory=datetime.now)

4. 智能体编排器：核心大脑的实现

这是整个工具链的“调度中心”。我们将实现一个遵循 ReAct (Reasoning + Acting) 范式的简单Agent。

4.1 Agent 主类设计 (core/agent.py)

# core/agent.py import json import re from typing import List, Dict, Any, Optional from loguru import logger from core.models import Message, ToolCall from core.memory.simple import SimpleMemory from llm.client import LLMClient from tools.registry import registry class Agent: """智能体核心编排器""" def __init__(self, llm_client: Optional[LLMClient] = None, memory: Optional[SimpleMemory] = None, system_prompt: Optional[str] = None): """ 初始化Agent Args: llm_client: LLM客户端实例 memory: 记忆实例 system_prompt: 系统提示词，定义Agent的角色和能力 """ self.llm = llm_client or LLMClient() self.memory = memory or SimpleMemory() self.system_prompt = system_prompt or self._default_system_prompt() # 初始化系统消息 if self.system_prompt: system_msg = Message(role="system", content=self.system_prompt) self.memory.add(system_msg) self.max_iterations = 10 # 防止无限循环 logger.info("Agent initialized.") def _default_system_prompt(self) -> str: """默认系统提示词，包含工具描述""" tools_desc = registry.tool_descriptions_for_prompt return f"""你是一个有帮助的AI助手，可以调用工具来解决问题。 你可以使用的工具如下： {tools_desc} 请遵循以下规则： 1. 仔细分析用户的问题。 2. 如果需要使用工具，请严格按照以下JSON格式回复： ```json {{"action": "tool_call", "tool": "工具名称", "parameters": {{"参数名": "参数值"}}}}

3. 如果不需要工具或已获得最终答案，请直接回复答案。

4. 保持回答简洁专业。 """

   def _extract_tool_call(self, response: str) -> Optional[Dict[str, Any]]: """从模型回复中提取工具调用指令""" # 尝试查找JSON块 json_pattern = r'json\s*(.*?)\s*' match = re.search(json_pattern, response, re.DOTALL) json_str = match.group(1) if match else response

   try: data = json.loads(json_str) if isinstance(data, dict) and data.get("action") == "tool_call": return data except json.JSONDecodeError: # 如果没有找到有效的JSON，尝试直接解析整个响应（简化处理） if '"action": "tool_call"' in response: try: # 尝试提取最内层的JSON对象 start = response.find('{') end = response.rfind('}') + 1 if start != -1 and end != 0: data = json.loads(response[start:end]) if data.get("action") == "tool_call": return data except: pass return None

   def _call_tool(self, tool_name: str, parameters: Dict[str, Any]) -> str: """调用指定工具并返回结果""" try: tool = registry.get_tool(tool_name) logger.info(f"调用工具: {tool_name}, 参数: {parameters}") result = tool.execute(**parameters) logger.info(f"工具结果: {result[:100]}...") return result except Exception as e: error_msg = f"工具调用失败: {e}" logger.error(error_msg) return error_msg

   def run(self, user_input: str) -> str: """ 运行Agent处理用户输入 Returns: 最终回复给用户的文本 """ logger.info(f"用户输入: {user_input}")

   # 1. 添加用户消息到记忆 user_msg = Message(role="user", content=user_input) self.memory.add(user_msg) # 2. 开始推理-行动循环 for iteration in range(self.max_iterations): logger.debug(f"--- 迭代 {iteration + 1} ---") # 2.1 准备对话上下文 context_messages = self.memory.get_context(limit=15) llm_messages = [msg.to_dict() for msg in context_messages] # 2.2 调用LLM获取响应 try: llm_response = self.llm.chat_completion(llm_messages, temperature=0.1) except Exception as e: error_msg = f"LLM调用失败: {e}" logger.error(error_msg) return error_msg # 2.3 分析响应，判断是否需要调用工具 tool_call_data = self._extract_tool_call(llm_response) if tool_call_data: # 需要调用工具 tool_name = tool_call_data.get("tool") parameters = tool_call_data.get("parameters", {}) # 调用工具 tool_result = self._call_tool(tool_name, parameters) # 将工具调用和结果添加到记忆 tool_call_msg = Message( role="tool", content=f"调用工具 '{tool_name}' 结果: {tool_result}" ) self.memory.add(tool_call_msg) # 继续下一轮迭代（让LLM基于工具结果继续思考） continue else: # 不需要调用工具，LLM给出了最终答案 assistant_msg = Message(role="assistant", content=llm_response) self.memory.add(assistant_msg) logger.info(f"Agent 最终回复: {llm_response[:200]}...") return llm_response # 如果达到最大迭代次数仍未得出最终答案 timeout_msg = "抱歉，处理超时。可能任务过于复杂或工具调用出现循环。" logger.warning(timeout_msg) return timeout_msg

   def reset(self) -> None: """重置Agent的记忆（除了系统提示）""" system_msg = None for msg in self.memory.get_context(): if msg.role == "system": system_msg = msg break

   self.memory.clear() if system_msg: self.memory.add(system_msg) logger.info("Agent记忆已重置。")

## 5. 组装与运行：构建完整的智能体系统 现在，我们将所有模块组装起来，创建一个可运行的智能体应用。 ### 5.1 应用主入口 (`main.py`) ```python # main.py import sys from loguru import logger from core.agent import Agent from tools.calculator import CalculatorTool from tools.weather import WeatherTool from tools.registry import registry def setup_logging(): """配置日志""" logger.remove() # 移除默认处理器 logger.add( sys.stderr, format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>", level="INFO" ) logger.add("agent.log", rotation="10 MB", level="DEBUG") # 文件日志 def initialize_tools(): """初始化并注册所有工具""" # 注册计算器工具 calc_tool = CalculatorTool() registry.register(calc_tool) # 注册天气查询工具 weather_tool = WeatherTool() registry.register(weather_tool) # 可以在这里注册更多工具... logger.info(f"已注册 {len(registry.list_tools())} 个工具") def main(): """主函数""" setup_logging() logger.info("=== 硬核Agent工具链启动 ===") # 1. 初始化工具 initialize_tools() # 2. 创建Agent实例 agent = Agent() # 3. 交互循环 print("\n" + "="*50) print("硬核Agent控制台") print("输入 'quit' 或 'exit' 退出") print("输入 'reset' 清空对话历史") print("="*50) while True: try: user_input = input("\n>>> 你: ").strip() if user_input.lower() in ['quit', 'exit', 'q']: print("再见！") break elif user_input.lower() == 'reset': agent.reset() print("对话历史已清空。") continue elif not user_input: continue # 运行Agent print("Agent 思考中...") response = agent.run(user_input) print(f"\n>>> Agent: {response}") except KeyboardInterrupt: print("\n\n程序被中断。") break except Exception as e: logger.error(f"处理用户输入时出错: {e}") print(f"抱歉，出错了: {e}") if __name__ == "__main__": main()

5.2 运行你的第一个智能体

确保你的.env文件中已正确配置了 LLM_API_KEY。然后运行：

python main.py

你将看到控制台启动，并提示你输入。尝试以下对话：

>>> 你: 北京现在的天气怎么样？ Agent 思考中... >>> Agent: 北京的天气：Clear +9°C >>> 你: 这个温度下，如果我要穿一件厚度为2.5的毛衣，体感温度会是多少？ Agent 思考中... （Agent可能会尝试调用计算器进行估算，或直接给出建议）

5.3 扩展：添加持久化层 (persistence/)

为了记录Agent的运行历史，我们可以将对话和工具调用保存到数据库。这里使用SQLite和SQLAlchemy作为示例。

# persistence/models.py from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, JSON from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from datetime import datetime from config.settings import settings Base = declarative_base() class ConversationRecord(Base): """对话记录表""" __tablename__ = 'conversations' id = Column(Integer, primary_key=True) session_id = Column(String(100), index=True) # 会话ID role = Column(String(20)) # user, assistant, tool, system content = Column(Text) timestamp = Column(DateTime, default=datetime.now) metadata = Column(JSON, default={}) # 存储额外信息，如工具调用参数 class ToolCallRecord(Base): """工具调用记录表""" __tablename__ = 'tool_calls' id = Column(Integer, primary_key=True) conversation_id = Column(Integer, index=True) # 关联的对话记录ID tool_name = Column(String(100)) parameters = Column(JSON) result = Column(Text) duration_ms = Column(Integer) # 调用耗时（毫秒） timestamp = Column(DateTime, default=datetime.now) success = Column(Integer, default=1) # 1成功，0失败

# persistence/database.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, scoped_session from persistence.models import Base, ConversationRecord, ToolCallRecord from config.settings import settings import uuid from contextlib import contextmanager class DatabaseManager: """数据库管理器""" def __init__(self, database_url: str = None): self.database_url = database_url or settings.DATABASE_URL self.engine = create_engine(self.database_url, echo=False) self.SessionLocal = scoped_session(sessionmaker(bind=self.engine)) # 创建表 Base.metadata.create_all(bind=self.engine) @contextmanager def get_session(self): """获取数据库会话的上下文管理器""" session = self.SessionLocal() try: yield session session.commit() except Exception: session.rollback() raise finally: session.close() def log_conversation(self, session_id: str, role: str, content: str, metadata: dict = None): """记录对话""" with self.get_session() as db: record = ConversationRecord( session_id=session_id, role=role, content=content, metadata=metadata or {} ) db.add(record) db.flush() # 获取ID return record.id def log_tool_call(self, conversation_id: int, tool_name: str, parameters: dict, result: str, duration_ms: int, success: bool = True): """记录工具调用""" with self.get_session() as db: record = ToolCallRecord( conversation_id=conversation_id, tool_name=tool_name, parameters=parameters, result=result, duration_ms=duration_ms, success=1 if success else 0 ) db.add(record) # 全局数据库管理器实例 db_manager = DatabaseManager()

然后，你可以在Agent的run方法中集成日志记录，在每次交互后保存到数据库。这为后续的分析、监控和Agent学习提供了数据基础。

6. 常见问题与排查指南

在搭建和运行过程中，你可能会遇到以下典型问题。

6.1 LLM API 调用失败

调试提示：在LLMClient的chat_completion方法中，打印出发送给API的完整消息列表，这有助于检查提示词是否被正确构建。

6.2 工具调用异常

6.3 无限循环或逻辑错误

6.4 性能问题

7. 工程化最佳实践与扩展方向

当你掌握了基础搭建后，以下实践能帮助你将其发展为生产可用的系统。

7.1 配置管理与安全

• 敏感信息分离：始终坚持使用.env文件或配置中心管理API Keys、数据库密码等，切勿硬编码。
• 配置验证：使用pydantic-settings对配置进行强类型验证，确保应用启动时配置完整正确。
• 工具安全：对于CalculatorTool这类执行动态代码的工具，生产环境必须替换eval()，使用安全的表达式解析库（如ast.literal_eval用于简单表达式，或numexpr、pandas.eval用于复杂计算）。

7.2 可观测性与监控

• 结构化日志：使用loguru或structlog输出JSON格式的日志，便于接入ELK等日志系统。记录关键事件：用户输入、LLM请求/响应、工具调用（入参、结果、耗时）、错误。
• 指标收集：集成Prometheus客户端，暴露指标如：请求数、平均响应时间、工具调用成功率、各环节耗时（LLM、工具、总耗时）。
• 链路追踪：为每个用户会话生成唯一session_id，并在所有日志和数据库记录中携带，方便问题追踪。

7.3 性能与稳定性优化

• 异步化：将LLMClient.chat_completion和工具调用改为异步函数（async/await），使用asyncio或anyio管理并发，大幅提升吞吐量。
• 缓存：对频繁且结果稳定的工具调用（如天气查询，可缓存5分钟）或LLM对相同问题的回复实施缓存，减少外部调用和成本。
• 限流与降级：为LLM API和关键工具实现限流（如token bucket）。当主要工具失败时，提供降级方案（如天气API失败时，返回缓存数据或友好提示）。
• 上下文管理：实现更高级的记忆模块，如：
  • 向量记忆：使用ChromaDB或FAISS存储长期记忆，通过语义搜索检索相关历史。
  • 摘要记忆：当对话过长时，调用LLM对旧对话进行摘要，保留核心信息，节省Token。

7.4 扩展工具生态

• 动态工具加载：支持从配置文件或特定目录自动发现和加载工具类，无需修改核心代码。
• 工具权限控制：为工具添加标签（如read,write,admin），并根据用户会话的权限动态过滤可用的工具列表。
• 工具组合与工作流：实现更复杂的规划器，让Agent能够自动将复杂任务分解为子任务，并顺序或并行调用多个工具（即实现智能体工作流）。

7.5 生产部署考虑

• 容器化：使用 Docker 打包应用，确保环境一致性。
• 健康检查：提供/health端点，检查LLM API连通性、数据库连接和工具状态。
• 配置热更新：实现不重启服务即可更新提示词、工具列表等配置的能力。
• 版本管理：对Agent的行为定义（系统提示词、工具集）进行版本控制，便于回滚和A/B测试。

从零搭建智能体工具链是一个深度理解AI Agent运作机制的过程。本文带你走完了从概念到可运行原型的关键路径，涵盖了模型集成、工具定义、记忆管理、任务编排和基础持久化。这套框架是一个坚实的起点，你可以在此基础上，根据实际业务需求，深入探索异步优化、高级记忆、复杂规划、评估体系等更专业的领域。真正的“硬核”之旅，始于你动手解决第一个实际业务问题之时。