Agent 产品化的项目复盘：从技术 Demo 到可交付产品的十个关键决策
一、引言
做技术的人容易陷入一个误区：把 Demo 跑通就当成了项目完成。去年参与过一个 Agent 产品的完整交付周期，让我深刻认识到，从一段能跑的 Python 脚本到客户愿意付费的产品，中间隔着的不是代码量，而是十个需要反复权衡的工程决策。这篇文章是对那个项目的复盘，不谈算法创新，只谈落地过程中的真实取舍。
项目背景不算复杂：为一家中型企业的客服部门构建一个智能工单分派 Agent。它需要理解用户意图、匹配历史相似工单、推荐处理方案给人工坐席。技术栈上，我们用 LangChain 做 Agent 编排，FastAPI 做服务层，前端一个轻量 Vue3 面板做坐席交互。听起来很标准的 LLM 应用，但真正扎进去才发现，Demo 和产品的差异不在模型能力上，而在工程基础设施。
下面按照决策的先后顺序，逐一复盘。
二、十个关键决策的复盘
决策一：Agent 架构选型——ReAct 还是 Plan-and-Execute
Demo 阶段用 ReAct 模式跑得很流畅，推理-行动循环能处理大多数单步查询。但上了真实业务数据后，问题暴露了：一个工单分派可能需要查客户历史记录、查产品知识库、查 SLA 规则、生成推荐——这四步如果在单一 ReAct 循环里串行，延迟直接飙到 15 秒以上。
最终选了 Plan-and-Execute + ReAct 的混合模式：先用一个轻量模型做计划分解，把复杂任务拆成子任务列表，再用 ReAct 执行每个子任务，中间结果通过共享上下文传递。
# 计划生成阶段的 Prompt 模板 PLAN_PROMPT = """你是一个工单分派计划生成器。给定以下工单信息，请将处理流程分解为步骤列表。 每个步骤必须包含：步骤名称、所需工具、预期输出。 工单内容：{ticket_content} 客户历史：{customer_context} 请严格按 JSON 格式输出： [ {"step": "意图识别", "tool": "intent_classifier", "output": "intent_label"}, {"step": "历史匹配", "tool": "similarity_search", "output": "top5_tickets"}, ... ] """这个决策的核心收益不是准确率提升，而是可观测性。计划步骤拆开后，每一步的延迟、成功率、异常都可以独立监控，出问题时能快速定位到具体阶段。
决策二：Prompt 管理——代码内嵌还是配置化
Demo 时所有 Prompt 硬编码在代码里，改一个字就要重新部署。产品化后 Prompt 迭代频率远超代码变更频率，而且不同客户可能有不同的 Prompt 变体。
方案是建一个 Prompt 管理中心，用 YAML 文件管理 Prompt 模板，支持变量注入和版本控制。
# prompts/ticket_assignment/v2.1.yaml version: "2.1" system: | 你是一个{domain}领域的工单分派助手。 当前角色：{agent_role} 处理规则： {rules} **重要**：如果置信度低于0.6，需要标记为"需人工确认"。 user_template: | 工单ID：{ticket_id} 客户等级：{customer_level} 问题描述：{description} 请完成以下步骤： 1. 识别问题类别 2. 匹配历史相似工单（至少3条） 3. 推荐处理方案配合一个简单的 Python 加载器：
import yaml from pathlib import Path from string import Template class PromptLoader: def __init__(self, base_path: str = "prompts"): self.base = Path(base_path) self._cache = {} def load(self, name: str, version: str = "latest") -> dict: cache_key = f"{name}:{version}" if cache_key not in self._cache: file_path = self.base / name / f"{version}.yaml" if not file_path.exists(): raise FileNotFoundError(f"Prompt {name}:{version} not found") with open(file_path) as f: self._cache[cache_key] = yaml.safe_load(f) return self._cache[cache_key] def render(self, name: str, variables: dict, version: str = "latest") -> tuple[str, str]: prompt = self.load(name, version) system = Template(prompt["system"]).safe_substitute(variables) user = Template(prompt["user_template"]).safe_substitute(variables) return system, user决策三：工具调用的可靠性保障
Agent 的灵魂是工具调用，但也是出错最多的地方。知识库检索接口偶尔超时、数据库连接池耗尽、第三方 API 限流——这些在 Demo 阶段一个 try-catch 就能兜底，但产品里必须建立系统性的容错机制。
核心策略三板斧：超时控制 + 重试退避 + 降级兜底。
import asyncio from typing import Optional, Callable, Any from functools import wraps class ToolExecutor: def __init__( self, timeout: float = 10.0, max_retries: int = 2, base_delay: float = 1.0 ): self.timeout = timeout self.max_retries = max_retries self.base_delay = base_delay async def execute( self, tool_fn: Callable, args: tuple = (), kwargs: dict = None, fallback: Any = None ) -> dict: kwargs = kwargs or {} last_error = None for attempt in range(self.max_retries + 1): try: result = await asyncio.wait_for( tool_fn(*args, **kwargs), timeout=self.timeout ) return {"status": "success", "data": result, "attempts": attempt + 1} except asyncio.TimeoutError: last_error = "timeout" except Exception as e: last_error = str(e) if attempt < self.max_retries: delay = self.base_delay * (2 ** attempt) await asyncio.sleep(delay) return { "status": "fallback", "data": fallback, "error": last_error, "attempts": self.max_retries + 1 }决策四：上下文窗口管理
Agent 对话越长，上下文窗口消耗越大，不仅成本上升，推理质量也会下降。真实场景中，一个工单处理可能涉及 10 轮以上的交互。
采用滑动窗口 + 摘要压缩策略：
- 保留最近 5 轮完整对话
- 更早的交互自动压缩为摘要
- 关键信息（客户ID、工单号、分类结果）进入结构化状态
from dataclasses import dataclass, field from typing import List @dataclass class ConversationState: ticket_id: str = "" customer_id: str = "" intent: str = "unknown" confidence: float = 0.0 resolved_step_ids: List[str] = field(default_factory=list) key_findings: List[str] = field(default_factory=list) class ContextManager: MAX_RECENT_TURNS = 5 def __init__(self, summarizer): self.summarizer = summarizer self.state = ConversationState() self.history = [] async def add_turn(self, role: str, content: str) -> None: self.history.append({"role": role, "content": content}) if len(self.history) > self.MAX_RECENT_TURNS * 2: old_turns = self.history[:-self.MAX_RECENT_TURNS * 2] summary = await self.summarizer.summarize(old_turns) self.history = [{"role": "system", "content": f"历史摘要：{summary}"}] + \ self.history[-self.MAX_RECENT_TURNS * 2:] def build_context(self) -> List[dict]: state_desc = f"当前状态：{self.state.__dict__}" return [ {"role": "system", "content": state_desc}, *self.history ]决策五：评估体系
Demo 阶段靠"感觉"判断 Agent 好不好用，产品化后必须有量化指标。建立了三层评估体系：
- 工具级：每次工具调用的成功率、P50/P95 延迟
- 任务级：工单分类准确率、方案推荐命中率
- 业务级：人工坐席采纳率、处理时长缩减比例
每一层对应不同的告警阈值和优化策略。
决策六到十（简表）
考虑到篇幅，剩余五个决策以要点形式复盘：
| 决策 | 核心问题 | 最终方案 | 关键教训 |
|---|---|---|---|
| 六、多租户隔离 | 不同客户的数据和配置隔离 | 数据库级隔离 + Prompt 模板级别隔离 | 不要试图用代码逻辑区分租户 |
| 七、流式响应 | 长任务必须有进度反馈 | SSE 推送步骤进度，前端状态机驱动 | WebSocket 太重，SSE 够用且运维简单 |
| 八、成本控制 | LLM API 费用增长不可控 | 分级模型策略：规划用小模型、执行用大模型 | 90% 的任务可以用小模型处理 |
| 九、权限与安全 | Agent 可以执行哪些操作 | 白名单机制：工具调用前校验权限范围 | 永远不要给 Agent 写数据库的权限 |
| 十、部署与运维 | 如何做到零停机更新 | 蓝绿部署 + Prompt 热加载 | Prompt 变更不需要重启服务 |
三、架构总览
整个系统的最终架构如下：
四、关键踩坑记录
坑一：ReAct 的幻觉循环。早期版本里 Agent 有时会陷入"调用工具→得到空结果→再次调用同一工具"的死循环。解法是加最大步数限制和一个简单的去重检测——连续两次相同工具调用直接中断，返回兜底方案。
坑二：Prompt 版本管理混乱。多人协作时经常发生 A 改了 Prompt、B 不知道、线上出现不一致。后来强制要求所有 Prompt 变更走 PR 流程，并且部署脚本自动比对线上版本与仓库版本的差异。
坑三：成本失控的恐慌。上线第一周，GPT-4 API 费用超出预期 3 倍。排查后发现是某些边缘 Case 触发了过长的 Agent 推理链。通过加 Token 消耗预算上限和模型降级策略，成本回归到可控范围。
五、结语
从 Demo 到产品的路，本质上是在可靠性、成本、体验三者之间找平衡。技术方案没有绝对的对错，关键在于在特定场景和资源约束下做出合适的取舍。
十个决策复盘下来，最核心的一点是：越早建立可观测性和评估体系，越能避免凭感觉做决策。数据会告诉你哪里该投入、哪里可以妥协。Agent 产品化不是技术的终点，而是工程化的起点——这一课，写在这个项目的每一个深夜调试和每一次线上事故里。
本文基于真实项目经验撰写，技术栈版本：Python 3.12 / FastAPI 0.110 / LangChain 0.1 / Vue 3.4
