为什么需要流程编排？
第一个 Agent 写出来很简单——一个 LLM 加上几个工具，就能对话、查天气、算数学题。但真实业务场景从来不是"一问一答"那么简单：
一个工单进来 → 需要先识别意图 → 如果是技术问题需要查知识库 → 知识库找不到要 escalation 到人工 → 如果是投诉要判断紧急程度 → 紧急投诉要自动拉群 → 处理完后要生成总结报告 → 归档。
这不是一个 Agent + 一轮对话能解决的。这是多个决策节点、多种工具、分支路由、状态持久化的组合，这就是流程编排要解决的问题。
基础概念
什么是 Agent 流程编排
Agent 流程编排（Orchestration）是将 LLM 的能力以有向图的方式组织起来，通过节点（Node）表示执行单元，边（Edge）表示流转规则，让 Agent 在预定义的结构中自主决策和执行。
传统 Agent： 流程编排 Agent：
用户输入 用户输入
│ │
LLM + Tools ├──→ 意图识别（LLM 节点）
│ │
输出 ├──→ 知识检索（Tool 节点）
│ │
├──→ 是否找到？
│ ├─ YES → 生成回答（LLM 节点）
│ └─ NO → 人工升级（Human 节点）
│
└──→ 归档记录（Tool 节点）流程编排 = 用图的方式定义 Agent 的"思考骨架"，LLM 在骨架内做智能填充。
核心组成
| 概念 | 说明 | 类比 |
|---|---|---|
| State（状态） | 在节点之间传递的共享数据，通常是一个 TypedDict | 流水线上的传送带 |
| Node（节点） | 执行逻辑的最小单元，可以是 LLM 调用、工具调用、人工干预、条件判断 | 生产线上的工位 |
| Edge（边） | 节点之间的流转关系，普通边直接传递，条件边根据 State 决定走向 | 传送带路径 |
| Graph（图） | 节点 + 边的完整定义，描述整个业务流程 | 完整的工厂布局 |
| Checkpoint（检查点） | 每一步的状态快照，支持暂停、恢复、回溯 | 存档点 |
三大主流框架对比
| 框架 | 定位 | 核心特点 | 适合场景 |
|---|---|---|---|
| LangGraph | 底层图编排引擎 | 状态图、条件分支、人机协作、checkpoint | 需要精细控制的复杂流程 |
| CrewAI | 多 Agent 协作层 | 角色分工、任务委派、顺序/层级结构 | 多个 Agent 角色协作 |
| Dify / Coze | 低代码平台 | 拖拽式 UI、内置生态、快速原型 | 非开发者或快速验证 |
本文重点讲 LangGraph，因为它既能让你理解编排的底层逻辑，又能在生产环境中直接使用。
工具使用：从简单调用到深度理解
工具（Tool）是 Agent 的手和脚。没有工具，LLM 只是一个聊天机器；有了工具，它能操作数据库、发送邮件、调用 API、读写文件。
工具的完整生命周期
定义 → 注册 → 选择 → 调用 → 结果处理 → 重试/纠错1. 定义工具
LangChain 提供了多种定义工具的方式：
python
from langchain_core.tools import tool
# 方式一：装饰器（最简单）
@tool
def search_knowledge_base(query: str) -> str:
"""在内部知识库中搜索技术文档。
Args:
query: 搜索关键词，建议使用技术术语
"""
results = kb.search(query, top_k=5)
return "\n---\n".join(
f"[{r['score']:.2f}] {r['title']}\n{r['content'][:300]}"
for r in results
)
# 方式二：继承 BaseTool（复杂逻辑）
from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field
class CreateJiraTicketInput(BaseModel):
title: str = Field(description="工单标题")
description: str = Field(description="问题详细描述")
priority: str = Field(description="优先级: P0/P1/P2/P3")
assignee: str = Field(description="指派人用户名")
class CreateJiraTicket(BaseTool):
name: str = "create_jira_ticket"
description: str = "在 Jira 中创建工单。当需要正式记录问题或提交给其他团队处理时使用"
args_schema: type[BaseModel] = CreateJiraTicketInput
def _run(self, title: str, description: str, priority: str, assignee: str) -> str:
resp = jira_client.create_issue(
project="SUPPORT",
summary=title,
description=description,
priority=priority,
assignee=assignee
)
return f"工单已创建: {resp.key} ({resp.url})"深度分析 — 为什么 Tool 的 description 比代码还重要？
LLM 选择工具的依据不是你的代码实现，而是
name和description。如果把 description 写成"搜索东西"，LLM 永远不知道该什么时候用；写成"在内部知识库中搜索技术文档，适用于用户提出技术问题需要查找解决方案的场景"，LLM 的选择准确率会大幅提升。实践建议：description 要包含三个信息——这个工具做什么、输入参数含义、什么场景下应该使用它。这是调优 Agent 行为的最低成本手段。
2. 注册工具
python
tools = [
search_knowledge_base,
CreateJiraTicket(),
escalate_to_human,
send_notification,
log_to_db,
]
# 绑定到 LLM
llm_with_tools = llm.bind_tools(tools)bind_tools 做了什么？它将每个工具的 Schema（name + description + 参数 JSON Schema）注入到 LLM 的系统提示中，LLM 可以从这些 Schema 中选择调用。
3. 工具调用机制
当 LLM 决定调用工具时，它不执行代码，而是返回一个 ToolCall 消息：
python
# LLM 返回的不是执行结果，而是一个调用指令
AIMessage(
content="",
tool_calls=[{
"name": "search_knowledge_base",
"args": {"query": "登录 502 错误 nginx"},
"id": "call_abc123"
}]
)流程编排的职责是：接收到 ToolCall → 解析 name/args → 找到对应工具函数 → 执行 → 将结果包装成 ToolMessage → 送回 LLM。
python
from langgraph.prebuilt import ToolNode
# LangGraph 提供的 ToolNode 封装了这个完整流程：
tool_node = ToolNode(tools) # 自动完成：解析→执行→包装→回传深度分析 — 工具并发调用
如果 LLM 一次返回了多个 ToolCall，你的编排框架应该并发执行它们，而不是串行等待。假设用户问"对比 A 和 B 两个方案的处理进度"，LLM 可能同时调用
query_ticket("A")和query_ticket("B")，并发执行节省的时间随工具数量线性增长。
业务实战：智能工单处理系统
下面我们从头构建一个完整的智能工单处理系统。这是一个能直接落地的案例——从接到用户工单到自动处理、人工升级、归档的全链路。
业务背景
某 SaaS 公司的客服团队每天接到 200+ 工单，类型覆盖：技术咨询（40%）、Bug 报告（25%）、投诉（15%）、功能建议（10%）、其他（10%）。当前流程全靠人工判断、手动分配，效率低下且容易遗漏紧急工单。
目标：用 Agent 自动化工单的前半程处理——意图识别、简单问题自动回复、复杂问题智能路由、紧急问题即时升级。
流程图
┌──────────────────────┐
│ 工单输入 │
│ (用户文本 + 元信息) │
└──────────┬───────────┘
│
┌──────────▼───────────┐
│ 意图识别 & 分类 │ ← LLM 节点
│ (技术/Bug/投诉/建议) │
└──────────┬───────────┘
│
┌─────────▼──────────┐
│ 是否技术类？ │ ← 条件路由
└──┬────────────┬────┘
YES │ │ NO
┌──────────────▼──┐ ┌───▼──────────────┐
│ 知识库检索 │ │ 按分类直接处理 │
│ (RAG Tool) │ │ 生成回复/流转 │
└──────┬──────────┘ └──────────────────┘
│
┌──────▼──────────┐
│ 知识库命中？ │ ← 条件路由
└──┬──────────┬───┘
YES │ │ NO
┌────────────▼──┐ ┌───▼─────────────┐
│ 生成技术回复 │ │ 判断是否紧急 │
│ (LLM 节点) │ │ (LLM 判断) │
└──────┬────────┘ └──┬──────────┬───┘
│ 紧急 │ │ 一般
│ ┌─────────▼──┐ ┌───▼───────────┐
│ │ 即时告警 │ │ 创建 Jira 工单 │
│ │ + 拉群(可选)│ │ 指派对应团队 │
│ └─────────┬──┘ └───┬───────────┘
│ │ │
└───────────────┴────┬────┘
│
┌──────────▼───────────┐
│ 归档 & 生成汇总报告 │ ← Tool 节点
│ (存 DB + 发送通知) │
└──────────────────────┘第一步：定义状态
State 是整个流程的"共享内存"，每个节点读取它、修改它、传递它。
python
from typing import TypedDict, Literal, Annotated, Optional
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
class TicketState(TypedDict):
# 用户输入的原始消息
messages: Annotated[list[BaseMessage], add_messages]
# --- 以下是业务状态字段 ---
# 用户信息
user_id: str
user_name: str
user_plan: str # 用户套餐: free/pro/enterprise
# 工单分类结果
intent: Optional[str] # technical/bug/complaint/feature_request/other
priority: Optional[str] # P0/P1/P2/P3
sentiment: Optional[str] # 用户情绪: negative/neutral/urgent
# 知识库检索结果
kb_results: Optional[str]
kb_score: Optional[float]
# 处理结果
resolution: Optional[str] # 最终解决方案
jira_ticket_key: Optional[str] # Jira 工单号
needs_escalation: bool
escalation_reason: Optional[str]
# 流转控制
next_step: Optional[str] # 显式控制下一步走向第二步：定义工具
python
from langchain_core.tools import tool
import json, datetime
@tool
def search_kb(query: str, top_k: int = 5) -> str:
"""搜索内部知识库。
当你需要查找技术文档、已知问题、解决方案时使用。
适合场景：用户提出技术问题、报错信息、操作疑问。
Args:
query: 搜索关键词，尽量精炼
top_k: 返回结果数量
"""
# 实际项目中接入向量数据库（Milvus/Qdrant/Weaviate）
# 这里以模拟数据演示
results = kb_database.similarity_search(query, k=top_k)
if not results:
return "未找到相关知识。"
lines = []
for i, doc in enumerate(results):
lines.append(
f"### 结果 {i+1} (相关度: {doc.score:.2f})\n"
f"**标题**: {doc.metadata['title']}\n"
f"**内容摘要**: {doc.page_content[:400]}\n"
)
return "\n---\n".join(lines)
@tool
def create_jira_ticket(title: str, description: str, priority: str, assignee: str = "support-team") -> str:
"""在 Jira 中创建工单。
当 AI 无法直接解决用户问题、需要人工介入时使用。
Args:
title: 工单标题，要包含问题类型和关键词
description: 完整的问题描述，包含用户原始问题、已做的排查步骤
priority: P0（系统宕机）/ P1（核心功能不可用）/ P2（一般）/ P3（小问题）
assignee: 默认指派到 support-team
"""
# 实际项目中调用 Jira API
ticket_id = f"SUPPORT-{datetime.datetime.now().strftime('%Y%m%d')}-{hash(title) % 10000:04d}"
return f"Jira 工单已创建: {ticket_id} | 优先级: {priority} | 指派人: {assignee}"
@tool
def escalate_to_oncall(user_id: str, reason: str, priority: str) -> str:
"""紧急升级到值班人员。
当检测到 P0/P1 级别故障或用户情绪异常激动时使用。
会通过企业微信 / 钉钉 / Slack 发送即时告警。
Args:
user_id: 受影响用户ID
reason: 升级原因描述
priority: 紧急程度
"""
# 实际项目中调用企业IM告警API
return (
f"🚨 紧急升级\n"
f"- 用户: {user_id}\n"
f"- 优先级: {priority}\n"
f"- 原因: {reason}\n"
f"- 状态: 已通知值班人员，预计5分钟内响应"
)
@tool
def archive_to_db(
user_id: str, intent: str, priority: str,
resolution: str, jira_key: str = ""
) -> str:
"""将工单处理记录归档到数据库。
每个工单处理完成后必须调用此工具保存记录。
"""
record = {
"user_id": user_id,
"intent": intent,
"priority": priority,
"resolution": resolution,
"jira_key": jira_key,
"timestamp": datetime.datetime.now().isoformat(),
}
# db.tickets.insert_one(record)
return f"工单已归档 (ID: {hash(json.dumps(record))})"
tools = [search_kb, create_jira_ticket, escalate_to_oncall, archive_to_db]第三步：定义节点函数
每个节点是一个独立的函数，输入和输出都是 State。
python
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage
from langgraph.graph import StateGraph, START, END
llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tools = llm.bind_tools(tools)
# ---- 节点 1：意图识别与分类 ----
CLASSIFY_PROMPT = """你是智能工单系统的分类器。根据用户消息，分析以下维度并返回 JSON：
1. intent: technical / bug / complaint / feature_request / other
2. priority: P0（系统宕机，多人受影响）/ P1（核心功能不可用）/ P2（使用有阻碍）/ P3（体验问题）
3. sentiment: urgent / negative / neutral
注意：
- 如果用户提到"不能用""崩了""报错无法操作"，通常是 P1 或 P0
- 如果用户态度愤怒、用了大量感叹号或威胁性语言，sentiment 为 urgent
- 只返回 JSON，不要多余解释"""
import json
def classify_intent(state: TicketState) -> TicketState:
"""意图识别节点：分析用户意图、优先级、情绪"""
last_msg = state["messages"][-1]
response = llm.invoke([
SystemMessage(content=CLASSIFY_PROMPT),
HumanMessage(content=last_msg.content)
])
# 解析 LLM 返回的分类结果
try:
result = json.loads(response.content.strip().removeprefix("```json").removesuffix("```"))
except json.JSONDecodeError:
result = {"intent": "other", "priority": "P2", "sentiment": "neutral"}
return {
"intent": result.get("intent", "other"),
"priority": result.get("priority", "P2"),
"sentiment": result.get("sentiment", "neutral"),
"messages": [AIMessage(content=f"[分类结果] 意图:{result.get('intent')} | 优先级:{result.get('priority')} | 情绪:{result.get('sentiment')}")]
}
# ---- 节点 2：知识库检索 ----
def search_knowledge(state: TicketState) -> TicketState:
"""知识检索节点：调用 LLM 自主决定搜索策略"""
search_prompt = """基于用户的原始问题，生成一个简练的搜索查询。
只返回搜索关键词，不要多余内容。"""
last_user_msg = next(m for m in state["messages"] if isinstance(m, HumanMessage))
search_query = llm.invoke([
SystemMessage(content=search_prompt),
HumanMessage(content=last_user_msg.content)
]).content.strip()
# 调用工具
result = search_kb.invoke({"query": search_query})
# 提取最高相关度
kb_score = 0.0
if "相关度:" in result:
import re
scores = re.findall(r"相关度:\s*([\d.]+)", result)
kb_score = max(float(s) for s in scores) if scores else 0.0
return {
"kb_results": result,
"kb_score": kb_score,
"messages": [AIMessage(content=f"[知识库检索] 查询词: {search_query} | 最高相关度: {kb_score:.2f}")]
}
# ---- 节点 3：生成技术回复 ----
def generate_tech_reply(state: TicketState) -> TicketState:
"""基于知识库结果生成技术回复"""
reply_prompt = f"""你是资深技术支持。基于知识库检索结果，给用户一个清晰、可操作的回复。
要求：
1. 如果知识库有明确方案，给出步骤化指引
2. 如果只有部分相关，提供排查思路
3. 语气专业但友好
4. 如果涉及付费功能，注明对企业版/专业版用户适用
知识库检索结果：
{state["kb_results"]}
用户问题：
{next(m for m in state["messages"] if isinstance(m, HumanMessage)).content}"""
reply = llm.invoke([SystemMessage(content=reply_prompt)])
return {
"resolution": reply.content,
"messages": [reply]
}
# ---- 节点 4：紧急处理 ----
def handle_urgent(state: TicketState) -> TicketState:
"""紧急工单处理：告警 + 升级"""
alert_msg = f"P{state['priority']}紧急工单 | 用户: {state['user_name']}({state['user_id']}) | {state.get('escalation_reason', '')}"
escalate_to_oncall.invoke({
"user_id": state["user_id"],
"reason": alert_msg,
"priority": state["priority"]
})
return {
"needs_escalation": True,
"escalation_reason": alert_msg,
"resolution": f"已紧急升级，值班人员将尽快联系您。参考工单号将在创建后通知。"
}
# ---- 节点 5：创建工单（非技术类或知识库未命中） ----
def route_to_human(state: TicketState) -> TicketState:
"""创建 Jira 工单，指派人工处理"""
last_user_msg = next(m for m in state["messages"] if isinstance(m, HumanMessage))
description = (
f"用户: {state['user_name']} ({state['user_id']})\n"
f"套餐: {state['user_plan']}\n"
f"分类: {state['intent']}\n"
f"原始问题:\n{last_user_msg.content}\n\n"
f"AI 已完成的排查:\n{state.get('kb_results', '未进行知识库检索')}"
)
result = create_jira_ticket.invoke({
"title": f"[{state['intent']}] {last_user_msg.content[:50]}",
"description": description,
"priority": state["priority"],
"assignee": "support-team"
})
return {
"jira_ticket_key": "SUPPORT-" + result.split("SUPPORT-")[1].split(" ")[0] if "SUPPORT-" in result else "",
"resolution": f"您的问题已转交专业团队处理。{result}",
"messages": [AIMessage(content=result)]
}
# ---- 节点 6：归档 ----
def archive(state: TicketState) -> TicketState:
"""归档工单处理记录"""
archive_to_db.invoke({
"user_id": state["user_id"],
"intent": state["intent"],
"priority": state["priority"],
"resolution": state["resolution"],
"jira_key": state.get("jira_ticket_key", "")
})
summary = (
f"### 工单处理完成\n"
f"- 用户: {state['user_name']} ({state['user_id']})\n"
f"- 意图: {state['intent']} | 优先级: {state['priority']}\n"
f"- 是否需要升级: {'是' if state.get('needs_escalation') else '否'}\n"
f"- Jira: {state.get('jira_ticket_key', 'N/A')}\n"
)
return {
"messages": [AIMessage(content=summary)]
}第四步：构建路由逻辑
路由函数是流程编排的核心——它根据当前 State 决定下一步走向。
python
def route_after_classify(state: TicketState) -> str:
"""分类后的路由：技术类问题先去知识库检索"""
if state["intent"] in ("technical", "bug"):
return "search_kb"
return "route_to_human"
def route_after_search(state: TicketState) -> str:
"""检索后的路由：命中→生成回复；未命中→判断紧急度"""
SCORE_THRESHOLD = 0.7 # 低于此分数认为未命中
if state["kb_score"] >= SCORE_THRESHOLD:
return "generate_reply"
# 未命中，判断是否需要紧急处理
if state["priority"] in ("P0", "P1") or state["sentiment"] == "urgent":
return "handle_urgent"
return "route_to_human"
def route_after_reply(state: TicketState) -> str:
"""生成回复后直接归档"""
return "archive"
def route_after_urgent(state: TicketState) -> str:
"""紧急处理后：先归档（同时在值班系统已创建告警）"""
return "archive"深度分析 — 路由的三种模式
- 确定性路由：根据 State 字段做 if/else 判断。如上面的
route_after_classify。- LLM 路由：让 LLM 读 State 中的消息和上下文，动态决定。适合边界模糊的场景。
- 混合路由：先确定性判断大类，再用 LLM 做细粒度选择。
实践建议：能用确定性路由的尽量不用 LLM 路由。每多一次 LLM 调用就多几百毫秒延迟 + token 成本。只在 State 中的信息不足以确定性判断时才引入 LLM。
第五步：组装 Graph
python
from langgraph.graph import StateGraph, START, END
# 创建状态图
workflow = StateGraph(TicketState)
# 添加所有节点
workflow.add_node("classify", classify_intent)
workflow.add_node("search_kb", search_knowledge)
workflow.add_node("generate_reply", generate_tech_reply)
workflow.add_node("handle_urgent", handle_urgent)
workflow.add_node("route_to_human", route_to_human)
workflow.add_node("archive", archive)
# 添加边
workflow.add_edge(START, "classify")
# 分类后的条件分支
workflow.add_conditional_edges(
"classify",
route_after_classify,
{
"search_kb": "search_kb",
"route_to_human": "route_to_human",
}
)
# 检索后的条件分支
workflow.add_conditional_edges(
"search_kb",
route_after_search,
{
"generate_reply": "generate_reply",
"handle_urgent": "handle_urgent",
"route_to_human": "route_to_human",
}
)
# 各终点都汇聚到归档
workflow.add_edge("generate_reply", "archive")
workflow.add_edge("handle_urgent", "archive")
workflow.add_edge("route_to_human", "archive")
# 归档后结束
workflow.add_edge("archive", END)
# 编译
app = workflow.compile()第六步：运行测试
python
# 测试用例 1：技术问题 - 知识库能解决
result1 = app.invoke({
"messages": [HumanMessage(content="登录页面一直转圈无法登录，尝试过清除缓存也没用")],
"user_id": "u_1001",
"user_name": "张三",
"user_plan": "pro",
"intent": None,
"priority": None,
"sentiment": None,
"kb_results": None,
"kb_score": None,
"resolution": None,
"jira_ticket_key": None,
"needs_escalation": False,
"escalation_reason": None,
"next_step": None,
})
print(result1["resolution"])
# 预期：返回基于知识库的步骤化解决方案
# 测试用例 2：紧急故障 - 需要告警升级
result2 = app.invoke({
"messages": [HumanMessage(content="系统完全崩溃了！！所有用户都无法访问！！已经半小时了！！")],
"user_id": "u_2001",
"user_name": "李四",
"user_plan": "enterprise",
"intent": None,
"priority": None,
"sentiment": None,
"kb_results": None,
"kb_score": None,
"resolution": None,
"jira_ticket_key": None,
"needs_escalation": False,
"escalation_reason": None,
"next_step": None,
})
print(f"升级状态: {result2['needs_escalation']}")
print(f"优先级: {result2['priority']}")
# 预期：P0/P1，已触发告警
# 测试用例 3：非技术问题
result3 = app.invoke({
"messages": [HumanMessage(content="能不能增加一个批量导出Excel的功能？我们现在要一周导出一次")],
"user_id": "u_3001",
"user_name": "王五",
"user_plan": "pro",
"intent": None,
"priority": None,
"sentiment": None,
"kb_results": None,
"kb_score": None,
"resolution": None,
"jira_ticket_key": None,
"needs_escalation": False,
"escalation_reason": None,
"next_step": None,
})
print(f"Jira 工单: {result3['jira_ticket_key']}")
# 预期：识别为 feature_request，创建 Jira 工单生产落地 Checklist
上面是核心流程，但要真正上线，还需要补齐以下环节：
| 环节 | 需要做的事 | 注意事项 |
|---|---|---|
| 日志与监控 | 每个节点记录耗时、输入输出摘要 | LangSmith / LangFuse 自动追踪 |
| 错误处理 | 节点级的 try/catch + 重试策略 | 别让一个工具异常导致整个流程挂掉 |
| Human-in-the-loop | 敏感操作（退款、权限变更）需要人工确认 | LangGraph 的 interrupt 机制 |
| Checkpoint | 支持暂停/恢复，排查问题可以回放 | 默认存内存，生产建议存 Redis/Postgres |
| 速率限制 | LLM API 调用加限流，避免成本失控 | 使用 token bucket 或 LangChain callback |
| AB 测试 | 灰度发布，先跑 10% 流量观察效果 | 建议和人工处理并行运行一段时间对比 |
| 反馈闭环 | 收集用户满意度，自动标记需改进的 case | 将低评分工单纳入知识库优化 |
Human-in-the-loop 示例
python
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt
checkpointer = MemorySaver()
# 在需要人工确认的节点中：
def safety_check(state: TicketState) -> TicketState:
if state["intent"] == "refund" and state["priority"] in ("P0", "P1"):
# 暂停流程，等待人工确认
response = interrupt({
"question": f"用户 {state['user_id']} 请求紧急退款，是否批准？",
"context": state
})
if not response.get("approved"):
return {"next_step": "reject_refund"}
return state总结
Agent 流程编排的本质是：把 LLM 的不确定性关进结构化的牢笼里。
- LLM 负责"模糊判断"（意图识别、情感分析、文本生成）
- 图结构负责"确定逻辑"（分支条件、流转顺序、异常兜底）
- 工具负责"执行落地"（查库、发消息、创工单）
三个层次各司其职，才有了能真正用于生产的 AI Agent。
不要让 LLM 做它能做的所有事，让它只做它擅长的事。剩下的交给代码。