AI Agent 生产级框架设计与实战落地
在实际开发中，把 AI Agent 投入生产环境往往比想象中复杂。你可能用过 Coze 搭建聊天机器人，或者在 Dify 上配置过知识库问答，但真正面对上下文记忆丢失、工具调用成功率低、成本控制混乱以及无法处理复杂多步任务时，现有的低代码平台往往显得力不从心。
本文将带你从零开始，手写一个生产级 AI Agent 框架，解决上述核心问题。
一、AI Agent 的核心架构
简单来说，AI Agent = LLM + 记忆 + 规划 + 工具。理解这个公式是构建系统的基础：
- 感知层 (Perception)：接收用户输入。
- 大脑层 (Brain)：LLM 推理引擎，负责决策。
- 记忆层 (Memory)：管理短期会话和长期知识。
- 规划层 (Planning)：将大目标拆解为可执行步骤。
- 工具层 (Tools)：通过 API 或函数调用扩展能力。
- 行动层 (Action)：执行具体操作并输出结果。
2026 年 Agent 技术栈全景
| 技术层级 | 主流框架/工具 | 推荐指数 | 适用场景 |
|---|---|---|---|
| 编排框架 | LangChain / LangGraph | ⭐⭐⭐⭐⭐ | 复杂工作流编排 |
| 运行时 | AutoGen / AgentScope | ⭐⭐⭐⭐ | 多 Agent 协作 |
| 向量数据库 | Milvus / Chroma | ⭐⭐⭐⭐⭐ | RAG 知识库 |
| 工具生态 | OpenAI Function Calling | ⭐⭐⭐⭐⭐ | 结构化工具调用 |
| 记忆管理 | MemGPT | ⭐⭐⭐⭐ | 长对话场景 |
| 评估框架 | Ragas / TruLens | ⭐⭐⭐⭐ | 生产环境监控 |
二、从零搭建生产级 Agent 框架
项目结构设计
合理的目录结构能显著提升可维护性。建议采用模块化设计：
agent-framework/
├── core/
│ ├── agent.py # Agent 核心类
│ ├── memory.py # 记忆管理模块
│ ├── planner.py # 任务规划器
│ └── tools.py # 工具注册器
├── memory/
│ ├── short_term.py # 短期记忆（Redis）
│ ├── long_term.py # 长期记忆（向量 DB）
│ └── semantic.py # 语义记忆检索
├── tools/
│ ├── base.py # 工具基类
│ ├── registry.py # 工具注册中心
│ └── builtin/ # 内置工具
├── evaluators/
│ ├── cost.py # 成本评估
│ └── performance.py # 性能评估
└── utils/
├── logger.py # 日志系统
└── retry.py # 重试机制
核心代码：Agent 基类
这里我们定义一个生产级的 Agent 基类。注意状态管理和迭代控制，这是防止死循环的关键。
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from enum import Enum
import asyncio
import json
class AgentState(Enum):
"""Agent 状态枚举"""
IDLE = "idle" # 空闲
THINKING = "thinking" # 思考中
ACTING = "acting" # 执行中
WAITING = "waiting" # 等待外部输入
ERROR = "error" # 错误状态
@dataclass
class Message:
"""消息数据结构"""
role: str # user / assistant / system / tool
content: str # 消息内容
tool_calls: Optional[List[Dict]] = None
timestamp: float = None
metadata: Dict[str, Any] = None
class BaseAgent:
"""生产级 Agent 基类"""
def __init__(self,
llm_client: Any,
memory_manager: Any = None,
tool_registry: Any = None,
max_iterations: int = 10,
verbose: bool = True):
self.llm = llm_client
self.memory = memory_manager
self.tools = tool_registry
self.max_iterations = max_iterations
self.verbose = verbose
self.state = AgentState.IDLE
self.conversation_history: List[Message] = []
async def run(self, user_input: str) -> str:
"""Agent 主执行循环"""
# 添加用户消息到历史
self.conversation_history.append(
Message(role="user", content=user_input)
)
self.state = AgentState.THINKING
for iteration in range(self.max_iterations):
self._log(f"迭代 {iteration + 1}/{self.max_iterations}")
# 1. 从记忆中检索相关信息
context = await self._retrieve_context(user_input)
# 2. 构建提示词
prompt = self._build_prompt(context)
# 3. LLM 推理
response = await self._llm_inference(prompt)
# 4. 检查是否需要调用工具
if response.tool_calls:
self.state = AgentState.ACTING
# 执行工具调用
tool_results = await self._execute_tools(response.tool_calls)
# 将工具结果添加到历史
for result in tool_results:
self.conversation_history.append(
Message(
role="tool",
content=result["content"],
tool_name=result["tool_name"]
)
)
else:
# 5. 无需工具调用，返回最终答案
self.state = AgentState.IDLE
self.conversation_history.append(
Message(role="assistant", content=response.content)
)
return response.content
return "超过最大迭代次数，任务未完成"
async def _retrieve_context(self, query: str) -> str:
"""从记忆中检索上下文"""
if not self.memory:
return ""
# 检索相关记忆（这里简化处理）
return await self.memory.search(query, top_k=3)
def _build_prompt(self, context: str) -> str:
"""构建系统提示词"""
system_prompt = f"""
你是一个智能 AI 助手。
# 可用工具
{self.tools.get_tool_descriptions() if self.tools else '无'}
# 相关记忆
{context}
# 任务要求
1. 分析用户需求
2. 如需信息查询或执行操作，调用相应工具
3. 基于工具结果给出准确答案
4. 如无法完成，明确说明原因
开始工作！
"""
return system_prompt
async def _llm_inference(self, prompt: str) -> Any:
"""LLM 推理（示例使用 OpenAI 格式）"""
messages = [
{"role": "system", "content": prompt},
*[{"role": m.role, "content": m.content} for m in self.conversation_history]
]
response = await self.llm.chat.completions.create(
model="gpt-4",
messages=messages,
tools=self.tools.get_tool_schemas() if self.tools else None,
temperature=0.7
)
return response.choices[0].message
async def _execute_tools(self, tool_calls: List[Dict]) -> List[Dict]:
"""执行工具调用"""
results = []
for call in tool_calls:
tool_name = call["function"]["name"]
arguments = json.loads(call["function"]["arguments"])
self._log(f"调用工具：{tool_name} | 参数：{arguments}")
try:
# 从工具注册表获取工具并执行
tool = self.tools.get_tool(tool_name)
result = await tool.execute(**arguments)
results.append({
"tool_name": tool_name,
"content": json.dumps(result, ensure_ascii=False)
})
except Exception as e:
results.append({
"tool_name": tool_name,
"content": json.dumps({"error": str(e)})
})
return results
def _log(self, message: str):
"""日志输出"""
if self.verbose:
print(f"[Agent] {message}")
记忆管理系统
记忆是 Agent 的'灵魂'。我们需要区分短期会话记忆和长期语义记忆。
from abc import ABC, abstractmethod
from typing import List, Dict, Any
import redis
import numpy as np
from datetime import datetime, timedelta
import json
class MemoryBackend(ABC):
"""记忆后端抽象基类"""
@abstractmethod
async def add(self, content: str, metadata: Dict = None) -> str:
"""添加记忆"""
pass
@abstractmethod
async def search(self, query: str, top_k: int = 5) -> List[Dict]:
"""搜索记忆"""
pass
class ShortTermMemory(MemoryBackend):
"""短期记忆：基于 Redis 的会话记忆"""
def __init__(self, redis_url: str = "redis://localhost:6379", ttl: int = 3600):
self.client = redis.from_url(redis_url)
self.ttl = ttl # 记忆过期时间（秒）
async def add(self, content: str, metadata: Dict = None) -> str:
"""添加会话记忆"""
memory_id = f"mem:{datetime.now().timestamp()}"
memory_data = {
"content": content,
"metadata": metadata or {},
"timestamp": datetime.now().isoformat()
}
self.client.setex(memory_id, self.ttl, json.dumps(memory_data, ensure_ascii=False))
return memory_id
async def search(self, query: str, top_k: int = 5) -> List[Dict]:
"""检索最近的相关记忆"""
# 简化版：返回最近的记忆
keys = self.client.keys("mem:*")
memories = []
for key in keys[-top_k:]:
data = json.loads(self.client.get(key))
memories.append(data)
return memories
class LongTermMemory(MemoryBackend):
"""长期记忆：基于向量数据库的语义记忆"""
def __init__(self, embedding_model: Any, vector_db: Any):
self.embedding_model = embedding_model
self.vector_db = vector_db
async def add(self, content: str, metadata: Dict = None) -> str:
"""添加长期记忆"""
# 生成向量嵌入
embedding = await self.embedding_model.embed(content)
# 存储到向量数据库
memory_id = self.vector_db.insert(
vector=embedding,
payload={"content": content, "metadata": metadata or {}}
)
return memory_id
async def search(self, query: str, top_k: int = 5) -> List[Dict]:
"""语义搜索长期记忆"""
# 查询向量嵌入
query_embedding = await self.embedding_model.embed(query)
# 向量检索
results = self.vector_db.search(
vector=query_embedding,
top_k=top_k,
score_threshold=0.7
)
return results
class HybridMemory:
"""混合记忆管理器：整合短期和长期记忆"""
def __init__(self, short_term: ShortTermMemory, long_term: LongTermMemory):
self.short_term = short_term
self.long_term = long_term
async def remember(self, content: str, importance: float = 0.5, metadata: Dict = None):
"""存储记忆（根据重要性决定存储位置）"""
# 始终存入短期记忆
await self.short_term.add(content, metadata)
# 重要记忆存入长期记忆
if importance > 0.7:
await self.long_term.add(content, metadata)
async def recall(self, query: str, top_k: int = 5) -> List[Dict]:
"""回忆相关记忆（整合短期和长期）"""
# 并行检索
short_results = await self.short_term.search(query, top_k // 2)
long_results = await self.long_term.search(query, top_k // 2)
# 合并去重
all_results = short_results + long_results
# 按相关性排序（这里简化）
return all_results[:top_k]
三、三大核心技术实现
ReAct 框架：推理 + 行动协同
ReAct（Reasoning + Acting）是目前 Agent 最主流的推理范式。它让模型在思考和行动之间交替，而不是盲目猜测。
流程逻辑：
- Thought: 分析问题，决定下一步做什么。
- Action: 选择工具并准备参数。
- Observation: 获取工具执行结果。
- Answer: 综合信息给出最终回答。
代码实现：
class ReActAgent(BaseAgent):
"""基于 ReAct 范式的 Agent"""
def _build_react_prompt(self, question: str) -> str:
return f"""
使用以下格式回答问题：
Question: {question}
Thought: 你应该思考做什么
Action: 要采取的操作，应该是 [{self.tools.get_tool_names()}] 中的一个
Observation: 操作的结果
... (这个 Thought/Action/Observation 可以重复 N 次)
Thought: 我现在知道最终答案了
Answer: 对原始问题的最终答案
开始!
Question: {question}
Thought:
"""
async def run(self, user_input: str) -> str:
"""ReAct 循环执行"""
prompt = self._build_react_prompt(user_input)
for _ in range(self.max_iterations):
# LLM 生成下一步动作
response = await self.llm.generate(prompt)
# 解析响应
thought, action, action_input = self._parse_react_response(response)
if not action:
# 没有动作，说明已有答案
return thought
# 执行动作
observation = await self._execute_action(action, action_input)
# 更新提示词
prompt += f"\n{response}\nObservation: {observation}\nThought:"
return "无法在指定迭代次数内完成"
def _parse_react_response(self, response: str) -> tuple:
"""解析 ReAct 响应"""
# 解析 Thought、Action、Action Input
# 这里简化处理，实际需要更复杂的解析
lines = response.strip().split('\n')
thought = ""
action = None
action_input = None
for line in lines:
if line.startswith("Thought:"):
thought = line.replace("Thought:", "").strip()
elif line.startswith("Action:"):
action = line.replace("Action:", "").strip()
elif line.startswith("Action Input:"):
action_input = line.replace("Action Input:", "").strip()
return thought, action, action_input
工具调用系统
工具是 Agent 能力的边界。我们通过装饰器快速注册工具，并自动生成 Schema 供 LLM 识别。
from typing import Callable, Dict, Any, List
import inspect
from pydantic import BaseModel, Field
class Tool(BaseModel):
"""工具基类"""
name: str = Field(description="工具名称")
description: str = Field(description="工具功能描述")
parameters: Dict[str, Any] = Field(default_factory=dict, description="参数 schema")
function: Callable = Field(description="工具执行函数")
class Config:
arbitrary_types_allowed = True
async def execute(self, **kwargs) -> Any:
"""执行工具"""
return await self.function(**kwargs)
def to_openai_schema(self) -> Dict:
"""转换为 OpenAI 函数调用格式"""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters
}
}
def tool(name: str = None, description: str = None):
"""工具装饰器"""
def decorator(func: Callable) -> Tool:
# 提取函数签名
sig = inspect.signature(func)
parameters = {}
for param_name, param in sig.parameters.items():
param_type = param.annotation
if param.annotation != inspect.Parameter.empty:
param_type = param.annotation
else:
param_type = "string"
type_name = param_type.__name__ if hasattr(param_type, "__name__") else "string"
parameters[param_name] = {
"type": type_name,
"description": f"参数 {param_name}"
}
return Tool(
name=name or func.__name__,
description=description or func.__doc__ or "",
parameters={
"type": "object",
"properties": parameters,
"required": [p for p in sig.parameters if p.default == inspect.Parameter.empty]
},
function=func
)
return decorator
# 工具使用示例
@tool(name="search_web", description="搜索网络信息")
async def search_web(query: str, num_results: int = 5):
"""搜索网络信息
Args:
query: 搜索关键词
num_results: 返回结果数量
"""
# 实际实现调用搜索 API
return f"找到 {num_results} 条关于 '{query}' 的结果"
@tool(name="get_weather", description="获取天气信息")
async def get_weather(location: str):
"""获取指定地点的天气信息
Args:
location: 城市名称
"""
# 实际实现调用天气 API
return f"{location} 今天晴，温度 25°C"
class ToolRegistry:
"""工具注册中心"""
def __init__(self):
self._tools: Dict[str, Tool] = {}
def register(self, tool: Tool):
"""注册工具"""
self._tools[tool.name] = tool
def get_tool(self, name: str) -> Tool:
"""获取工具"""
return self._tools.get(name)
def get_tool_names(self) -> List[str]:
"""获取所有工具名称"""
return list(self._tools.keys())
def get_tool_descriptions(self) -> str:
"""获取工具描述文本"""
descriptions = []
for tool in self._tools.values():
descriptions.append(f"- {tool.name}: {tool.description}")
return "\n".join(descriptions)
def get_tool_schemas(self) -> List[Dict]:
"""获取 OpenAI 格式的工具 schema"""
return [tool.to_openai_schema() for tool in self._tools.values()]
任务规划器
对于复杂任务，单轮对话不够用，需要规划器将目标拆解为子任务序列。
class TaskPlanner:
"""任务分解与规划器"""
def __init__(self, llm_client: Any):
self.llm = llm_client
async def plan(self, goal: str) -> List[Dict]:
"""将目标分解为子任务列表
Returns:
[ {"task": "任务描述", "order": 1, "dependencies": []}, ... ]
"""
prompt = f"""
将以下目标分解为具体的、可执行的子任务列表。
目标：{goal}
请按以下格式输出：
1. [任务描述]
2. [任务描述]
...
要求：
- 每个任务应该独立且可执行
- 任务之间应该有逻辑顺序
- 尽量细化到可以直接执行
"""
response = await self.llm.generate(prompt)
# 解析任务列表
tasks = []
for line in response.strip().split('\n'):
if line.strip():
# 提取任务描述
task_desc = line.split('.', 1)[1].strip() if '.' in line else line.strip()
tasks.append({
"task": task_desc,
"order": len(tasks) + 1,
"status": "pending"
})
return tasks
async def execute_plan(self, agent: BaseAgent, tasks: List[Dict]) -> Dict:
"""执行任务计划
Returns:
{ "success": bool, "completed_tasks": List[Dict], "failed_tasks": List[Dict], "final_result": Any }
"""
completed = []
failed = []
for task in tasks:
print(f"\n执行任务 {task['order']}: {task['task']}")
try:
# 使用 Agent 执行单个任务
result = await agent.run(task['task'])
task['status'] = 'completed'
task['result'] = result
completed.append(task)
except Exception as e:
task['status'] = 'failed'
task['error'] = str(e)
failed.append(task)
return {
"success": len(failed) == 0,
"completed_tasks": completed,
"failed_tasks": failed,
"final_result": completed[-1]['result'] if completed else None
}
四、实战案例：智能客服 Agent
智能客服是 AI Agent 最典型的应用场景。我们来实现一个政务大厅智能客服，具备政策问答、办事流程引导、工单生成及人工转接能力。
场景分析
- 政策咨询：RAG 检索知识库。
- 办事指引：流程引导 Agent。
- 投诉建议：情绪安抚 + 记录。
- 复杂问题：意图识别后转人工。
完整实现
import asyncio
from typing import Optional
class CustomerServiceAgent(ReActAgent):
"""智能客服 Agent"""
def __init__(self, knowledge_base, ticket_system, *args, **kwargs):
super().__init__(*args, **kwargs)
self.knowledge_base = knowledge_base
self.ticket_system = ticket_system
# 注册客服专用工具
self._register_customer_service_tools()
def _register_customer_service_tools(self):
"""注册客服工具"""
@self.tools.register
@tool(name="search_policy", description="搜索政策信息")
async def search_policy(query: str):
"""从知识库搜索相关政策
Args:
query: 政策关键词
"""
results = await self.knowledge_base.search(query, top_k=3)
return "\n".join([r['content'] for r in results])
@self.tools.register
@tool(name="get_process_guide", description="获取办事流程")
async def get_process_guide(service_type: str):
"""获取指定业务的办事流程
Args:
service_type: 业务类型（如：身份证办理、社保卡申领）
"""
guide = await self.knowledge_base.get_guide(service_type)
return guide
@self.tools.register
@tool(name="create_ticket", description="创建工单")
async def create_ticket(category: str, description: str, priority: str = "normal"):
"""创建服务工单
Args:
category: 工单类别
description: 问题描述
priority: 优先级（low/normal/high）
"""
ticket_id = await self.ticket_system.create(
category=category,
description=description,
priority=priority
)
return f"工单已创建，编号：{ticket_id}，我们将在 1 个工作日内处理"
@self.tools.register
@tool(name="transfer_to_human", description="转人工客服")
async def transfer_to_human(reason: str):
"""转接到人工客服
Args:
reason: 转接原因
"""
queue_number = await self.ticket_system.human_transfer(reason)
return f"已为您转接人工客服，当前排队人数：{queue_number}人，预计等待时间：{queue_number * 2}分钟"
async def handle_customer_query(self, user_input: str) -> str:
"""处理客户咨询"""
# 意图识别
intent = await self._detect_intent(user_input)
# 根据意图调整系统提示
system_prompt = self._get_system_prompt(intent)
# 执行
return await self.run(user_input)
async def _detect_intent(self, user_input: str) -> str:
"""意图识别"""
intent_prompt = f"""
分类以下用户咨询的意图类型：
用户输入：{user_input}
意图类型：
1. policy_inquiry - 政策咨询
2. process_guide - 办事流程咨询
3. complaint - 投诉建议
4. complex - 复杂问题需人工
只返回意图类型代码：
"""
response = await self.llm.generate(intent_prompt)
return response.strip()
def _get_system_prompt(self, intent: str) -> str:
"""根据意图获取系统提示"""
prompts = {
"policy_inquiry": "你是政策咨询专员，请准确引用政策文件内容...",
"process_guide": "你是办事引导员，请给出清晰的办事步骤...",
"complaint": "你是投诉处理专员，请先安抚情绪，再记录问题...",
"complex": "你是客服助理，对于复杂问题，请主动建议转人工..."
}
return prompts.get(intent, "你是智能客服助手...")
# 使用示例
async def main():
from openai import AsyncOpenAI
# 初始化
llm_client = AsyncOpenAI(api_key="your-api-key")
knowledge_base = MockKnowledgeBase() # 模拟知识库
ticket_system = MockTicketSystem() # 模拟工单系统
agent = CustomerServiceAgent(
llm_client=llm_client,
memory_manager=HybridMemory(
short_term=ShortTermMemory(),
long_term=LongTermMemory()
),
tool_registry=ToolRegistry(),
knowledge_base=knowledge_base,
ticket_system=ticket_system
)
# 处理咨询
response = await agent.handle_customer_query("我想办理社保卡，需要准备什么材料？")
print(response)
if __name__ == "__main__":
asyncio.run(main())
性能对比
| 指标 | 传统规则客服 | 基础 Chatbot | 智能 Agent |
|---|---|---|---|
| 问题解决率 | 35% | 60% | 85% |
| 平均响应时间 | 5 分钟 | 2 秒 | 3 秒 |
| 多轮对话能力 | ❌ | ⚠️ | ✅ |
| 工具调用能力 | ❌ | ❌ | ✅ |
| 学习进化能力 | ❌ | ⚠️ | ✅ |
| 运营成本 | 高 | 低 | 中 |
五、性能优化与成本控制
成本分析
AI Agent 的主要成本来源通常是 Token 消耗，其次是向量数据库和缓存服务。
- LLM Token 消耗: 约 45%
- 向量数据库: 约 25%
- Redis 缓存: 约 12%
- API 调用: 约 10%
- 其他: 约 8%
优化策略
| 优化项 | 策略 | 预期节省 |
|---|---|---|
| Prompt 优化 | 精简系统提示词 | 20-30% |
| 模型选择 | 混合使用 GPT-4/GPT-3.5 | 40-50% |
| 缓存策略 | 重复问题命中缓存 | 30-40% |
| Token 限制 | 动态裁剪上下文 | 15-20% |
| 批量处理 | 合并多个请求 | 10-15% |
智能缓存实现：
import hashlib
import time
from functools import wraps
def smart_cache(ttl: int = 3600):
"""智能缓存装饰器"""
cache = {}
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
# 生成缓存键
key = hashlib.md5(f"{func.__name__}{args}{kwargs}".encode()).hexdigest()
# 检查缓存
if key in cache:
cache_data = cache[key]
if time.time() - cache_data['timestamp'] < ttl:
print("缓存命中!")
return cache_data['result']
# 执行函数
result = await func(*args, **kwargs)
# 存储缓存
cache[key] = {'result': result, 'timestamp': time.time()}
return result
return wrapper
return decorator
# 使用示例
class OptimizedAgent(BaseAgent):
@smart_cache(ttl=1800)
async def _llm_inference(self, prompt: str):
"""带缓存的 LLM 推理"""
return await super()._llm_inference(prompt)
总结
AI Agent 正在从'玩具'向'生产力工具'转变。掌握 Agent 开发已成为 AI 工程师的核心竞争力。
关键要点：
- 记忆是核心竞争力：混合记忆架构（短期 + 长期）是最佳实践。
- 工具调用决定能力边界：丰富的工具生态意味着更强的 Agent 能力。
- 成本控制是关键：智能缓存 + 模型混合可节省 50%+ 成本。
- 评估体系必不可少：建立完善的监控和评估体系。
下一步学习：
- 深入学习 LangChain / LangGraph
- 研究多 Agent 协作模式
- 探索 Agent + RAG 最佳实践
- 建立 Agent 评估体系
参考资源：
