AI Agent 架构深度解析：从原理到生产级实现的完整指南

┌─────────────────────────────────────────────────────┐
│                    Agent 五层架构                    │
├─────────────────────────────────────────────────────┤
│  Layer 1: 感知层 (Perception)                       │
│  用户意图理解 · 上下文解析 · 多模态输入处理          │
├─────────────────────────────────────────────────────┤
│  Layer 2: 决策层 (Decision / Planning)              │
│  任务分解 · 策略选择 · 下一步行动决策                │
├─────────────────────────────────────────────────────┤
│  Layer 3: 执行层 (Action / Tool Use)                │
│  工具调用 · 参数校验 · 结果解析 · 错误恢复          │
├─────────────────────────────────────────────────────┤
│  Layer 4: 记忆层 (Memory)                           │
│  短期记忆 · 长期记忆 · 工作记忆 · 记忆压缩          │
├─────────────────────────────────────────────────────┤
│  Layer 5: 反思层 (Reflection / Meta-Cognition)      │
│  自我评估 · 策略调整 · 知识沉淀 · 持续改进          │
└─────────────────────────────────────────────────────┘

用户：帮我分析竞品A的市场策略

Step 1 (Planning)：
  计划：
  1. 搜索竞品A的基本信息
  2. 搜索竞品A的定价策略
  3. 搜索竞品A的营销渠道
  4. 综合分析并输出报告

Step 2-5：依次执行，每步完成后评估是否需要调整计划

Plan 阶段（无工具调用）：
  #E1 = search("竞品A 定价")
  #E2 = search("竞品A 营销渠道")
  #E3 = analyze(#E1, #E2) → 综合报告

Execute 阶段：
  并发执行 #E1 和 #E2 → 获得结果
  执行 #E3 → 最终回答

Step 1: 用户问"如何优化数据库性能"
  ├─ 路径A：先查慢查询 → 评分 8/10
  ├─ 路径B：先查索引 → 评分 6/10
  └─ 路径C：先查连接池 → 评分 4/10
  → 选择路径A

Step 2: 基于慢查询结果
  ├─ 路径A1：添加索引 → 评分 9/10
  └─ 路径A2：重写SQL → 评分 7/10
  → 选择路径A1

def select_strategy(task_complexity: str, cost_sensitivity: str) -> str:
    if task_complexity == 'high' and cost_sensitivity == 'low':
        return 'Tree-of-Thought'
    elif task_complexity == 'high' and cost_sensitivity == 'high':
        return 'Plan-and-Solve'
    elif task_complexity == 'low' and cost_sensitivity == 'high':
        return 'ReWOO'
    else:
        return 'ReAct'  # 默认选择，平衡性好

from typing import Callable, Any, Dict, Optional
from pydantic import BaseModel, Field
import inspect

class ToolParameter(BaseModel):
    """工具参数定义。用 Pydantic 做自动校验。"""
    name: str
    type: str
    description: str  # 这个描述直接给 LLM 看，写得好坏影响调用准确率
    required: bool = True
    default: Any = None
    enum: Optional[list] = None  # 限定可选值，减少 LLM 传错参数

class ToolDefinition(BaseModel):
    """工具的完整定义。"""
    name: str
    description: str  # 包含"什么时候用"和"什么时候不要用"
    parameters: list[ToolParameter]
    func: Callable  # 实际执行的函数
    require_confirmation: bool = False  # 危险操作需要用户确认
    timeout: int = 30  # 超时时间（秒）
    retry: int = 1  # 失败重试次数

class ToolExecutor:
    """安全的工具执行器。"""

    DANGEROUS_PATTERNS = [
        r'rm\s+-rf', r'DROP\s+TABLE', r'DELETE\s+FROM',
        r'os\.system', r'subprocess', r'eval\(',
    ]

    def execute(self, tool: ToolDefinition, params: dict) -> dict:
        """执行工具调用，带安全检查和超时控制。"""
        # 1. 参数校验
        for param in tool.parameters:
            if param.required and param.name not in params:
                raise ValueError(f"缺少必要参数: {param.name}")
            if param.enum and params.get(param.name) not in param.enum:
                raise ValueError(
                    f"参数 {param.name} 值 {params.get(param.name)} 不在允许范围 {param.enum}"
                )

        # 2. 危险操作拦截
        params_str = json.dumps(params, ensure_ascii=False)
        for pattern in self.DANGEROUS_PATTERNS:
            if re.search(pattern, params_str, re.IGNORECASE):
                return {
                    'success': False,
                    'error': f'检测到危险操作模式: {pattern}',
                    'blocked': True,
                }

        # 3. 需要确认的操作
        if tool.require_confirmation:
            return {
                'success': False,
                'error': '此操作需要用户确认',
                'require_confirmation': True,
            }

        # 4. 执行（带超时和重试）
        for attempt in range(tool.retry + 1):
            try:
                import signal
                signal.alarm(tool.timeout)
                result = tool.func(**params)
                signal.alarm(0)
                return {'success': True, 'result': result, 'attempt': attempt + 1}
            except Exception as e:
                if attempt == tool.retry:
                    return {'success': False, 'error': str(e), 'attempts': attempt + 1}

        return {'success': False, 'error': '未知错误'}

人类的记忆                  Agent 的记忆
─────────                  ────────────
感官记忆（几秒）    ←→     上下文窗口（当前对话的原始Token）
工作记忆（几分钟）   ←→     对话摘要 + 关键信息提取
长期记忆（几天-年）  ←→     向量数据库 + 结构化知识库

class SlidingWindowMemory:
    def __init__(self, max_tokens: int = 8000):
        self.max_tokens = max_tokens
        self.messages = []

    def add(self, message: dict):
        self.messages.append(message)
        # 从前面截断，保证总 Token 不超限
        while self._estimate_tokens() > self.max_tokens and len(self.messages) > 2:
            self.messages.pop(1)  # 保留 system prompt

    def _estimate_tokens(self) -> int:
        return sum(len(str(m.get('content', ''))) // 2 for m in self.messages)

class SummaryMemory:
    def __init__(self, llm_client, summarize_every: int = 5):
        self.llm = llm_client
        self.summarize_every = summarize_every
        self.summary = ""
        self.recent_messages = []

    def add(self, message: dict):
        self.recent_messages.append(message)
        if len(self.recent_messages) >= self.summarize_every * 2:
            self._compress()

    def _compress(self):
        to_compress = self.recent_messages[:self.summarize_every]
        self.recent_messages = self.recent_messages[self.summarize_every:]

        compressed = self.llm.chat.completions.create(
            model='deepseek-chat',
            messages=[{
                'role': 'user',
                'content': f'将以下对话压缩为一段摘要，保留关键信息：\n{to_compress}'
            }],
        )
        self.summary = f"{self.summary}\n{compressed.choices[0].message.content}"

class LongTermMemory:
    """基于向量数据库的长期记忆。存储用户偏好、历史决策、学到的知识。"""

    def __init__(self, vector_db_client, embedding_model='text-embedding-3-small'):
        self.db = vector_db_client
        self.embedding_model = embedding_model

    def store(self, content: str, metadata: dict = None):
        """存储一条记忆。自动提取关键信息作为元数据。"""
        embedding = self._embed(content)
        self.db.insert(
            collection='agent_memory',
            vector=embedding,
            payload={
                'content': content,
                'timestamp': time.time(),
                'access_count': 0,
                **(metadata or {}),
            }
        )

    def retrieve(self, query: str, top_k: int = 5, time_decay: bool = True) -> list:
        """检索相关记忆。支持时间衰减——越久远的记忆权重越低。"""
        embedding = self._embed(query)
        results = self.db.search(
            collection='agent_memory',
            vector=embedding,
            limit=top_k * 2,  # 多取一些做重排序
        )

        # 时间衰减重排序
        now = time.time()
        for r in results:
            age_days = (now - r['payload']['timestamp']) / 86400
            decay = 1.0 / (1.0 + 0.1 * age_days) if time_decay else 1.0
            r['score'] = r['score'] * decay * (1 + 0.01 * r['payload']['access_count'])

        results.sort(key=lambda r: r['score'], reverse=True)
        return results[:top_k]

┌─────────────────────────────────────────────┐
│              混合记忆管理策略                  │
├─────────────────────────────────────────────┤
│  上下文窗口（4K Token）                       │
│  ├─ System Prompt                            │
│  ├─ 长期记忆检索结果（动态注入）              │
│  ├─ 最近 3 轮完整对话                        │
│  └─ 剩余空间 → 更早对话的摘要                │
├─────────────────────────────────────────────┤
│  摘要缓存                                    │
│  └─ 每 5 轮对话压缩一次，保持 3 轮摘要        │
├─────────────────────────────────────────────┤
│  长期记忆（向量数据库）                       │
│  ├─ 用户偏好（语言、风格、常用工具）          │
│  ├─ 历史成功决策（好的工具调用模式）          │
│  └─ 学到的知识（新的术语、修正的错误认知）    │
└─────────────────────────────────────────────┘

def build_context(self, user_message: str) -> list:
    messages = [{'role': 'system', 'content': self.system_prompt}]

    # 注入相关的长期记忆
    memories = self.long_term.retrieve(user_message, top_k=3)
    if memories:
        memory_text = '\n'.join(f"- {m['payload']['content']}" for m in memories)
        messages.append({'role': 'system', 'content': f'相关历史记忆：\n{memory_text}'})

    # 最近对话 + 摘要
    messages.extend(self.short_term.get_context())

    # 当前用户消息
    messages.append({'role': 'user', 'content': user_message})

    return messages

def reflect_on_task(self, task: str, trajectory: list, outcome: str) -> dict:
    """任务完成后的反思，提取可复用的经验。"""
    reflection_prompt = f"""评估以下 Agent 执行过程，提取经验教训：

任务：{task}
执行过程：
{trajectory}

最终结果：{outcome}

请分析：
1. 哪些步骤做得好？为什么？
2. 哪些步骤可以改进？怎么改？
3. 如果在类似任务中，应该注意什么？

以 JSON 格式输出：{{"good": [...], "improve": [...], "lesson": "..."}}"""

    reflection = self.llm.chat.completions.create(
        model='deepseek-chat',
        messages=[{'role': 'user', 'content': reflection_prompt}],
        temperature=0.1,
    )

    lesson = json.loads(reflection.choices[0].message.content)

    # 存入长期记忆，供未来任务参考
    self.long_term.store(
        content=f'任务：{task}\n教训：{lesson["lesson"]}',
        metadata={'type': 'reflection', 'task': task, 'outcome': outcome}
    )

    return lesson

class AgentRateLimiter:
    def __init__(self, max_steps_per_task=15, max_tokens_per_task=50000,
                 max_cost_per_task=0.50, cooldown_seconds=60):
        self.max_steps = max_steps_per_task
        self.max_tokens = max_tokens_per_task
        self.max_cost = max_cost_per_task
        self.cooldown = cooldown_seconds
        self.task_count = 0
        self.last_task_time = 0

    def can_proceed(self, task_stats: dict) -> tuple[bool, str]:
        """检查是否可以继续执行。"""
        if task_stats['steps'] >= self.max_steps:
            return False, f'超过最大步数 {self.max_steps}'
        if task_stats['total_tokens'] >= self.max_tokens:
            return False, f'超过最大 Token {self.max_tokens}'
        if task_stats['total_cost'] >= self.max_cost:
            return False, f'超过最大费用 ¥{self.max_cost}'

        now = time.time()
        if now - self.last_task_time  dict:
        """获取执行摘要。"""
        return {
            'task_id': self.task_id,
            'total_steps': len(self.traces),
            'total_tokens': sum(t['tokens'] for t in self.traces),
            'total_cost': sum(t['cost'] for t in self.traces),
            'total_latency_ms': sum(t['latency_ms'] for t in self.traces),
            'avg_tokens_per_step': sum(t['tokens'] for t in self.traces) / max(len(self.traces), 1),
        }

def __init__(self, approval_threshold: float = 0.7):
    self.threshold = approval_threshold  # 低于此置信度的决策需要确认

def should_ask_human(self, action: dict, confidence: float) -> bool:
    # 高风险操作：总是确认
    if action['tool'] in ['delete_file', 'run_shell', 'send_email', 'make_payment']:
        return True
    # 低置信度：需要确认
    if confidence  如果觉得有用，欢迎 **点赞 + 收藏 + 关注**。后续会继续拆解 RAG 系统、MCP 协议、多 Agent 协作等核心主题，全部附可运行代码。