从 Prompt 到 Workflow:面向中级开发者的 AI Agent 实战设计与落地指南
很多团队一开始做 AI 应用,路径都很像:
- 先写一个 Prompt;
- 接上大模型 API;
- 演示时效果不错;
- 一上生产,开始出现“偶尔能用,但总觉得不稳”。
我自己做这类系统时,最深的感受是:Prompt 解决的是“如何让模型回答”,Workflow 解决的是“如何让系统可靠完成任务”。二者不是替代关系,而是层级关系。Prompt 是最小控制单元,Workflow 是把模型、工具、状态、规则和异常处理编排起来的执行系统。
这篇文章不讲“Agent 很厉害”的空话,而是从工程落地角度,带你把一个“单轮问答”逐步升级成“可观测、可重试、可扩展”的 AI Agent 工作流。
背景与问题
为什么单个 Prompt 很快会碰到天花板
对于中级开发者来说,Prompt 工程并不陌生。你可能已经写过这样的东西:
- 指定角色:你是资深客服;
- 指定格式:请输出 JSON;
- 指定约束:不要编造信息;
- 增加 few-shot 示例。
这些手段在简单、封闭、单步任务里很好用,比如:
- 摘要
- 分类
- 改写
- 提取结构化信息
但当任务变成下面这种形式时,单个 Prompt 就开始吃力:
- “读取用户需求 → 查询知识库 → 判断是否需要调用外部系统 → 汇总结果 → 生成可执行建议”
- “解析工单 → 判断优先级 → 匹配历史方案 → 生成回复 → 如果置信度低则转人工”
- “读取报错日志 → 定位模块 → 生成排查步骤 → 调用诊断脚本 → 汇总结论”
这类任务的问题,不在于模型“不聪明”,而在于它本质上已经不是一次生成,而是一个多步骤决策与执行过程。
生产环境里最常见的四类问题
1. 输出不稳定
同样输入,多次调用结果不一致。演示时像魔法,上线后像抽奖。
2. 上下文失控
对话一长,模型开始遗忘关键约束;或者上下文越来越贵,延迟和成本一起飙升。
3. 工具调用不可控
模型可能错误选择工具、重复调用工具,甚至把不该执行的参数传出去。
4. 出错无法定位
当结果不对时,你不知道问题出在:
- Prompt 设计
- 检索召回
- 工具接口
- 编排逻辑
- 模型能力边界
这也是很多团队从“Prompt 工程”走向“Workflow 编排”的直接原因。
核心原理
从 Prompt 到 Workflow,本质上发生了什么
可以把两者理解成两个层级:
- Prompt 层:定义单步行为
- Workflow 层:定义多步协作
一个更工程化的视角是:
| 层级 | 关注点 | 典型问题 |
|---|---|---|
| Prompt | 怎么说清楚任务 | 输出格式、语气、约束、示例 |
| Tool | 怎么接入外部能力 | 检索、数据库、HTTP API、脚本 |
| Memory/State | 怎么保存过程信息 | 对话状态、任务状态、阶段结果 |
| Workflow | 怎么组织多步执行 | 分支、重试、回退、审批、人机协同 |
| Observability | 怎么监控与排查 | 日志、trace、指标、样本回放 |
换句话说,Agent 不是一个“更长的 Prompt”,而是一套把模型纳入软件系统的运行机制。
一个实用的 Agent 架构
下面这个架构适合大多数中等复杂度业务场景:
flowchart TD
A[用户请求] --> B[输入预处理]
B --> C[任务路由器]
C --> D[规划器 Planner]
D --> E[工具执行器 Tool Executor]
E --> F[状态存储 State]
F --> G[结果汇总器]
G --> H[输出后处理]
H --> I[用户响应]
E --> J[知识库/搜索]
E --> K[业务 API]
E --> L[代码执行/脚本]
C --> M[降级策略]
D --> N[人工审核]
这个架构里,真正重要的不是“Planner 够不够智能”,而是每个节点都能被约束、记录和替换。
设计 Agent 时,我建议先回答 5 个问题
1. 任务是开放式还是收敛式?
- 开放式:如创意写作、头脑风暴
- 收敛式:如工单分类、SQL 生成、数据抽取
如果是收敛式任务,应该优先把输出边界定义死,而不是追求“像人”。
2. 模型到底负责什么?
不要把所有逻辑都扔给模型。通常更合理的分工是:
- 模型负责:理解、判断、生成、规划
- 程序负责:校验、执行、权限、重试、持久化
3. 哪些步骤必须可重试?
例如:
- 检索失败可重试
- 外部 API 超时可重试
- LLM 输出解析失败可重试
- 涉及写操作的步骤不能盲目重试,要有幂等设计
4. 哪些环节需要“确定性”?
模型天然是概率系统,但业务流程里有些地方必须确定:
- JSON 格式校验
- 参数范围校验
- 工具权限校验
- 状态机迁移规则
5. 失败时如何降级?
成熟系统不是“永不失败”,而是“失败可控”:
- 低置信度时转人工
- 工具失败时返回保守结果
- 检索失败时只基于已有上下文回答
- 超时后返回部分结果
方案对比:三种常见落地路径
方案一:大 Prompt 单体模式
做法:把角色、规则、流程、格式、工具说明全塞进一个 Prompt。
优点:
- 开发快
- 原型验证方便
- 适合 Demo
缺点:
- 难维护
- 难调试
- 难扩展
- 随上下文增长而变脆
适用场景:
- 1~2 周内验证价值
- 低风险内部工具
- 单步任务
方案二:Prompt + 工具调用
做法:模型决定是否调用工具,程序负责执行工具并回填结果。
优点:
- 功能明显增强
- 可接入真实数据
- 架构相对简单
缺点:
- 工具选择容易失控
- 上下文管理变复杂
- 很多业务规则仍混在 Prompt 中
适用场景:
- FAQ + 检索
- 查询类助手
- 运维诊断助手
方案三:状态驱动的 Workflow Agent
做法:将任务拆成显式状态和步骤,由编排层控制流转,模型只参与需要语义理解的节点。
优点:
- 稳定
- 可观测
- 易于审计
- 易于扩展
缺点:
- 初期设计成本更高
- 需要更明确的任务建模
适用场景:
- 工单处理
- 业务审批
- 复杂客服
- 多工具协同系统
如果你已经是中级开发者,我的建议很直接:不要停留在“大 Prompt 单体模式”太久。原型可以这么做,但一旦要上线,尽快向“状态驱动 Workflow”迁移。
核心原理拆解:Prompt、Tool、State、Workflow 如何协同
1. Prompt 是“局部智能”,不是“全局控制器”
Prompt 最适合做这些事:
- 意图识别
- 任务分类
- 文本抽取
- 参数补全
- 结果总结
但不适合承担:
- 权限控制
- 流程状态迁移
- 资金、库存、订单等关键业务规则
2. Tool 是模型接触真实世界的手
没有 Tool,Agent 只能“说”;有了 Tool,它才能“做”。
典型工具包括:
- 搜索/向量检索
- 数据库查询
- HTTP API
- 本地脚本
- 代码执行沙箱
关键原则是:工具描述给模型看,工具权限由系统决定。
3. State 是 Workflow 的骨架
很多失败的 Agent 项目,问题不是模型不行,而是没有状态管理。只靠对话历史堆上下文,迟早出事。
你至少要区分三种状态:
- 会话状态:用户是谁、当前目标是什么
- 任务状态:进行到哪一步、每一步结果是什么
- 系统状态:重试次数、超时标记、人工接管标记
4. Workflow 是把不确定性关进笼子里
Workflow 的作用不是让模型更自由,而是让系统更可控。常见控制手段包括:
- 显式步骤拆分
- 条件分支
- 重试与超时
- 输出校验
- 人工审核点
- 回滚与补偿
下面用时序图看一次典型执行流程。
sequenceDiagram
participant U as 用户
participant W as Workflow
participant L as LLM
participant T as Tool
participant S as State Store
U->>W: 提交工单描述
W->>L: 分类与提取关键信息
L-->>W: 工单类型/优先级/参数
W->>S: 保存阶段结果
W->>T: 查询知识库或历史工单
T-->>W: 返回候选方案
W->>L: 基于候选方案生成处理建议
L-->>W: 建议 + 置信度
W->>S: 更新状态
alt 置信度高
W-->>U: 自动回复建议
else 置信度低
W-->>U: 转人工并附带摘要
end
实战代码(可运行)
下面我们用 Python 做一个最小可运行示例:工单助手 Agent。
这个例子不会直接依赖真实大模型 API,这样你复制下来就能跑。为了体现架构思想,我会用一个“FakeLLM”模拟模型输出,重点放在:
- Prompt 组织
- 工具调用
- 状态管理
- Workflow 编排
- 校验与降级
目标场景
输入一段用户报障描述,例如:
“支付页面一直报 502,已经持续 10 分钟,影响下单。”
系统完成:
- 意图识别与字段提取;
- 查询知识库;
- 生成处理建议;
- 根据置信度决定自动回复还是转人工。
项目结构
agent_demo/
├── app.py
└── requirements.txt
requirements.txt
这个示例只用标准库,其实可以为空。为了形式完整,给一个文件:
# no external dependencies
app.py
import json
import re
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional
class TaskStatus(str, Enum):
INIT = "INIT"
CLASSIFIED = "CLASSIFIED"
RETRIEVED = "RETRIEVED"
GENERATED = "GENERATED"
COMPLETED = "COMPLETED"
HUMAN_HANDOFF = "HUMAN_HANDOFF"
FAILED = "FAILED"
@dataclass
class AgentState:
request_id: str
user_input: str
status: TaskStatus = TaskStatus.INIT
ticket_info: Dict[str, Any] = field(default_factory=dict)
kb_results: List[Dict[str, Any]] = field(default_factory=list)
suggestion: Optional[str] = None
confidence: float = 0.0
retry_count: int = 0
logs: List[str] = field(default_factory=list)
def log(self, message: str):
ts = time.strftime("%Y-%m-%d %H:%M:%S")
self.logs.append(f"[{ts}] {message}")
class FakeLLM:
"""
用规则模拟 LLM,方便本地直接运行。
实际项目里你可以替换成 OpenAI / Azure / 其他模型 SDK。
"""
def classify_ticket(self, text: str) -> Dict[str, Any]:
text_lower = text.lower()
category = "general"
priority = "medium"
if any(k in text_lower for k in ["502", "504", "超时", "错误", "报错"]):
category = "incident"
if any(k in text_lower for k in ["支付", "下单", "订单"]):
category = "payment_incident"
if any(k in text_lower for k in ["一直", "持续", "全部用户", "无法", "影响"]):
priority = "high"
duration_match = re.search(r"(\d+)\s*分钟", text)
duration_min = int(duration_match.group(1)) if duration_match else None
return {
"category": category,
"priority": priority,
"duration_min": duration_min,
"raw_summary": text[:120]
}
def generate_suggestion(self, ticket_info: Dict[str, Any], kb_results: List[Dict[str, Any]]) -> Dict[str, Any]:
category = ticket_info.get("category", "general")
priority = ticket_info.get("priority", "medium")
if not kb_results:
return {
"suggestion": "未检索到可靠方案,建议先收集错误日志、时间范围、影响范围,并转人工排查。",
"confidence": 0.35
}
top = kb_results[0]
base = f"建议优先参考知识库方案:{top['title']}。处理步骤:{top['solution']}"
if category == "payment_incident":
base += " 同时建议检查支付网关健康状态、上游依赖和最近发布记录。"
confidence = 0.85 if priority == "high" and category in ["incident", "payment_incident"] else 0.72
return {
"suggestion": base,
"confidence": confidence
}
class KnowledgeBaseTool:
def __init__(self):
self.docs = [
{
"title": "502 网关错误排查",
"keywords": ["502", "网关", "超时", "bad gateway"],
"solution": "检查反向代理、上游服务存活、网络连通性与超时配置。"
},
{
"title": "支付服务异常处理",
"keywords": ["支付", "下单", "订单", "网关"],
"solution": "检查支付网关、订单服务、库存服务与调用链日志,确认是否有版本发布。"
},
{
"title": "通用报错信息收集",
"keywords": ["报错", "错误", "异常"],
"solution": "收集 traceId、错误码、影响范围、开始时间和最近变更记录。"
}
]
def search(self, query: str) -> List[Dict[str, Any]]:
scores = []
q = query.lower()
for doc in self.docs:
score = sum(1 for kw in doc["keywords"] if kw.lower() in q)
if score > 0:
scores.append((score, doc))
scores.sort(key=lambda x: x[0], reverse=True)
return [doc for _, doc in scores[:3]]
class TicketAgentWorkflow:
def __init__(self, llm: FakeLLM, kb_tool: KnowledgeBaseTool):
self.llm = llm
self.kb_tool = kb_tool
def run(self, request_id: str, user_input: str) -> AgentState:
state = AgentState(request_id=request_id, user_input=user_input)
state.log("Workflow started")
try:
self._classify(state)
self._retrieve(state)
self._generate(state)
self._finalize(state)
except Exception as e:
state.status = TaskStatus.FAILED
state.log(f"Workflow failed: {e}")
return state
def _classify(self, state: AgentState):
state.log("Classifying ticket")
ticket_info = self.llm.classify_ticket(state.user_input)
self._validate_ticket_info(ticket_info)
state.ticket_info = ticket_info
state.status = TaskStatus.CLASSIFIED
state.log(f"Classified result: {json.dumps(ticket_info, ensure_ascii=False)}")
def _retrieve(self, state: AgentState):
state.log("Retrieving knowledge base")
query = f"{state.user_input} {state.ticket_info.get('category', '')}"
results = self.kb_tool.search(query)
state.kb_results = results
state.status = TaskStatus.RETRIEVED
state.log(f"Retrieved {len(results)} documents")
def _generate(self, state: AgentState):
state.log("Generating suggestion")
result = self.llm.generate_suggestion(state.ticket_info, state.kb_results)
self._validate_generation(result)
state.suggestion = result["suggestion"]
state.confidence = result["confidence"]
state.status = TaskStatus.GENERATED
state.log(f"Generated confidence={state.confidence}")
def _finalize(self, state: AgentState):
state.log("Finalizing result")
if state.confidence >= 0.75:
state.status = TaskStatus.COMPLETED
state.log("Auto response selected")
else:
state.status = TaskStatus.HUMAN_HANDOFF
state.log("Human handoff selected")
@staticmethod
def _validate_ticket_info(ticket_info: Dict[str, Any]):
required = ["category", "priority", "raw_summary"]
for key in required:
if key not in ticket_info:
raise ValueError(f"Missing field in ticket_info: {key}")
if ticket_info["priority"] not in ["low", "medium", "high"]:
raise ValueError("Invalid priority")
@staticmethod
def _validate_generation(result: Dict[str, Any]):
if "suggestion" not in result or "confidence" not in result:
raise ValueError("Invalid generation result")
if not isinstance(result["confidence"], (int, float)):
raise ValueError("Confidence must be numeric")
if not 0 <= result["confidence"] <= 1:
raise ValueError("Confidence out of range")
def pretty_print(state: AgentState):
print("=" * 60)
print("request_id:", state.request_id)
print("status:", state.status.value)
print("ticket_info:", json.dumps(state.ticket_info, ensure_ascii=False, indent=2))
print("kb_results:", json.dumps(state.kb_results, ensure_ascii=False, indent=2))
print("confidence:", state.confidence)
print("suggestion:", state.suggestion)
print("logs:")
for log in state.logs:
print(" -", log)
print("=" * 60)
if __name__ == "__main__":
llm = FakeLLM()
kb_tool = KnowledgeBaseTool()
workflow = TicketAgentWorkflow(llm, kb_tool)
demo_input = "支付页面一直报 502,已经持续 10 分钟,影响下单。"
state = workflow.run(request_id="req-1001", user_input=demo_input)
pretty_print(state)
运行方式
python app.py
你会看到什么
程序会输出:
- 工单分类结果
- 命中的知识库文档
- 最终建议
- 置信度
- 自动回复还是转人工
- 整个执行日志
这个例子虽然是“最小实现”,但已经体现了一个重要思想:不要让模型直接从输入跳到最终答案,中间要有可检查、可记录的步骤。
把示例升级成真实生产架构
上面的代码只是骨架。真到生产环境,通常会继续演进成下面这种状态机。
stateDiagram-v2
[*] --> INIT
INIT --> CLASSIFIED: 提取意图/参数
CLASSIFIED --> RETRIEVED: 知识检索
RETRIEVED --> GENERATED: 生成方案
GENERATED --> COMPLETED: 置信度高
GENERATED --> HUMAN_HANDOFF: 置信度低
INIT --> FAILED: 输入非法
CLASSIFIED --> FAILED: 输出校验失败
RETRIEVED --> FAILED: 工具异常且不可恢复
GENERATED --> FAILED: 生成结果不合法
FAILED --> HUMAN_HANDOFF: 触发降级
生产化增强项
1. 模型输出强约束
真实模型调用时,我建议至少做两层约束:
- Prompt 中要求输出 JSON
- 程序侧使用 schema 校验
例如可以用 pydantic 或 JSON Schema 来约束:
from pydantic import BaseModel, Field
class TicketInfo(BaseModel):
category: str = Field(...)
priority: str = Field(...)
duration_min: int | None = None
raw_summary: str
2. 工具层做幂等和超时控制
比如调用外部 API 时:
- 设置超时
- 记录 request_id
- 失败重试带退避
- 对写操作做幂等键
3. 把日志升级成 Trace
不要只打字符串日志,最好按一步一 trace 记录:
- step_name
- input
- output
- latency_ms
- token_usage
- error_type
这样你排查时会轻松很多。
常见坑与排查
这是我在 Agent 项目里最常见、也最容易被低估的部分。
坑一:把业务流程全塞进 Prompt
现象:
- Prompt 越写越长
- 改一个规则牵一发而动全身
- 某些边界条件总失效
原因:
- 把本应由程序控制的流程逻辑交给了模型
建议:
- Prompt 只描述当前步骤目标
- 分支逻辑放到 Workflow
- 规则校验放到代码层
坑二:工具描述模糊,模型乱调用
现象:
- 本该查订单却查了库存
- 重复调用同一个工具
- 参数不完整甚至错误
原因:
- 工具说明不清晰
- 工具输入输出约束太弱
- 没有调用白名单
建议:
- 给每个工具写明确“何时调用/何时不要调用”
- 参数做 schema 校验
- 高风险工具必须显式审批
坑三:上下文越积越多,性能越来越差
现象:
- 响应变慢
- 成本上升
- 模型注意力分散
原因:
- 把整个历史对话无差别塞给模型
建议:
- 区分长期记忆与短期上下文
- 做摘要压缩
- 每一步只传必要上下文
坑四:输出是 JSON,但经常“不完全是 JSON”
现象:
- 模型前后加解释文字
- 字段缺失
- 引号不规范
原因:
- 仅靠自然语言要求,不够强
建议:
- 结合结构化输出能力
- 对解析失败做自动重试
- 失败时走兜底分支
坑五:没有置信度和降级机制
现象:
- 系统看起来总能回答
- 但偶尔一本正经地胡说
建议:
- 每一步建立置信号
- 低置信度转人工
- 高风险场景默认保守回答
一条实用排查路径
当 Agent 输出错误时,我一般按这个顺序查:
- 输入是否被清洗错误
- 分类/路由是否偏了
- 检索是否召回了错误知识
- 模型生成是否误读了工具结果
- 后处理是否截断或覆盖了结果
- 最终策略是否在高风险场景下没触发降级
别一上来就怀疑“模型太笨”。很多时候,问题出在模型之前或之后。
安全/性能最佳实践
Agent 一旦能调工具,就不再只是“文本应用”,而是“可执行系统”。这时候安全和性能必须前置考虑。
安全最佳实践
1. 明确工具权限分级
至少分三类:
- 只读工具:知识库查询、状态查询
- 低风险写工具:打标签、创建草稿
- 高风险写工具:发券、退款、删数据、执行命令
高风险工具不要让模型直接调用到底,必须有人审或有策略门禁。
2. 防 Prompt Injection
尤其是接入网页、邮件、知识库时,要默认外部文本不可信。
措施包括:
- 将“工具返回内容”和“系统指令”隔离
- 不允许外部内容覆盖系统规则
- 对工具结果做白名单提取,而不是原样拼接
3. 做敏感信息脱敏
在进入模型前处理:
- 手机号
- 身份证号
- token
- cookie
- 内部主机名
- 数据库连接串
4. 输出做策略审查
即使模型生成成功,也要在出站前检查:
- 是否泄露内部信息
- 是否包含危险指令
- 是否越权回答
- 是否触发合规风险
性能最佳实践
1. 先减少步骤,再优化模型
别一上来就追求更快的模型。很多性能问题,是因为 Workflow 步骤设计过多。
经验上,能 3 步完成就别拆成 7 步。
2. 对检索和工具结果做缓存
适合缓存的内容:
- 热门知识库查询
- 静态配置
- 模板化结果
- 工具元数据
3. 把并行的步骤并行化
例如:
- 同时查多个知识源
- 同时拉多个只读接口
- 并行跑轻量校验
4. 控制上下文大小
最直接的优化手段往往不是换模型,而是:
- 压缩历史
- 截断无关内容
- 结构化中间结果
- 避免重复传输相同信息
容量估算与取舍分析
对于 architecture 类型文章,我想再补一个常被忽略的问题:Agent 系统怎么估容量。
一个粗略估算公式
总延迟大致可以看成:
总延迟 ≈ 路由耗时 + 检索耗时 + 模型生成耗时 + 工具调用耗时 + 后处理耗时
如果一个请求包含:
- 1 次分类:400ms
- 1 次检索:150ms
- 1 次生成:800ms
- 1 次业务 API:300ms
- 后处理:50ms
那么单请求大致在:
1700ms
如果你的目标是 P95 在 2 秒内,这个设计基本可接受;如果步骤再翻倍,就危险了。
吞吐量如何看
粗略方式:
并发需求 ≈ QPS × 平均响应时间
例如:
- 目标 QPS = 20
- 平均响应时间 = 1.8 秒
则系统至少要承载:
36 个并发中的任务
如果模型 API、检索服务、业务接口三者中有一个扛不住,整体就会抖。
取舍建议
如果你更在意上线速度
选择:
- 少状态
- 少步骤
- 明确降级
- 人工兜底
如果你更在意稳定性
选择:
- 显式状态机
- 强校验
- 更多日志和 trace
- 工具权限隔离
如果你更在意成本
选择:
- 低成本模型做分类和路由
- 高成本模型只用于关键生成
- 检索优先,生成兜底
- 对热点结果做缓存
一个实用落地模板
如果你正准备在团队里推动 Agent 项目,可以按这个最小模板推进:
第一阶段:原型验证
目标:
- 证明任务可被模型理解
- 确认工具接入有价值
交付:
- 单步 Prompt
- 1~2 个工具
- 人工观察结果
第二阶段:流程化
目标:
- 将任务拆成可验证步骤
- 明确输入输出和失败路径
交付:
- Workflow 编排
- 状态定义
- 输出校验
- 降级策略
第三阶段:生产化
目标:
- 可监控、可回放、可审计
交付:
- trace
- 指标
- 告警
- 重试策略
- 权限控制
- A/B 测试
这个节奏很重要。不要一上来就做“全自动超级 Agent”,那通常既慢又不稳。
总结
从 Prompt 到 Workflow,不是“把提示词写得更复杂”,而是把 AI 能力纳入可控的软件架构。
如果只记住三句话,我希望是:
- Prompt 决定单步效果,Workflow 决定整体可靠性。
- 模型负责理解与生成,程序负责约束、执行与兜底。
- 真正能上线的 Agent,一定是有状态、有日志、有降级的。
最后给你几个可执行建议:
- 如果你现在只有一个超长 Prompt,先把它拆成 2~3 个显式步骤;
- 为每一步定义结构化输入输出,不要只传自然语言;
- 给每个工具加上权限边界、超时和参数校验;
- 把“低置信度转人工”作为默认能力,而不是失败补丁;
- 在上线前,先准备一批真实失败样本做回放测试。
边界条件也要说清楚:如果你的任务本身是高创造性、低确定性的内容生成,Workflow 不必过度复杂;但只要任务涉及查询、决策、调用外部系统、业务责任,就应该尽快从 Prompt 思维升级到 Workflow 思维。
当你完成这一步,Agent 才真正从“会聊天”变成“能交付”。