7 月 Agent 开发月度回顾：从架构设计到性能优化再到生产落地

发布时间：2026/8/1 1:18:13

7 月 Agent 开发月度回顾：从架构设计到性能优化再到生产落地

一、深度引言与场景痛点

7 月折腾了整整一个月 Agent，从月初信心满满地画架构图，到月中被各种 corner case 打得鼻青脸肿，再到月底终于跑通了第一个生产级 Agent 流水线——这一路，跌宕起伏。

先说最疼的几个坑：

第一个坑是记忆管理。Agent 跟用户多轮对话之后，上下文窗口爆炸式增长，Token 成本线性飙升。月初我们用的是 naive 的全量上下文方案，跑到第 8 轮对话时，单次请求的 Token 量已经突破 12000，响应延迟奔着 8 秒去了。用户那边卡得怀疑人生，我们也怀疑人生。

第二个坑是工具调用链路的不确定性。Agent 调用外部 API 时，偶尔超时、偶尔返回格式异常、偶尔下游服务挂掉。每次出错，Agent 就直接"思考"到超时，然后丢一条 Sorry。没有优雅降级，没有重试策略，没有 fallback 路径——说白了就是裸奔。

第三个坑是多 Agent 协作时的消息路由。三个 Agent（意图识别Agent、执行Agent、总结Agent）串行工作，每个环节都可能成为瓶颈。更致命的是，并行场景完全没考虑，白白浪费了时间。

二、底层机制与原理深度剖析

先上一张我们最终稳定下来的 Agent 架构图：

这个架构的核心设计哲学是分层解耦 + 可观测。每一层都可以独立迭代、独立监控、独立扩容。

路由层的安全护栏是整个系统的第一道防线。我们用了关键词 + 语义双模式检测，既保证了常见攻击词的拦截效率，又能通过 embedding 相似度捕获变体攻击。

记忆层的总结器是解决 Token 爆炸的关键。我们不是简单截断历史消息，而是用一个小模型（qwen2-7b 就很够用）对历史对话做渐进式摘要。每 5 轮对话触发一次总结，将原文替换为摘要，上下文窗口从 12000 Token 降到了 3000 Token 左右。

执行层的审计 Agent是这次迭代最大的创新点。传统 Agent 执行完工具调用就直接返回结果，如果返回的是错误数据，用户一脸懵。现在我们加了一个轻量级审计节点，对执行结果做 schema 校验 + 语义合理性检查，不通过就触发重规划。

三、生产级代码实现

以下是核心的 Agent 调度器实现，带完整的异常处理和优雅降级：

import asyncio import json from dataclasses import dataclass, field from enum import Enum from typing import Any, Optional import structlog from tenacity import ( retry, stop_after_attempt, wait_exponential, retry_if_exception_type, ) logger = structlog.get_logger() class AgentState(Enum): IDLE = "idle" PLANNING = "planning" EXECUTING = "executing" AUDITING = "auditing" FALLBACK = "fallback" DONE = "done" ERROR = "error" @dataclass class AgentContext: """Agent 执行上下文，贯穿整个调用链路。""" session_id: str user_input: str history: list[dict] = field(default_factory=list) plan: list[dict] = field(default_factory=list) tool_results: list[dict] = field(default_factory=list) state: AgentState = AgentState.IDLE error_count: int = 0 max_retries: int = 3 class AgentPipelineError(Exception): """Agent 流水线基础异常。""" pass class ToolExecutionError(AgentPipelineError): """工具调用失败异常，会触发重试。""" pass class AuditFailureError(AgentPipelineError): """审计不通过异常，会触发重规划。""" pass class AgentOrchestrator: """Agent 编排器：负责整个 Agent 流水线的调度与容错。""" def __init__( self, guardrail_threshold: float = 0.75, summary_interval: int = 5, ): self.guardrail_threshold = guardrail_threshold self.summary_interval = summary_interval self.fallback_response = "抱歉，我暂时无法处理这个请求。请稍后重试或换一种方式描述你的需求。" async def run(self, ctx: AgentContext) -> str: """主入口：运行完整的 Agent 流水线。""" try: # 1. 安全护栏 if not await self._run_guardrail(ctx): logger.warning("guardrail_blocked", session_id=ctx.session_id) return "抱歉，你的请求包含不安全的内容，我无法处理。" # 2. 意图路由 intent = await self._route_intent(ctx) # 3. 记忆管理（含渐进式总结） ctx = await self._manage_memory(ctx) # 4. 规划 -> 执行 -> 审计 循环 ctx.state = AgentState.PLANNING for attempt in range(ctx.max_retries): try: plan = await self._plan(ctx, intent) ctx.plan = plan ctx.state = AgentState.EXECUTING result = await self._execute_plan(ctx) ctx.tool_results = result ctx.state = AgentState.AUDITING if await self._audit(ctx): ctx.state = AgentState.DONE return await self._format_response(ctx) else: logger.warning( "audit_failed", session_id=ctx.session_id, attempt=attempt + 1, ) ctx.error_count += 1 continue except (ToolExecutionError, AuditFailureError) as e: logger.error( "pipeline_error", session_id=ctx.session_id, error=str(e), attempt=attempt + 1, ) ctx.error_count += 1 if ctx.error_count >= ctx.max_retries: break await asyncio.sleep(1.0 * (attempt + 1)) # 线性退避 # 5. 优雅降级：所有重试耗尽 ctx.state = AgentState.FALLBACK logger.error( "pipeline_exhausted", session_id=ctx.session_id, error_count=ctx.error_count, ) return self.fallback_response except Exception as e: ctx.state = AgentState.ERROR logger.exception( "unexpected_error", session_id=ctx.session_id, error=str(e), ) return self.fallback_response async def _run_guardrail(self, ctx: AgentContext) -> bool: """安全护栏检查。""" blocked_patterns = [ "DROP TABLE", "<script>", "../", "__import__", "eval(" ] input_lower = ctx.user_input.lower() for pattern in blocked_patterns: if pattern.lower() in input_lower: return False return True async def _route_intent(self, ctx: AgentContext) -> str: """意图识别与路由。""" # 简化版：实际项目中替换为 LLM 调用 intents = { "查询": ["查", "搜索", "找", "是什么"], "执行": ["帮我", "执行", "运行", "创建"], "分析": ["分析", "总结", "对比", "评估"], } for intent_name, keywords in intents.items(): if any(kw in ctx.user_input for kw in keywords): return intent_name return "通用" async def _manage_memory(self, ctx: AgentContext) -> AgentContext: """记忆管理：控制上下文窗口大小。""" if len(ctx.history) >= self.summary_interval: # 触发渐进式摘要，用小模型压缩历史 recent = ctx.history[-self.summary_interval :] older = ctx.history[: -self.summary_interval] summary_prompt = ( "将以下对话历史压缩为一段简洁的摘要，" "保留关键信息和用户偏好：\n" + json.dumps(older, ensure_ascii=False) ) # 实际项目中这里调用 LLM summary ctx.history = [ {"role": "system", "content": f"历史摘要：{summary_prompt[:200]}..."} ] + recent logger.info( "memory_summarized", session_id=ctx.session_id, original_len=len(ctx.history) + self.summary_interval, compressed_len=len(ctx.history), ) return ctx async def _plan(self, ctx: AgentContext, intent: str) -> list[dict]: """规划阶段：将用户意图拆解为可执行的步骤。""" # 实际项目中调用 LLM 生成计划 return [ {"tool": "search", "args": {"query": ctx.user_input}}, {"tool": "fetch", "args": {"url": "{{result.url}}"}}, ] @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10), retry=retry_if_exception_type(ToolExecutionError), ) async def _execute_tool(self, tool_call: dict) -> dict: """执行单个工具调用，带指数退避重试。""" # 实际项目中对接到真实的工具网关 tool_name = tool_call.get("tool", "unknown") logger.info("tool_executing", tool=tool_name) # 模拟工具调用 await asyncio.sleep(0.2) return {"tool": tool_name, "status": "success", "data": "mock_result"} async def _execute_plan(self, ctx: AgentContext) -> list[dict]: """批量执行计划中的所有步骤。""" tasks = [self._execute_tool(step) for step in ctx.plan] results = await asyncio.gather(*tasks, return_exceptions=True) output = [] for i, result in enumerate(results): if isinstance(result, Exception): logger.error( "tool_failed", step=i, error=str(result), ) output.append({ "tool": ctx.plan[i].get("tool", "unknown"), "status": "error", "error": str(result), }) else: output.append(result) return output async def _audit(self, ctx: AgentContext) -> bool: """审计执行结果。""" for result in ctx.tool_results: if result.get("status") == "error": return False if "data" not in result or result["data"] is None: return False return True async def _format_response(self, ctx: AgentContext) -> str: """格式化最终回复。""" data_parts = [] for r in ctx.tool_results: if r.get("status") == "success": data_parts.append(str(r.get("data", ""))) return "\n".join(data_parts) if data_parts else self.fallback_response

四、边界分析与架构权衡

这个月积累下来的几个关键 trade-off：

1. 延迟 vs 准确性
审计 Agent 是双刃剑。加了审计，P99 延迟多出 300-500ms，但错误率从 8% 降到了 1.2%。如果你的场景是客服对话，这个延迟代价完全值得；如果是实时推荐，可能需要换个方案——用采样审计而非全量审计。

2. 渐进式摘要 vs 上下文完整性
每 5 轮触发摘要确实丢了部分细节信息。实测下来，摘要后的上下文在 DSTC 类型任务上准确率下降约 3%，但在 open-ended 对话中几乎无感知。如果你的业务场景需要精确的多轮状态追踪，摘要间隔可以拉长到 8-10 轮，或者引入分层摘要（关键轮次保留原文）。

3. 串行编排 vs 并行编排
目前是串行编排（Planner → Executor → Auditor），简单可靠。但某些独立工具调用天然可并行（比如同时查天气和查新闻）。下个月计划引入 DAG 级别的任务编排，将无依赖的工具调用并行化。

4. 错误重试 vs 快速失败
tenacity的指数退避重试在小概率故障场景效果好，但如果下游服务已经彻底挂了，重试只是浪费时间和资源。更优的策略是引入熔断器模式——连续 N 次失败后直接走 fallback。