工作流模式（上）

一个不够用

你有没有遇到过这样的场景？

这些问题，本质上都是编排问题：不是 LLM 不够聪明，而是你没有给它安排好工作流。

开始之前

基础设置

安装依赖

npm install @langchain/langgraph @langchain/core @langchain/anthropic

初始化 LLM

// 导入 DeepSeek 聊天模型
import { ChatDeepSeek } from "@langchain/deepseek";

const llm = new ChatDeepSeek({
  model: "deepseek-chat",
  apiKey: "<your_deepseek_key>"
});

LLM 增强能力

import * as z from "zod";                          // Zod：运行时类型校验库，用来定义 Schema
import { tool } from "@langchain/core/tools";      // LangChain 工具定义函数

// ========== 1. 结构化输出 ==========
// 场景：你希望 LLM 不返回自由文本，而是返回严格符合 Schema 的 JSON 对象
// 原理：withStructuredOutput 内部会把 Zod Schema 转成 JSON Schema 发给模型，
//        模型会被强制按格式输出，返回结果会经过 Zod 校验
const SearchQuery = z.object({
  search_query: z.string().describe("Optimized web search query."),    // .describe() 的文字会作为字段说明传给模型
  justification: z.string().describe("Why this query is relevant."),
});
const structuredLlm = llm.withStructuredOutput(SearchQuery);
// 用法：const result = await structuredLlm.invoke("...");
// result = { search_query: "...", justification: "..." }  ← 保证是合法 JSON

// ========== 2. 工具调用 ==========
// 场景：你希望 LLM 能"调用"外部函数（如计算器、搜索、数据库查询）
// 原理：bindTools 把工具定义发给模型，模型决定是否需要调用工具、传什么参数；
//        返回的 aiMsg.tool_calls 里包含工具名和参数，由你来实际执行
const multiply = tool(
  ({ a, b }) => a * b,                             // 第 1 个参数：工具的实际执行函数
  {
    name: "multiply",                               // 工具名称（模型通过这个名字调用）
    description: "Multiply two numbers",            // 工具描述（模型根据描述判断何时使用）
    schema: z.object({                              // 参数 Schema（告诉模型需要传哪些参数）
      a: z.number(),
      b: z.number(),
    }),
  }
);
const llmWithTools = llm.bindTools([multiply]);
// 用法：const aiMsg = await llmWithTools.invoke("3乘以5等于几？");
// aiMsg.tool_calls = [{ name: "multiply", args: { a: 3, b: 5 } }]

工作流与智能体

LangGraph 的优势在于：工作流和智能体用同一套 API（StateGraph、addNode、addEdge），你随时可以把工作流的某个节点替换成智能体决策。

模式一：链式调用

每个 LLM 调用处理前一个调用的输出，像流水线一样一步步推进。适合可以分解为多个可验证子步骤的任务。

链式调用代码

import { StateGraph, StateSchema, GraphNode, ConditionalEdgeRouter, START, END } from "@langchain/langgraph";
import * as z from "zod";

// 定义图状态 —— 每个字段对应流水线的一个阶段
const State = new StateSchema({
  topic: z.string(),          // 输入主题
  joke: z.string(),           // 原始笑话
  improvedJoke: z.string(),   // 改进后的笑话
  finalJoke: z.string(),      // 最终润色
});

// 节点 1：生成初始笑话
const generateJoke: GraphNode<typeof State> = async (state) => {
  const msg = await llm.invoke(`Write a short joke about ${state.topic}`);
  return { joke: msg.content };
};

// 节点 2：改进笑话（添加双关语）
const improveJoke: GraphNode<typeof State> = async (state) => {
  const msg = await llm.invoke(
    `Make this joke funnier by adding wordplay: ${state.joke}`
  );
  return { improvedJoke: msg.content };
};

// 节点 3：最终润色（添加反转）
const polishJoke: GraphNode<typeof State> = async (state) => {
  const msg = await llm.invoke(
    `Add a surprising twist to this joke: ${state.improvedJoke}`
  );
  return { finalJoke: msg.content };
};

📄 完整代码：02_prompt_chaining.ts

质量门控

核心机制：addConditionalEdges + ConditionalEdgeRouter

// 质量门函数 —— 检查笑话是否包含 ? 或 !（简单判断是否有笑点）
const checkPunchline: ConditionalEdgeRouter<typeof State, "improveJoke"> = (state) => {
  if (state.joke?.includes("?") || state.joke?.includes("!")) {
    return "Pass";   // 有笑点 → 继续改进
  }
  return "Fail";     // 没笑点 → 直接结束
};

// 构建完整工作流
const chain = new StateGraph(State)
  .addNode("generateJoke", generateJoke)
  .addNode("improveJoke", improveJoke)
  .addNode("polishJoke", polishJoke)
  // 入口
  .addEdge(START, "generateJoke")
  // 条件边 —— generateJoke 之后根据质量门控决定去向
  .addConditionalEdges("generateJoke", checkPunchline, {
    Pass: "improveJoke",    // 通过 → 改进
    Fail: END         // 不通过 → 直接结束
  })
  .addEdge("improveJoke", "polishJoke")
  .addEdge("polishJoke", END)
  .compile();

// 调用
const result = await chain.invoke({ topic: "cats" });

📄 完整代码：02_prompt_chaining.ts

模式二：并行执行

多个 LLM 调用同时运行，互不等待。适合子任务之间没有依赖关系的场景。也可以对同一任务运行多次，取最优结果。

并行执行代码

import { StateGraph, StateSchema, GraphNode, START, END } from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  topic: z.string(),
  joke: z.string(),
  story: z.string(),
  poem: z.string(),
  combinedOutput: z.string(),
});

// 三个并行的 LLM 节点 —— 互不依赖，同时运行
const callLlm1: GraphNode<typeof State> = async (state) => {
  const msg = await llm.invoke(`Write a joke about ${state.topic}`);
  return { joke: msg.content };
};

const callLlm2: GraphNode<typeof State> = async (state) => {
  const msg = await llm.invoke(`Write a story about ${state.topic}`);
  return { story: msg.content };
};

const callLlm3: GraphNode<typeof State> = async (state) => {
  const msg = await llm.invoke(`Write a poem about ${state.topic}`);
  return { poem: msg.content };
};

// 聚合节点 —— 等所有并行节点完成后执行
const aggregator: GraphNode<typeof State> = async (state) => {
  const combined = `Story:\n${state.story}\n\nJoke:\n${state.joke}\n\nPoem:\n${state.poem}`;
  return { combinedOutput: combined };
};

// 关键：START 连接到三个节点 → 自动并行
const parallelWorkflow = new StateGraph(State)
  .addNode("callLlm1", callLlm1)
  .addNode("callLlm2", callLlm2)
  .addNode("callLlm3", callLlm3)
  .addNode("aggregator", aggregator)
  .addEdge(START, "callLlm1")    // 并行入口
  .addEdge(START, "callLlm2")    // 并行入口
  .addEdge(START, "callLlm3")    // 并行入口
  .addEdge("callLlm1", "aggregator")   // 汇入聚合
  .addEdge("callLlm2", "aggregator")
  .addEdge("callLlm3", "aggregator")
  .addEdge("aggregator", END)
  .compile();

const result = await parallelWorkflow.invoke({ topic: "cats" });

📄 完整代码：02_parallelization.ts

模式三：智能路由

先分类输入，再路由到专门的处理节点。每个处理节点只做自己擅长的事，输出质量更高。

路由代码实现

import { StateGraph, StateSchema, GraphNode, ConditionalEdgeRouter, START, END } from "@langchain/langgraph";
import * as z from "zod";

// 路由决策 Schema —— 用结构化输出让 LLM 做选择题
const routeSchema = z.object({
  step: z.enum(["poem", "story", "joke"]).describe("The next step in the routing process"),
});
const router = llm.withStructuredOutput(routeSchema);

const State = new StateSchema({
  input: z.string(),
  decision: z.string(),
  output: z.string(),
});

// 三个专门的处理节点
const llmCall1: GraphNode<typeof State> = async (state) => {
  const result = await llm.invoke([
    { role: "system", content: "You are an expert storyteller." },
    { role: "user", content: state.input },
  ]);
  return { output: result.content };
};
// llmCall2（喜剧专家）、llmCall3（诗歌专家）结构相同，省略……

// 路由节点 —— 调用结构化输出做分类
const llmCallRouter: GraphNode<typeof State> = async (state) => {
  const decision = await router.invoke([
    { role: "system", content: "Route to story, joke, or poem based on the user's request." },
    { role: "user", content: state.input },
  ]);
  return { decision: decision.step };
};

// 条件边 —— 根据路由决策分发
const routeDecision: ConditionalEdgeRouter<typeof State, "llmCall1" | "llmCall2" | "llmCall3"> = (state) => {
  if (state.decision === "story") return "llmCall1";
  else if (state.decision === "joke") return "llmCall2";
  else return "llmCall3";
};

📄 完整代码：02_routing.ts

路由图构建

// 构建路由工作流
const routerWorkflow = new StateGraph(State)
  .addNode("llmCall1", llmCall1)
  .addNode("llmCall2", llmCall2)
  .addNode("llmCall3", llmCall3)
  .addNode("llmCallRouter", llmCallRouter)
  // 入口 → 路由器
  .addEdge(START, "llmCallRouter")
  // 路由器 → 条件分发（三选一）
  .addConditionalEdges("llmCallRouter", routeDecision, ["llmCall1", "llmCall2", "llmCall3"])
  // 三个处理节点各自通向终点
  .addEdge("llmCall1", END)
  .addEdge("llmCall2", END)
  .addEdge("llmCall3", END)
  .compile();

// 测试路由 —— 输入 "写个关于猫的笑话" 会走 llmCall2
const result = await routerWorkflow.invoke({
  input: "Write me a joke about cats"
});
console.log(result.output);

📄 完整代码：02_routing.ts

排错指南

动手练习

构建一个内容处理管道

需求：对一篇英文文章做三步处理。

提示

• State 需要：article、summary、translation、finalOutput
• 质量门控检查 summary.split(" ").length
• 参考链式调用的 addConditionalEdges 写法

参考答案

import { StateGraph, StateSchema, GraphNode, ConditionalEdgeRouter, START, END } from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  article: z.string(),
  summary: z.string(),
  translation: z.string(),
  finalOutput: z.string(),
});

// Step 1：生成摘要
const summarize: GraphNode<typeof State> = async (state) => {
  const msg = await llm.invoke(
    `Summarize this article in under 50 words:\n\n${state.article}`
  );
  return { summary: msg.content as string };
};

// Step 2：翻译为中文
const translate: GraphNode<typeof State> = async (state) => {
  const msg = await llm.invoke(
    `Translate the following summary to Chinese:\n\n${state.summary}`
  );
  return { translation: msg.content as string };
};

// Step 3：格式化输出
const format: GraphNode<typeof State> = async (state) => {
  const output = `=== 文章摘要 ===\n${state.translation}\n${"—".repeat(20)}`;
  return { finalOutput: output };
};

// 质量门控：摘要太长（超过 80 词）则跳过翻译
const checkSummaryLength: ConditionalEdgeRouter<typeof State, "translate"> = (state) => {
  const wordCount = state.summary.split(" ").length;
  return wordCount <= 80 ? "Pass" : "Fail";
};

// 构建图
const pipeline = new StateGraph(State)
  .addNode("summarize", summarize)
  .addNode("translate", translate)
  .addNode("format", format)
  .addEdge(START, "summarize")
  .addConditionalEdges("summarize", checkSummaryLength, {
    Pass: "translate",
    Fail: END,
  })
  .addEdge("translate", "format")
  .addEdge("format", END)
  .compile();

本章回顾

延伸阅读