把大模型变成一支团队：生产级 Agent 工作流框架与落地 Demo 全解析

内容分享4小时前发布

0 0 0

一、为什么生产环境需要「Workflow + Agentic AI」

许多人一开始玩大模型，就是一条 llm(prompt)，最多再加个 RAG。
但一旦落到生产环境，你会发现几个痛点：

业务流程是多步骤的
先解析需求 → 再检索资料 → 再调用内部 API → 再写总结 → 再让风控审核
单次 LLM 调用搞不定，需要一个「有状态的多阶段流程」
需要多个“角色”协作
研究员 Agent、写手 Agent、代码工具 Agent、审计 Agent……
每个 Agent 有不同指令、工具权限、记忆和上下文
要能观测、回放和调试
出了事故要知道：哪一步 prompt 崩了？
哪个 Agent 调了危险工具？哪个节点超时了？
要可维护、可演进
流程需要版本化：v1 / v2 / AB test
模型可以换，策略可以调，流程结构最好别动

于是就催生了一类专门的 Workflow / Agentic 框架：

把「LLM + 工具 + Agent + 多步骤流程」统一建模、可视化、可回放、可监控。

二、生产环境常用的 Workflow / Agentic 框架盘点

1. LangGraph：LangChain 团队的「有状态工作流」方案

定位：

一个 低层级编排框架，用有向图（StateGraph）来描述复杂、多步、状态机式的 Agent 工作流。

特点：

用 节点（node）+ 边（edge）+ 状态（state） 来描述流程
天然支持 多 Agent、循环、分支、人类介入（human-in-the-loop）
内建 持久化（persistence），可以恢复长对话、长任务
和 LangChain 生态深度整合（工具、RAG、模型适配）
易于接入 observability（列如 Langfuse）做 tracing 和调试Langfuse

适合场景：

已经在用 LangChain，希望对「复杂多步骤、多 Agent 流程」有精细控制的团队。

2. Agno + AgentOS：框架 + 运行时的一体化方案

定位：

多 Agent 框架 + 高性能运行时（AgentOS），主打「在自己云里跑安全高效的多智能体系统」。agno-v2.mintlify.app+2GitHub+2

特点：

提供 Agent / Teams / Workflows 的完整模型
自带一个 FastAPI 的 AgentOS 应用，几行代码就能起一个有 UI 的后端
有控制平面：测试、监控、管理 Agent，适合企业生产落地
官方示例许多，从股票分析、数据分析到业务 AgentWorkOS+2Tinybird+2

适合场景：

Python 团队，希望快速从「脚本」进化到「有 UI、有控制台、有监控的 Agent 系统」。

3. Microsoft Agent Framework / AutoGen 系列

AutoGen：

早期的 多 Agent 会话框架，主打「让多个 Agent 对话解决任务」。

Microsoft Agent Framework（新一代）：

面向 .NET 和 Python 的开源开发包，统一了 AutoGen 和 Semantic Kernel 的能力。
提供 Agent、工具、工作流、记忆、评估 等组件
和 Azure 生态（OpenAI / 应用服务 / 监控）整合度高

适合场景：

Azure 体系、.NET / Python 混合技术栈、对企业级集成要求高的团队。

4. CrewAI：把 Agent 组织成「一支团队」

定位：

一个 Python 框架，专门用来编排一组「角色扮演」Agent（Crew）。CrewAI+4CrewAI Documentation+4GitHub+4

特点：

把 Agent 抽象成 角色 + 目标 + 工具
支持任务分解、协作、汇报（很像真实团队的工作流）
编程模型简单，上手快

适合场景：

做内容生产、调研、写代码等「多角色协作任务」，想要快速落地多 Agent 原型。

5. 其他值得关注的方向（简单带一下）

Simpliflow：轻量级、声明式配置的 Agentic workflow 框架，强调用 JSON 配置线性流程、集成 LiteLLM，适合快速搭建确定性流程。
Aragog：研究型系统，用「动态模型路由」优化 agentic workflow 的吞吐和延迟，说明未来生产环境会越来越重点关注 多模型路由 + 成本/性能权衡。

三、生产级 Agentic Workflow 一般长成什么样？

不管选哪个框架，成熟的生产架构大致有这些共性：

1. 流程建模：DAG / 状态机，而不是一堆 if-else

LangGraph 用 StateGraph 来描述状态机：节点是函数，边定义执行顺序和条件
Agno 用 Workflow(steps=[…]) 串联步骤，每个步骤可以调用不同的 Agent

在工程实践上，一般会：

把流程拆成：接收请求 → 解析意图 → 调用工具/RAG → 汇总 → 审核 → 输出
在框架里用节点/步骤显式表达，方便调试和版本管理

2. 状态管理：让 Agent 有「长期记忆」

生产环境常见做法：

用框架自带的 state store / memory / persistence：

如 LangGraph 的持久化存储，能恢复中断的对话和流程
Agno 提供 Sqlite / Postgres 等 DB 后端，用于存对话和会话状态

业务侧一般会做：

为每个用户 / 会话 / 任务分配唯一 ID
所有 Agent 和节点都通过这个 ID 读写状态

3. 观测与调试：Tracing 是必选项

在多 Agent、长工作流场景下，没有 tracing 等于「盲飞」。

框架 + Observability 组合：

LangGraph + Langfuse：完整 tracing、span、可视化工作流调用链Langfuse
Agno 自带 AgentOS 控制面板，可以查看会话/Agent 状态
也可以接入 AgentOps、OpenTelemetry 等通用方案AgentOps

实际落地提议：

每个节点要有明确的输入/输出日志
所有工具调用（数据库、HTTP、内部服务）都要有统一审计

4. 安全、风控与人类介入

工具调用需要：

白名单 / 黑名单
参数校验（防止 prompt 注入让 Agent 拼出危险命令）

在关键步骤加 human-in-the-loop：

列如：真正执行转账、下单、写入生产库前，必须人工点击确认

对于敏感场景（金融、医疗），一般会加：

规则引擎 + 模型判别双重保险

5. 评估与持续优化

用真实或构造的 评估集 回放 workflow，衡量：

准确率、回答质量、工具调用正确率
成本、平均延迟、长尾错误类型

像 Aragog 这种研究，说明未来会越来越多：

动态模型选择（大模型/小模型切换）
成本/性能的自适应调度arXiv

四、一个超简的 LangGraph 多 Agent Workflow Demo

下面这个 Demo 展示了一个超级简化的生产思路：

用户提一个问题 →
「研究员 Agent」先梳理背景和要点 →
「写手 Agent」基于研究结果写一份结构化说明。

1. 安装依赖（示例）

pip install -U langgraph langchain-openai langchain-core

并设置环境变量：

export OPENAI_API_KEY="your-api-key"

2. 代码示例：双节点工作流

说明：代码基于 LangGraph 官方文档的用法做的简化示例，实际项目可根据自身需要扩展。

from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

# 这里用一个模型实例，生产环境一般会通过配置/网关统一管理模型
llm = ChatOpenAI(model="gpt-4.1-mini", temperature=0.2)


# --- 节点 1：Research Agent，负责梳理背景和要点 ---
def research_node(state: MessagesState):
    user_question = state["messages"][-1].content

    prompt = f"""
你是一个企业内部的“研究员 Agent”，负责在执行前先把问题拆解清楚。

请针对下面这个问题，输出：
1）问题背景和业务场景假设
2）需要查清的关键子问题（用条目列出）
3）可能需要调用的内部/外部数据源或工具
4）后续给“写作 Agent”的要点总结（用要点列出）

用户问题：{user_question}
"""
    reply = llm.invoke([HumanMessage(content=prompt)])
    # 把模型回复追加到 messages 中，作为新的状态
    return {"messages": state["messages"] + [reply]}


# --- 节点 2：Writer Agent，负责写成结构化说明 ---
def writer_node(state: MessagesState):
    research_notes = state["messages"][-1].content

    prompt = f"""
你是“写作 Agent”，负责把研究员的要点，写成一份面向业务同学的说明文。

要求：
- 结构清晰，有小标题
- 解释清楚为什么要这么做
- 简要说明可以如何落地实施（步骤或提议）

以下是研究员整理的要点，请据此生成说明文（不要照抄原文，要自己组织语言）：

{research_notes}
"""
    reply = llm.invoke([HumanMessage(content=prompt)])
    return {"messages": state["messages"] + [reply]}


# --- 构建 LangGraph 工作流 ---
graph = StateGraph(MessagesState)

# 注册节点
graph.add_node("research", research_node)
graph.add_node("writer", writer_node)

# 配置执行顺序：START -> research -> writer -> END
graph.add_edge(START, "research")
graph.add_edge("research", "writer")
graph.add_edge("writer", END)

# 编译得到可调用的 app
app = graph.compile()


if __name__ == "__main__":
    # 初始 state：只包含用户的一条消息
    init_state = {
        "messages": [
            HumanMessage(
                content="我们想在生产环境中落地一个 Agentic AI 工作流，优化内部知识库搜索和问答体验，该怎么设计整体架构？"
            )
        ]
    }

    result = app.invoke(init_state)

    print("====== 最终对话 ======")
    for msg in result["messages"]:
        role = "用户" if msg.type == "human" else "AI"
        print(f"[{role}] {msg.content}
")