1. 引言

随着大语言模型（LLM）能力的飞速提升，AI Agent（智能体）已从实验室概念走向生产实践。一个成熟的 Agent 不仅能理解自然语言，更能自主规划、调用工具、执行任务并持续学习。本文将带你全面理解 Agent 的核心架构，并手把手教你如何快速搭建一套可用于生产的 Agent 系统。

下面是 Agent 系统的整体工作流程概览：

2. 什么是 AI Agent？

AI Agent 是一个能够感知环境、自主决策并执行动作的智能实体。与单纯的 LLM 问答不同，Agent 具备以下核心特征：

• 自主性：无需人工持续干预，能独立完成任务分解与执行。
• 工具使用：能够调用外部 API、数据库、搜索引擎等工具。
• 记忆能力：维护短期（对话上下文）与长期（向量数据库）记忆。
• 规划能力：将复杂目标拆解为可执行的子任务序列。

下面用一张图对比 LLM 问答与 Agent 的区别：

3. Agent 核心架构拆解

一个生产级 Agent 通常由以下模块构成，它们协同工作形成完整的智能闭环：

3.1 感知层（Perception）

负责接收并理解用户输入，包括文本、语音、图像等多模态信息。感知层需要做意图识别、实体抽取与上下文理解。

from pydantic import BaseModel
from typing import Optional

class UserIntent(BaseModel):
    """感知层输出：用户意图与实体"""
    intent: str  # 如 "query_order", "cancel_order", "complaint"
    entities: dict[str, str]  # 如 {"order_id": "12345", "product": "手机"}
    confidence: float  # 置信度 0~1
    raw_input: str

def perception_layer(user_input: str) -> UserIntent:
    """感知层处理入口"""
    # 实际项目中调用 LLM 或分类模型
    prompt = f"分析用户意图：{user_input}"
    # llm_response = llm.invoke(prompt)
    # 模拟返回
    return UserIntent(
        intent="query_order",
        entities={"order_id": "12345"},
        confidence=0.95,
        raw_input=user_input
    )

3.2 规划层（Planning）

这是 Agent 的"大脑"，负责：

• 任务分解：将用户目标拆解为可执行的步骤（如 ReAct、Plan-and-Execute 模式）。
• 路径选择：根据当前状态选择最优执行路径。
• 错误恢复：当某一步失败时，自动回退或重试。

3.3 工具层（Tool Use）

Agent 通过工具与外部世界交互。常见工具类型包括：

• 信息检索：搜索引擎、文档库、数据库查询。
• 代码执行：Python 解释器、SQL 执行器。
• 业务 API：CRM、ERP、邮件系统等企业服务。
• 文件操作：读写文件、生成报表。

from typing import Callable, Any
import json

class ToolRegistry:
    """工具注册中心，管理所有可用工具"""
    
    def __init__(self):
        self._tools: dict[str, Callable] = {}
        self._descriptions: dict[str, str] = {}
    
    def register(self, name: str, func: Callable, description: str):
        """注册工具"""
        self._tools[name] = func
        self._descriptions[name] = description
    
    def call(self, name: str, **kwargs) -> Any:
        """调用工具"""
        if name not in self._tools:
            raise ValueError(f"未知工具: {name}")
        return self._tools[name](**kwargs)
    
    def get_tool_descriptions(self) -> str:
        """获取所有工具描述（供 LLM 选择工具用）"""
        return json.dumps(self._descriptions, ensure_ascii=False, indent=2)

# 示例：注册业务工具
registry = ToolRegistry()
registry.register("query_order", query_order_api, "根据订单号查询订单状态")
registry.register("search_kb", search_knowledge_base, "搜索内部知识库文档")
registry.register("send_email", send_email_api, "发送邮件给指定用户")

3.4 记忆层（Memory）

• 短期记忆：当前对话的上下文窗口，由 LLM 的 context 管理。
• 长期记忆：使用向量数据库（如 Chroma、Pinecone）存储历史交互与知识片段，支持语义检索。
• 工作记忆：任务执行过程中的中间状态与临时数据。

from typing import List, Dict, Any
from datetime import datetime

class AgentMemory:
    """Agent 记忆系统"""
    
    def __init__(self, max_short_term: int = 10):
        self.short_term: List[Dict] = []  # 短期记忆
        self.long_term: Dict[str, Any] = {}  # 长期记忆（简化版）
        self.working_memory: Dict[str, Any] = {}  # 工作记忆
        self.max_short_term = max_short_term
    
    def add_to_short_term(self, role: str, content: str):
        """添加短期记忆"""
        self.short_term.append({
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat()
        })
        # 超出长度时丢弃最早的
        if len(self.short_term) > self.max_short_term:
            self.short_term.pop(0)
    
    def save_to_long_term(self, key: str, value: Any):
        """保存到长期记忆"""
        self.long_term[key] = {
            "value": value,
            "saved_at": datetime.now().isoformat()
        }
    
    def get_context(self) -> str:
        """获取当前对话上下文（供 LLM 使用）"""
        return "\n".join(
            f"{m['role']}: {m['content']}" 
            for m in self.short_term[-5:]  # 最近5轮
        )

3.5 执行层（Execution）

负责实际调用工具、处理返回结果，并将结果反馈给规划层进行下一步决策。执行层需要处理并发、超时、限流等生产环境问题。

import asyncio
from concurrent.futures import ThreadPoolExecutor
import time

class ExecutionEngine:
    """执行引擎：管理工具调用的并发与超时"""
    
    def __init__(self, max_workers: int = 5, default_timeout: int = 30):
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
        self.default_timeout = default_timeout
    
    async def execute_tool(self, tool_func, *args, timeout: int = None, **kwargs):
        """异步执行工具调用，带超时控制"""
        timeout = timeout or self.default_timeout
        loop = asyncio.get_event_loop()
        
        try:
            # 在线程池中执行同步函数
            result = await asyncio.wait_for(
                loop.run_in_executor(self.executor, tool_func, *args, **kwargs),
                timeout=timeout
            )
            return {"status": "success", "data": result}
        except asyncio.TimeoutError:
            return {"status": "timeout", "error": f"工具调用超时 ({timeout}s)"}
        except Exception as e:
            return {"status": "error", "error": str(e)}
    
    async def execute_parallel(self, tasks: List[dict]) -> List[dict]:
        """并行执行多个工具调用"""
        coroutines = [
            self.execute_tool(task["func"], *task.get("args", []), **task.get("kwargs", {}))
            for task in tasks
        ]
        return await asyncio.gather(*coroutines)

4. 快速搭建生产 Agent 的步骤

下面用一张流程图展示从零搭建生产 Agent 的完整路径：

4.1 选择 Agent 框架

推荐以下成熟框架：

框架	特点	适用场景	学习成本
LangChain	生态丰富，社区活跃	通用 Agent 开发	中等
CrewAI	多 Agent 协作	复杂工作流	低
AutoGen	微软出品，多 Agent 对话	研究原型	高
Dify	低代码平台	快速验证	极低

4.2 定义 Agent 角色与目标

from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import Tool
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# 初始化 LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)

# 定义工具
def search_knowledge_base(query: str) -> str:
    """搜索内部知识库"""
    # 实际调用向量数据库
    return f"关于 {query} 的文档摘要"

def execute_python(code: str) -> str:
    """安全执行 Python 代码（沙箱环境）"""
    # 实际项目中应使用沙箱执行
    try:
        # exec(code)  # 生产环境请使用沙箱
        return f"代码执行成功"
    except Exception as e:
        return f"执行错误: {e}"

tools = [
    Tool(
        name="知识库搜索",
        func=search_knowledge_base,
        description="搜索内部知识库，获取产品文档和常见问题"
    ),
    Tool(
        name="代码执行器",
        func=execute_python,
        description="执行 Python 代码，用于数据分析和计算"
    ),
]

# 自定义 Agent Prompt
agent_prompt = PromptTemplate.from_template("""
你是一个专业的 AI 助手。请遵循以下原则：
1. 先理解用户问题，再选择合适的工具
2. 如果工具返回错误，尝试其他方式
3. 每一步都要解释你的思考过程
4. 最终给出清晰、完整的答案

工具列表：
{tools}

工具名称格式：{tool_names}

当前对话历史：
{chat_history}

用户输入：{input}

你的思考过程：
{agent_scratchpad}
""")

# 创建 Agent
agent = create_react_agent(
    llm=llm,
    tools=tools,
    prompt=agent_prompt
)

# 创建执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5,  # 最大迭代次数
    early_stopping_method="generate"  # 提前停止策略
)

s,
prompt=agent_prompt
)

executor_with_memory = AgentExecutor(
agent=agent_with_memory,
tools=tools,
memory=memory, # 注入短期记忆
verbose=True,
max_iterations=5
)

### 4.4 实现错误处理与重试机制

```python
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def safe_tool_call(tool_func, *args, **kwargs):
    """带重试的工具调用"""
    try:
        return tool_func(*args, **kwargs)
    except Exception as e:
        print(f"工具调用失败: {e}")
        raise

4.5 部署与监控

• API 封装：使用 FastAPI 将 Agent 包装为 RESTful 服务。
• 日志记录：记录每次 Agent 的思考链（Chain-of-Thought）与工具调用日志。
• 性能监控：跟踪响应时间、工具调用成功率、Token 消耗等指标。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class QueryRequest(BaseModel):
    message: str
    user_id: str

@app.post("/agent/chat")
async def agent_chat(request: QueryRequest):
    # 调用 Agent 并记录日志
    response = agent.invoke({"input": request.message})
    return {"response": response}

5. 生产环境注意事项

5.1 安全性

• 工具权限最小化：只给 Agent 必要的工具权限。
• 输入过滤：防止 Prompt 注入攻击。
• 结果审核：对 Agent 生成的代码或敏感操作进行人工审核。

5.2 性能优化

• 缓存：对重复查询结果进行缓存。
• 异步执行：使用异步框架处理并发请求。
• 模型选择：简单任务用小模型（如 GPT-4o-mini），复杂推理用大模型。

5.3 成本控制

• Token 预算：设置每次对话的 Token 上限。
• 工具调用限制：限制单次任务的工具调用次数。
• 模型降级：当大模型超预算时自动降级到小模型。

6. 实战案例：智能客服 Agent

# 智能客服 Agent 示例
customer_service_agent = create_agent(
    tools=[
        order_query_tool,      # 查询订单
        return_process_tool,   # 退货处理
        knowledge_base_tool,   # 知识库搜索
    ],
    system_prompt="""
    你是一个专业的客服 Agent。请遵循以下原则：
    1. 先理解用户问题，再选择工具
    2. 如果工具返回错误，尝试其他方式
    3. 无法解决时，转接人工客服
    """
)

7. 总结

搭建生产级 AI Agent 需要理解其核心架构（感知-规划-工具-记忆-执行），并关注安全性、性能与成本。通过选择合适的框架、合理配置工具与记忆系统、实现健壮的错误处理，你可以在短时间内构建出可靠的 Agent 应用。

下一步建议：从一个小型场景（如文档问答助手）开始实践，逐步扩展到多 Agent 协作系统。