深入理解 AgentFlow:AI Agent 工作流编排的艺术
去年下半年,我开始频繁用 AI Agent 做复杂任务——先是意图识别加内容生成,后来加了知识检索、质量审核、多轮校验。每个 Agent 单独跑都没问题,但串在一起之后,状态传递、分支路由、错误恢复全成了麻烦事。这篇文章讲讲我为什么从手写胶水代码切换到 AgentFlow,以及在实际项目中如何使用它构建可靠的多 Agent 协作系统。
去年下半年,我开始频繁用 AI Agent 做复杂任务。
一开始很简单——用户提一个问题,Agent 生成回答,结束。后来需求变多了:要先识别意图,再查知识库,再生成回复,再让另一个 Agent 审核一遍,最后决定是直接发还是转人工。
每个 Agent 单独跑都没问题。但把它们串在一起之后,麻烦就来了。
状态怎么在 Agent 之间传递?错误了怎么恢复?某个分支走不通怎么降级?这些问题让我意识到:单 Agent 是模型能力的问题,多 Agent 协作是工程能力的问题。
后来我用了 AgentFlow,这些问题才有了比较系统的解决方案。
从手写胶水代码说起
最开始的做法很直接——每个 Agent 的输入输出手写拼接。
# 伪代码示意
intent = intent_agent.run(user_input)
if intent == "question":
knowledge = search_agent.run(user_input)
response = generate_agent.run(knowledge)
elif intent == "complaint":
response = escalate_agent.run(user_input)
# ...三个 Agent 以内还能应付。到了六个 Agent 的时候,代码变成了一个巨大的 if-else 嵌套,加了新分支忘了改老分支,状态字段越来越多,我自己都搞不清某个变量是在哪一步被赋值的。
更麻烦的是错误处理。有一次知识检索超时了,后面的生成 Agent 收到了空数据,直接生成了一段“我无法回答”的回复,用户收到之后投诉了。日志里看不出问题在哪,因为错误发生在 Agent 内部,外层代码没有捕获到。
这时候我才认真开始找编排框架。
AgentFlow 解决了什么问题
AgentFlow 的核心设计围绕四个概念展开,刚好对应我之前手写时遇到的麻烦。
Node 是工作流里的一个执行单元,通常封装一个 Agent 或一个函数。每个 Node 只干一件事,输入输出明确。Edge 定义 Node 之间的执行顺序,相当于把之前散落在代码各处的调用关系显式地画了出来。Conditional Edge 根据当前状态决定下一步走向——这个解决了我的 if-else 噩梦。State 是贯穿整个工作流的共享上下文,所有 Agent 读写同一个 State 对象,不用再手动传参。
这个设计让我感觉挺舒服的——它没有发明什么新概念,就是把分布式系统里那套 DAG 编排的思路搬到了 Agent 场景,但搬得恰到好处。
一个真实案例:智能客服工作流
我们最近用 AgentFlow 重构了一个客服系统,这个案例比较典型,可以看看完整的实现。
interface CustomerServiceState extends State {
userQuery: string;
intent: 'question' | 'complaint' | 'refund' | 'unknown';
context: string;
response: string;
needHuman: boolean;
}
const intentAgent = new Agent({
name: 'intent-agent',
model: 'gpt-4',
systemPrompt: `将用户输入分类为: question / complaint / refund / unknown`,
});
const knowledgeAgent = new Agent({
name: 'knowledge-agent',
model: 'gpt-4',
tools: [vectorSearchTool],
});
const responseAgent = new Agent({
name: 'response-agent',
model: 'gpt-4',
systemPrompt: '基于检索到的知识生成客服回复',
});
const escalateAgent = new Agent({
name: 'escalate-agent',
model: 'gpt-4',
systemPrompt: '生成转人工话术并创建工单',
});
const routeByIntent = (state: CustomerServiceState): string => {
if (state.intent === 'unknown') return 'escalate';
if (state.intent === 'complaint') return 'escalate';
return 'knowledge';
};
const flow = new AgentFlow<CustomerServiceState>({
nodes: {
intent: intentAgent.asNode(),
knowledge: knowledgeAgent.asNode(),
response: responseAgent.asNode(),
escalate: escalateAgent.asNode(),
},
edges: [
{ from: 'START', to: 'intent' },
ConditionalEdge.from('intent', routeByIntent, {
knowledge: 'knowledge',
escalate: 'escalate',
}),
{ from: 'knowledge', to: 'response' },
{ from: 'response', to: 'END' },
{ from: 'escalate', to: 'END' },
],
});这段代码清晰了很多。意图识别、知识检索、回复生成、转人工——四个 Node 各司其职,路由逻辑集中在 routeByIntent 里,新增一个意图只需要改这个函数,不用在业务代码里到处散落条件判断。
执行路径对用户来说是这样的:
两个让我省力的高级特性
除了基础编排,AgentFlow 有两个高级功能在实际使用中帮了大忙。
第一个是并行执行。
有一次我们需要同时做摘要、翻译和情感分析,三个任务互不依赖。以前我是串行写的——等摘要完成再翻译,翻译完再做情感分析,每次调用一次模型,总耗时是三者之和。后来改成并行,总耗时降到了最慢那个任务的时间,快了两倍多。
const flow = new AgentFlow({
nodes: {
analyze: analyzeAgent.asNode(),
summarize: summarizeAgent.asNode(),
translate: translateAgent.asNode(),
merge: mergeAgent.asNode(),
},
edges: [
{ from: 'START', to: 'analyze' },
{ from: 'analyze', to: 'summarize' },
{ from: 'analyze', to: 'translate' },
{ from: 'summarize', to: 'merge' },
{ from: 'translate', to: 'merge' },
{ from: 'merge', to: 'END' },
],
});第二个是子流程复用。
内容审核这套逻辑在很多地方都要用——用户提交评论要审、AI 生成的回复要审、第三方内容接入也要审。我把审核逻辑封装成一个 ContentReviewSubflow,在主流程里当成一个普通 Node 调用。
const contentReviewSubflow = new SubFlow({
nodes: {
check: checkAgent.asNode(),
filter: filterAgent.asNode(),
},
// ...
});
const mainFlow = new AgentFlow({
nodes: {
generate: generateAgent.asNode(),
review: contentReviewSubflow.asNode(),
publish: publishAgent.asNode(),
},
// ...
});这份复用带来一个额外的好处:审核规则的修改只需要改一个地方,所有使用它的流程自动生效。在以前手写胶水的年代,改一个规则要在三四个地方同步修改,漏一个就是线上事故。
关于可观测性的一个细节
AgentFlow 内置了事件监听机制,可以用它把执行过程串到日志系统里。
flow.on('node:start', (event) => {
// 记录开始
});
flow.on('node:complete', (event) => {
// 记录耗时,方便找出瓶颈节点
});
flow.on('error', (event) => {
// 错误上报,不需要等到用户反馈才知道出问题了
});我在生产环境加了这几行之后,排查问题的效率高了很多。以前要大海捞针地翻日志,现在打开监控面板就能看到每个节点的耗时和成功率。哪个 Agent 经常超时、哪个节点错误率高,一目了然。
一点关于选型的感受
市面上能做 Agent 编排的方案不少,我简单对比了一下:
| 特性 | AgentFlow | LangGraph | Dify | AutoGen |
|---|---|---|---|---|
| 可视化编排 | ✅ | ✅ | ✅ | ❌ |
| 条件路由 | ✅ | ✅ | ✅ | ✅ |
| 并行执行 | ✅ | ✅ | ❌ | ✅ |
| 子流程复用 | ✅ | ❌ | ✅ | ❌ |
| 错误重试 | ✅ | ❌ | ✅ | ❌ |
| TypeScript 原生 | ✅ | ❌ | ❌ | ❌ |
| 学习曲线 | 低 | 中 | 低 | 高 |
AgentFlow 不是功能最多的,但它的 API 设计让我觉得“刚刚好”。上手快,文档清晰,类型安全做得比较到位。对比下来,它更适合想快速落地、不想在编排框架本身上花太多时间的方向。
几个从踩坑里来的建议
实际用下来,有几个建议想分享:
状态设计要精简。 不要把大对象整个塞进 State,只放必要的字段。State 会在节点之间传递,数据量太大会影响性能,也容易造成字段混乱。
节点职责单一。 一个 Node 只做一件事。我一开始把“生成回复”和“审核回复”放在同一个 Node 里,后来想加条件分支时才发现拆开更灵活。
加超时控制。 模型调用可能卡住,要给每个 Node 设置合理的超时时间。默认的超时一般是 60 秒,但对简单的分类任务来说,5 秒没返回就应该走降级了。
错误兜底要提前想好。 LLM 调用可能失败,要有降级方案。比如知识检索超时,就直接转人工,而不是把空数据传给下一个 Agent 让它生成一个莫名其妙的回复。
写在最后
多 Agent 协作这件事,本质上是把“一个很聪明但不太靠谱的模型”用工程手段变得可靠。AgentFlow 做的是把编排这部分标准化了——让开发者不用每次从头搭积木,把精力放在业务逻辑本身。
我现在的习惯是:只要涉及两个以上的 Agent 协作,就先画一下 DAG,再在 AgentFlow 里定义出来。看起来多了一个步骤,实际上省了大量调试时间。
代码逻辑画出来之后,自己容易看清楚哪里有坑,团队也容易对齐。这就是编排框架最大的价值——让复杂的协作过程,变得可设计、可执行、可追溯。