AI 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, 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}")
            context = await self._retrieve_context(user_input)
            prompt = self._build_prompt(context)
            response = await self._llm_inference(prompt)
            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:
                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 最主流的推理范式。

用户问题 -> 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):
            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 响应"""
        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: 返回结果数量"""
    return f"找到 {num_results} 条关于 '{query}' 的结果"

@tool(name="get_weather", description="获取天气信息")
async def get_weather(location: str):
    """获取指定地点的天气信息 Args: location: 城市名称"""
    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]:
        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": []}, ... ]"""
        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:
                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 最典型的应用场景。我们来实现一个政务大厅智能客服，具备：

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

用户咨询 -> 意图识别 -> 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 的主要成本来源：

• LLM Token 消耗 (45%)
• 向量数据库 (25%)
• Redis 缓存 (12%)
• API 调用 (10%)
• 其他 (8%)

5.2 优化策略

智能缓存实现：

import hashlib
from functools import wraps
import time

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)

六、完整源码

快速开始

# 克隆项目
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 工程师的核心竞争力。

关键要点

1. 记忆是 Agent 的核心竞争力 - 混合记忆架构（短期 + 长期）是最佳实践
2. 工具调用决定 Agent 能力边界 - 丰富的工具生态 = 更强的 Agent 能力
3. 成本控制是生产化关键 - 智能缓存 + 模型混合可节省 50%+ 成本
4. 评估体系必不可少 - 建立完善的监控和评估体系

下一步学习

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

参考资源

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