Agent 产品化的项目复盘：从技术 Demo 到可交付产品的十个关键决策-开发者社区

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 变更不需要重启服务