2026年AI Agent实战：从玩具到生产力的落地手册（附源码）

欢迎文末添加好友交流，共同进步！

“ 俺はモンキー・D・ルフィ。海贼王になる男だ！”

• • 前言
  • 目录
  • 一、AI Agent 的核心架构
  • • 1.1 什么是AI Agent？
    • 1.2 2026年Agent技术栈全景
  • 二、从零搭建生产级Agent框架
  • • 2.1 项目结构设计
    • 2.2 核心代码：Agent基类
    • 2.3 记忆管理系统
  • 三、三大核心技术实现
  • • 3.1 ReAct框架：推理+行动协同
    • 3.2 工具调用系统
    • 3.3 任务规划器
  • 四、实战案例：智能客服Agent
  • • 4.1 场景分析
    • 4.2 完整实现
    • 4.3 性能对比
  • 五、性能优化与成本控制
  • • 5.1 成本分析
    • 5.2 优化策略
  • 六、完整源码
  • • 快速开始
    • 项目结构
  • 总结
  • • 关键要点
    • 下一步学习
  • 参考资源

前言

2023年是ChatGPT元年，2024年是多模态爆发之年，而2026年则是AI Agent的落地元年。

你可能用过Coze搭建过聊天机器人，也可能在Dify上配置过知识库问答，但当真正要把AI Agent投入生产环境时，你会发现：

• ❌ 上下文记忆经常丢失
• ❌ 工具调用成功率只有60%
• ❌ 成本控制一团混乱
• ❌ 无法处理复杂的多步任务

本文将带你从零开始，手写一个生产级AI Agent框架，解决上述所有问题。文末附完整源码。

目录

• 一、AI Agent 的核心架构
• 二、从零搭建生产级Agent框架
• 三、三大核心技术实现
• 四、实战案例：智能客服Agent
• 五、性能优化与成本控制
• 六、完整源码

一、AI Agent 的核心架构

1.1 什么是AI Agent？

简单来说，AI Agent = LLM + 记忆 + 规划 + 工具

#mermaid-svg-Ldw8lKpDH05BLpRf{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-Ldw8lKpDH05BLpRf .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-Ldw8lKpDH05BLpRf .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-Ldw8lKpDH05BLpRf .error-icon{fill:#552222;}#mermaid-svg-Ldw8lKpDH05BLpRf .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-Ldw8lKpDH05BLpRf .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-Ldw8lKpDH05BLpRf .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-Ldw8lKpDH05BLpRf .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-Ldw8lKpDH05BLpRf .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-Ldw8lKpDH05BLpRf .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-Ldw8lKpDH05BLpRf .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-Ldw8lKpDH05BLpRf .marker{fill:#333333;stroke:#333333;}#mermaid-svg-Ldw8lKpDH05BLpRf .marker.cross{stroke:#333333;}#mermaid-svg-Ldw8lKpDH05BLpRf svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-Ldw8lKpDH05BLpRf p{margin:0;}#mermaid-svg-Ldw8lKpDH05BLpRf .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-Ldw8lKpDH05BLpRf .cluster-label text{fill:#333;}#mermaid-svg-Ldw8lKpDH05BLpRf .cluster-label span{color:#333;}#mermaid-svg-Ldw8lKpDH05BLpRf .cluster-label span p{background-color:transparent;}#mermaid-svg-Ldw8lKpDH05BLpRf .label text,#mermaid-svg-Ldw8lKpDH05BLpRf span{fill:#333;color:#333;}#mermaid-svg-Ldw8lKpDH05BLpRf .node rect,#mermaid-svg-Ldw8lKpDH05BLpRf .node circle,#mermaid-svg-Ldw8lKpDH05BLpRf .node ellipse,#mermaid-svg-Ldw8lKpDH05BLpRf .node polygon,#mermaid-svg-Ldw8lKpDH05BLpRf .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-Ldw8lKpDH05BLpRf .rough-node .label text,#mermaid-svg-Ldw8lKpDH05BLpRf .node .label text,#mermaid-svg-Ldw8lKpDH05BLpRf .image-shape .label,#mermaid-svg-Ldw8lKpDH05BLpRf .icon-shape .label{text-anchor:middle;}#mermaid-svg-Ldw8lKpDH05BLpRf .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-Ldw8lKpDH05BLpRf .rough-node .label,#mermaid-svg-Ldw8lKpDH05BLpRf .node .label,#mermaid-svg-Ldw8lKpDH05BLpRf .image-shape .label,#mermaid-svg-Ldw8lKpDH05BLpRf .icon-shape .label{text-align:center;}#mermaid-svg-Ldw8lKpDH05BLpRf .node.clickable{cursor:pointer;}#mermaid-svg-Ldw8lKpDH05BLpRf .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-Ldw8lKpDH05BLpRf .arrowheadPath{fill:#333333;}#mermaid-svg-Ldw8lKpDH05BLpRf .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-Ldw8lKpDH05BLpRf .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-Ldw8lKpDH05BLpRf .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-Ldw8lKpDH05BLpRf .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-Ldw8lKpDH05BLpRf .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-Ldw8lKpDH05BLpRf .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-Ldw8lKpDH05BLpRf .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-Ldw8lKpDH05BLpRf .cluster text{fill:#333;}#mermaid-svg-Ldw8lKpDH05BLpRf .cluster span{color:#333;}#mermaid-svg-Ldw8lKpDH05BLpRf div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-Ldw8lKpDH05BLpRf .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-Ldw8lKpDH05BLpRf rect.text{fill:none;stroke-width:0;}#mermaid-svg-Ldw8lKpDH05BLpRf .icon-shape,#mermaid-svg-Ldw8lKpDH05BLpRf .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-Ldw8lKpDH05BLpRf .icon-shape p,#mermaid-svg-Ldw8lKpDH05BLpRf .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-Ldw8lKpDH05BLpRf .icon-shape rect,#mermaid-svg-Ldw8lKpDH05BLpRf .image-shape rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-Ldw8lKpDH05BLpRf .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-Ldw8lKpDH05BLpRf .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-Ldw8lKpDH05BLpRf :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

用户输入

感知层 Perception

大脑层 BrainLLM推理引擎

记忆层 Memory短期+长期记忆

规划层 Planning任务分解

工具层 ToolsAPI/函数调用

决策层 Decision

行动层 Action

输出结果

1.2 2026年Agent技术栈全景

二、从零搭建生产级Agent框架

2.1 项目结构设计

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 # 重试机制

2.2 核心代码：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, # LLM客户端
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}")

2.3 记忆管理系统

from abc import ABC, abstractmethod
from typing import List, Dict, Any
import redis
import numpy as np
from datetime import datetime, timedelta

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]

三、三大核心技术实现

3.1 ReAct框架：推理+行动协同

ReAct（Reasoning + Acting）是目前Agent最主流的推理范式。

#mermaid-svg-UofScqICarC5OfgW{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-UofScqICarC5OfgW .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-UofScqICarC5OfgW .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-UofScqICarC5OfgW .error-icon{fill:#552222;}#mermaid-svg-UofScqICarC5OfgW .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-UofScqICarC5OfgW .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-UofScqICarC5OfgW .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-UofScqICarC5OfgW .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-UofScqICarC5OfgW .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-UofScqICarC5OfgW .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-UofScqICarC5OfgW .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-UofScqICarC5OfgW .marker{fill:#333333;stroke:#333333;}#mermaid-svg-UofScqICarC5OfgW .marker.cross{stroke:#333333;}#mermaid-svg-UofScqICarC5OfgW svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-UofScqICarC5OfgW p{margin:0;}#mermaid-svg-UofScqICarC5OfgW .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-UofScqICarC5OfgW .cluster-label text{fill:#333;}#mermaid-svg-UofScqICarC5OfgW .cluster-label span{color:#333;}#mermaid-svg-UofScqICarC5OfgW .cluster-label span p{background-color:transparent;}#mermaid-svg-UofScqICarC5OfgW .label text,#mermaid-svg-UofScqICarC5OfgW span{fill:#333;color:#333;}#mermaid-svg-UofScqICarC5OfgW .node rect,#mermaid-svg-UofScqICarC5OfgW .node circle,#mermaid-svg-UofScqICarC5OfgW .node ellipse,#mermaid-svg-UofScqICarC5OfgW .node polygon,#mermaid-svg-UofScqICarC5OfgW .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-UofScqICarC5OfgW .rough-node .label text,#mermaid-svg-UofScqICarC5OfgW .node .label text,#mermaid-svg-UofScqICarC5OfgW .image-shape .label,#mermaid-svg-UofScqICarC5OfgW .icon-shape .label{text-anchor:middle;}#mermaid-svg-UofScqICarC5OfgW .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-UofScqICarC5OfgW .rough-node .label,#mermaid-svg-UofScqICarC5OfgW .node .label,#mermaid-svg-UofScqICarC5OfgW .image-shape .label,#mermaid-svg-UofScqICarC5OfgW .icon-shape .label{text-align:center;}#mermaid-svg-UofScqICarC5OfgW .node.clickable{cursor:pointer;}#mermaid-svg-UofScqICarC5OfgW .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-UofScqICarC5OfgW .arrowheadPath{fill:#333333;}#mermaid-svg-UofScqICarC5OfgW .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-UofScqICarC5OfgW .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-UofScqICarC5OfgW .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-UofScqICarC5OfgW .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-UofScqICarC5OfgW .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-UofScqICarC5OfgW .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-UofScqICarC5OfgW .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-UofScqICarC5OfgW .cluster text{fill:#333;}#mermaid-svg-UofScqICarC5OfgW .cluster span{color:#333;}#mermaid-svg-UofScqICarC5OfgW div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-UofScqICarC5OfgW .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-UofScqICarC5OfgW rect.text{fill:none;stroke-width:0;}#mermaid-svg-UofScqICarC5OfgW .icon-shape,#mermaid-svg-UofScqICarC5OfgW .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-UofScqICarC5OfgW .icon-shape p,#mermaid-svg-UofScqICarC5OfgW .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-UofScqICarC5OfgW .icon-shape rect,#mermaid-svg-UofScqICarC5OfgW .image-shape rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-UofScqICarC5OfgW .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-UofScqICarC5OfgW .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-UofScqICarC5OfgW :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

是

否

用户问题

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

3.2 工具调用系统

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 else "string"
parameters[param_name] = {
"type": param_type.__name__ if hasattr(param_type, "__name__") else "string",
"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()]

3.3 任务规划器

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": []},
{"task": "任务描述", "order": 2, "dependencies": [1]},
…
]
"""
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

4.1 场景分析

智能客服是AI Agent最典型的应用场景。我们来实现一个政务大厅智能客服，具备：

• 政策问答
• 办事流程引导
• 工单生成
• 人工转接

#mermaid-svg-KPwPMXNDoAXsK09e{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-KPwPMXNDoAXsK09e .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-KPwPMXNDoAXsK09e .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-KPwPMXNDoAXsK09e .error-icon{fill:#552222;}#mermaid-svg-KPwPMXNDoAXsK09e .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-KPwPMXNDoAXsK09e .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-KPwPMXNDoAXsK09e .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-KPwPMXNDoAXsK09e .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-KPwPMXNDoAXsK09e .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-KPwPMXNDoAXsK09e .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-KPwPMXNDoAXsK09e .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-KPwPMXNDoAXsK09e .marker{fill:#333333;stroke:#333333;}#mermaid-svg-KPwPMXNDoAXsK09e .marker.cross{stroke:#333333;}#mermaid-svg-KPwPMXNDoAXsK09e svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-KPwPMXNDoAXsK09e p{margin:0;}#mermaid-svg-KPwPMXNDoAXsK09e .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-KPwPMXNDoAXsK09e .cluster-label text{fill:#333;}#mermaid-svg-KPwPMXNDoAXsK09e .cluster-label span{color:#333;}#mermaid-svg-KPwPMXNDoAXsK09e .cluster-label span p{background-color:transparent;}#mermaid-svg-KPwPMXNDoAXsK09e .label text,#mermaid-svg-KPwPMXNDoAXsK09e span{fill:#333;color:#333;}#mermaid-svg-KPwPMXNDoAXsK09e .node rect,#mermaid-svg-KPwPMXNDoAXsK09e .node circle,#mermaid-svg-KPwPMXNDoAXsK09e .node ellipse,#mermaid-svg-KPwPMXNDoAXsK09e .node polygon,#mermaid-svg-KPwPMXNDoAXsK09e .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-KPwPMXNDoAXsK09e .rough-node .label text,#mermaid-svg-KPwPMXNDoAXsK09e .node .label text,#mermaid-svg-KPwPMXNDoAXsK09e .image-shape .label,#mermaid-svg-KPwPMXNDoAXsK09e .icon-shape .label{text-anchor:middle;}#mermaid-svg-KPwPMXNDoAXsK09e .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-KPwPMXNDoAXsK09e .rough-node .label,#mermaid-svg-KPwPMXNDoAXsK09e .node .label,#mermaid-svg-KPwPMXNDoAXsK09e .image-shape .label,#mermaid-svg-KPwPMXNDoAXsK09e .icon-shape .label{text-align:center;}#mermaid-svg-KPwPMXNDoAXsK09e .node.clickable{cursor:pointer;}#mermaid-svg-KPwPMXNDoAXsK09e .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-KPwPMXNDoAXsK09e .arrowheadPath{fill:#333333;}#mermaid-svg-KPwPMXNDoAXsK09e .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-KPwPMXNDoAXsK09e .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-KPwPMXNDoAXsK09e .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-KPwPMXNDoAXsK09e .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-KPwPMXNDoAXsK09e .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-KPwPMXNDoAXsK09e .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-KPwPMXNDoAXsK09e .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-KPwPMXNDoAXsK09e .cluster text{fill:#333;}#mermaid-svg-KPwPMXNDoAXsK09e .cluster span{color:#333;}#mermaid-svg-KPwPMXNDoAXsK09e div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-KPwPMXNDoAXsK09e .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-KPwPMXNDoAXsK09e rect.text{fill:none;stroke-width:0;}#mermaid-svg-KPwPMXNDoAXsK09e .icon-shape,#mermaid-svg-KPwPMXNDoAXsK09e .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-KPwPMXNDoAXsK09e .icon-shape p,#mermaid-svg-KPwPMXNDoAXsK09e .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-KPwPMXNDoAXsK09e .icon-shape rect,#mermaid-svg-KPwPMXNDoAXsK09e .image-shape rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-KPwPMXNDoAXsK09e .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-KPwPMXNDoAXsK09e .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-KPwPMXNDoAXsK09e :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

政策咨询

办事指引

投诉建议

复杂问题

用户咨询

意图识别

RAG检索知识库

流程引导Agent

工单生成

人工转接

生成回复

排队等待

4.2 完整实现

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())

4.3 性能对比

五、性能优化与成本控制

5.1 成本分析

AI Agent的主要成本来源：

#mermaid-svg-y5POQx4abUYwKkkx{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-y5POQx4abUYwKkkx .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-y5POQx4abUYwKkkx .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-y5POQx4abUYwKkkx .error-icon{fill:#552222;}#mermaid-svg-y5POQx4abUYwKkkx .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-y5POQx4abUYwKkkx .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-y5POQx4abUYwKkkx .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-y5POQx4abUYwKkkx .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-y5POQx4abUYwKkkx .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-y5POQx4abUYwKkkx .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-y5POQx4abUYwKkkx .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-y5POQx4abUYwKkkx .marker{fill:#333333;stroke:#333333;}#mermaid-svg-y5POQx4abUYwKkkx .marker.cross{stroke:#333333;}#mermaid-svg-y5POQx4abUYwKkkx svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-y5POQx4abUYwKkkx p{margin:0;}#mermaid-svg-y5POQx4abUYwKkkx .pieCircle{stroke:#000000;stroke-width:2px;opacity:0.7;}#mermaid-svg-y5POQx4abUYwKkkx .pieOuterCircle{stroke:#000000;stroke-width:1px;fill:none;}#mermaid-svg-y5POQx4abUYwKkkx .pieTitleText{text-anchor:middle;font-size:25px;fill:#000000;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}#mermaid-svg-y5POQx4abUYwKkkx .slice{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;fill:#000000;font-size:17px;}#mermaid-svg-y5POQx4abUYwKkkx .legend text{fill:#000000;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:17px;}#mermaid-svg-y5POQx4abUYwKkkx :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

45%

25%

12%

10%

8%

Agent月度成本构成（万次调用）

LLM Token消耗

向量数据库

Redis缓存

API调用

其他

5.2 优化策略

智能缓存实现：

import hashlib
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)

六、完整源码

项目地址: https://github.com/your-repo/agent-framework Star支持: 如果这个项目对你有帮助，请给个⭐

快速开始

# 克隆项目
git clone https://github.com/your-repo/agent-framework.git

# 安装依赖
pip install -r requirements.txt

# 配置环境变量
cp .env.example .env
# 编辑 .env 填入你的API密钥

# 运行示例
python examples/customer_service.py

项目结构

agent-framework/
├── src/
│ ├── agent/
│ │ ├── base.py # Agent基类
│ │ ├── react.py # ReAct实现
│ │ └── planner.py # 任务规划器
│ ├── memory/
│ │ ├── short_term.py # 短期记忆
│ │ ├── long_term.py # 长期记忆
│ │ └── hybrid.py # 混合记忆
│ ├── tools/
│ │ ├── base.py # 工具基类
│ │ ├── registry.py # 工具注册
│ │ └── builtin/
│ │ ├── search.py
│ │ ├── weather.py
│ │ └── calculator.py
│ └── utils/
│ ├── cache.py # 缓存工具
│ ├── logger.py # 日志工具
│ └── retry.py # 重试机制
├── examples/
│ ├── customer_service.py # 智能客服示例
│ ├── research_agent.py # 研究助手示例
│ └── code_agent.py # 代码助手示例
├── tests/
│ ├── test_agent.py
│ ├── test_memory.py
│ └── test_tools.py
├── docs/
│ ├── API.md
│ ├── ARCHITECTURE.md
│ └── TUTORIAL.md
├── requirements.txt
├── setup.py
└── README.md

总结

AI Agent正在从"玩具"向"生产力工具"转变。2026年，掌握Agent开发将成为AI工程师的核心竞争力。

关键要点

下一步学习

• 📖 深入学习 LangChain / LangGraph
• 🔬 研究多Agent协作模式
• 🚀 探索 Agent + RAG 最佳实践
• 📊 建立Agent评估体系

参考资源

• OpenAI Function Calling文档
• LangChain Agent文档
• AutoGen论文
• AgentScope框架

✍️ 坚持用 清晰易懂的图解 + 可落地的代码，让每个知识点都 简单直观！

💡 座右铭：“道路是曲折的，前途是光明的！”