基于本文回答

播面 播面

刷题像听歌，多听自然懂
0
评论
请从左侧选择一个分类

在 LangGraph 中，app.invoke() 和 app.stream() 在不同业务场景下的使用抉择

知识点图片

在 LangGraph 中，app.invoke() 和 app.stream() 是执行图（Graph）的两种核心方式。选择哪一种，本质上是在“等待最终结果”和“实时感知过程”之间做权衡。

以下是针对不同业务场景的详细抉择指南：

一、 app.invoke()：只关心最终结果的“一锤子买卖”

invoke() 会阻塞当前线程，直到 LangGraph 遍历完所有节点，到达 END 节点后，一次性返回最终的完整状态（State）。

1. 核心特点

  • 低开发复杂度：标准的 Request-Response 模式，易于与现有系统（如传统的 RESTful API）集成。
  • 高延迟感知：用户必须等待所有 LLM 调用、工具执行完毕，如果图的逻辑复杂（如多次循环反思），等待时间可能长达几十秒到数分钟。

2. 适用业务场景

  • 后台批处理 / 离线任务：
    • 场景：每日定时抓取新闻并生成总结报告、批量数据清洗、ETL 数据转换管道。
    • 原因：没有人类用户在屏幕前苦苦等待，系统只关心最终生成的报告是否成功存入数据库。
  • 微服务间的同步调用：
    • 场景：主系统调用 AI 评审系统审核一段代码，主系统需要拿到完整的 JSON 格式评审结果再进行下一步业务流转。
  • 极短链路的 AI 操作：
    • 场景：简单的意图分类、少量的文本翻译（整个 Graph 运行时间在 1-2 秒内）。
  • 严格依赖完整上下文的后处理：
    • 场景：需要拿到 Agent 最终生成的所有数据进行复杂的结构化校验（如 Pydantic 验证），如果不通过需要内部重试，对外不暴露中间错误状态。

二、 app.stream()：注重交互体验的“流式传输”

stream() 是一个生成器（Generator），它会在图执行的过程中不断产出（Yield）中间结果。在 LangGraph 中，stream 非常强大，因为它不仅能流式输出 LLM 的字（Token），还能流式输出图的节点状态（State）。

1. 核心特点

  • 极低的感知延迟：用户能立刻看到系统在工作（如输出文字、或者提示正在调用某个工具）。
  • 高开发复杂度：前端需要支持 SSE (Server-Sent Events) 或 WebSocket；后端需要解析流式数据并进行分发。

2. LangGraph 特有的 Streaming 模式及其业务场景

LangGraph 的 stream() 支持多个 stream_mode，你需要根据业务挑选：

  • 场景 A：C端智能助手 / Chatbot （打字机效果）
    • 配置：stream_mode="messages"
    • 抉择：如果你的业务是类似 ChatGPT 的问答机器人，用户需要实时看到 LLM 生成的每一个字。必须使用此模式。
  • 场景 B：复杂 Agent 工作流可视化（展示思考过程）
    • 配置：stream_mode="updates" 或 stream_mode="values"
    • 抉择：如果你的 Agent 会调用工具、会自我反思（例如：AutoGPT）。用户需要看到进度条或状态提示（如：“🔍 正在搜索网页...”、“🧠 正在反思结果...”、“✅ 搜索完毕”）。此时流式输出的不是字，而是图每次流转经过节点时的状态更新。
  • 场景 C：Human-in-the-loop（人机协同/断点审批）
    • 抉择：当 Agent 执行敏感操作（如付款、发送邮件）需要人类介入审批时。stream 可以在到达断点（interrupt）时将当前状态推给前端，前端渲染审批按钮，等待用户操作后继续。

三、 综合对比抉择矩阵

评估维度 app.invoke() 场景 app.stream() 场景
受众目标 机器程序、后台服务、异步队列 真实人类用户 (C端/B端 UI)
网络协议 传统 HTTP 请求 (REST API) Server-Sent Events (SSE), WebSockets
响应时效性 总体执行时间短，或不在乎等待 执行时间长 (>3秒)，必须降低用户感知延迟
错误容忍度 内部可静默重试，失败返回统一错误 允许向用户暴露中间步骤的失败或重试过程
前端复杂度 低（接收一次 JSON 即可） 高（需要解析数据流，处理 Markdown 渲染和组件切换）
资源占用 较小（一次性释放） 较大（长时间保持 TCP 长连接）

四、 业务架构师的“决策树”

当你拿到一个基于 LangGraph 的业务需求时，问自己以下三个问题：

  1. 有没有人在 UI 界面前“傻等”？
    • 没有（比如是消息队列消费、定时任务） 👉 直接选 invoke()。
    • 有 👉 进入第 2 步。
  2. 整个 Graph 执行完大约需要多久？
    • 小于 2 秒（简单的单节点 LLM 或规则判断）👉 选 invoke()，为了这 2 秒搞流式传输性价比太低。
    • 大于 2-3 秒，或者包含未知的多次循环（Tool Calling / ReAct） 👉 必须选 stream()。
  3. 如果选了 Stream，你需要给前端展示什么？
    • 只要最终答案的逐字输出 👉 使用 stream(..., stream_mode="messages")。
    • 需要展示 Agent 每一步调用了什么工具、搜索了什么词 👉 使用 stream(..., stream_mode="updates")。
    • 两者都要（当前最流行的 AI 交互模式：上面是过程指示器，下面是逐字回复）👉 使用多路流合并 stream(..., stream_mode=["messages", "updates"])，并在后端做好数据分类再推给前端。

五、 进阶建议

在实际的大型企业项目中，往往不是二选一，而是结合使用：

  • 混合架构：对外层暴露 WebSocket 提供 stream() 能力以服务 Web 端；同时封装一个异步消息队列消费者，在后台使用 invoke() 处理批量任务。
  • 状态保存（Checkpointer）：无论使用 invoke 还是 stream，在 LangGraph 中都强烈建议开启持久化（如 PostgresSaver），这样即使 stream 过程中网络断开，也可以通过 invoke 恢复状态或重新接管当前会话。
00:00
00:00