AI Agent 实战指南：从零搭建生产级框架与核心实现

2.2 核心代码：Agent 基类

这是整个框架的心脏。我们需要定义 Agent 的状态流转，并处理主执行循环。注意这里使用了 async 来支持异步 IO，这对高并发场景下的 LLM 调用至关重要。

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}")

2.3 记忆管理系统

记忆是 Agent 区别于普通 Chatbot 的关键。我们设计了混合记忆架构：短期记忆用于会话上下文，长期记忆用于知识沉淀。

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 最主流的推理范式。它让模型在思考和行动之间交替进行，比单纯的 Chain-of-Thought 更适合需要外部交互的场景。

流程如下：

1. Thought: 分析问题
2. Action: 调用工具
3. Observation: 工具结果
4. 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 工具调用系统

工具是 Agent 能力的边界。我们使用装饰器模式来快速注册工具，并自动转换为 OpenAI 兼容的 Schema 格式。

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

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:
                # 使用 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 最典型的应用场景之一。我们来实现一个政务大厅智能客服，具备政策问答、办事流程引导、工单生成和人工转接能力。

4.1 场景分析

• 政策咨询: 需要 RAG 检索知识库
• 办事指引: 结构化流程引导
• 投诉建议: 情绪安抚 + 记录
• 复杂问题: 主动建议转人工

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 的主要成本来源通常是 Token 消耗和 API 调用。合理控制上下文长度和模型选择能显著降低成本。

5.2 优化策略

智能缓存实现：

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 工程师的核心竞争力。

关键要点

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

下一步学习

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

参考资源

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