分享一个把多智能体工作流从冗余代码中解耦的方案

阿伟 中级 5小时前 83 浏览 15 点赞 约 1 分钟

现在的 AI Agent 框架有个很恶心的通病:要么是像 n8n 那种低代码工具,逻辑稍微复杂点(比如带条件的循环或状态削减)就直接卡死;要么就是像 LangGraph 这种纯代码框架,灵活是灵活,但写个简单的三智能体分发流程,得先写一堆 TypedDict 定义状态,再一个个初始化节点,最后还得配置 checkpointer。这种大量的样板代码(Boilerplate)严重干扰了对业务逻辑本身的思考。

我觉得理想的状态应该是:用一种中立的、可移植的描述语言来定义工作流,而不是把逻辑死死地绑在某个特定框架的 Python 代码里。

这就是我尝试解决的问题,我搞了一个叫 OpenAgentFlow 的东西,核心逻辑是定义一种 .oaf 规范。简单来说,它就像是 REST API 里的 OpenAPI 协议,只负责描述状态机和智能体之间的通信流,而不关心具体的执行环境。

看个实操例子,下面这个客户支持分单的工作流,用 .oaf 写只需要 40 行左右,如果用常规 Python 框架写,没个 200 行代码根本搞不定:

workflow "Support Ticket Triage" {

 config {
 max_iterations: 3
 timeout_seconds: 45
 }

 state {
 ticket_text: string
 user_tier: string
 category: string
 urgency: string
 suggested_reply: string
 }

 agent Classifier {
 instructions: """
 Analyze the incoming support ticket.
 Categorize into: 'Billing', 'Technical Bug', or 'Feature Request'.
 Assess urgency level: 'Low', 'Medium', or 'Critical'.
 """
 model: "gemini-2.0-flash"
 temperature: 0.1
 inputs: [ticket_text, user_tier]
 outputs: [category, urgency]
 }

 agent Responder {
 instructions: """
 Draft a helpful response based on category and urgency.
 If urgency is 'Critical', note that high-priority alert has been dispatched.
 """
 model: "gemini-2.0-flash"
 temperature: 0.6
 inputs: [ticket_text, user_tier, category, urgency]
 outputs: [suggested_reply]
 }

 flow {
 start -> Classifier 
 Classifier -> Responder 
 Responder -> end
 }
}

这种定义方式最大的好处是把「逻辑设计」和「代码实现」分开了。你只需要关注 Agent 的输入输出和流转顺序,而不需要在代码里纠结怎么路由节点。对于需要快速迭代工作流的实战场景,这种解耦能省掉大量重复的部署和调试时间。

LangChainAI编程AIAI编程实战python

全部回复 (4)

数据分析师大山 中级 9小时前
确实,我之前试过把状态抽成单例,省得到处传字典,清爽多了。
0 回复
运营喵小柯 中级 9小时前
那要是多节点并发的时候,怎么保证状态同步不乱?
0 回复
阿Leo的日常 中级 9小时前
这种方案大概率得靠分布式锁吧?不然并发高了绝对得崩。
0 回复
夜猫子创业者 专家 9小时前
记得把版本控制加上,不然改个节点逻辑全线崩掉,得回滚半天。
0 回复

发表回复

支持 Markdown 格式