工作流模式（下）

更复杂的场景

第 02 章的三种模式遇到了新问题

这三个问题对应三种新模式：编排模式解决动态分派、评估优化解决反馈循环、智能体解决自主决策。

模式四：编排模式

编排者（Orchestrator）把大任务拆成若干子任务，分派给工作者（Worker），等所有工作者完成后综合结果。关键区别：子任务的数量在运行时才确定。

Send 动态分发

核心 API：Send 类

import { StateGraph, StateSchema, ReducedValue, GraphNode, Send, START, END } from "@langchain/langgraph";

Send 让你从条件边函数里动态地把任务分发到指定节点。每次 new Send("节点名", { 数据 }) 就创建一个独立的工作者调用。

// 条件边路由函数 —— 返回 Send 数组，每个 Send 创建一个工作者
const assignWorkers: ConditionalEdgeRouter<typeof State, "llmCall"> = (state) => {
  // state.sections 是编排者 LLM 拆分出的子任务列表
  // 有几个 section，就创建几个工作者
  return state.sections.map(
    (section) => new Send("llmCall", { sections: [section] })
  );
};

为什么需要 ReducedValue？

多个工作者并行运行，每个都往同一个 state 字段写结果。如果用普通字段，后者会覆盖前者。ReducedValue 用 reducer 函数把所有工作者的结果合并到一个数组里。

// 用 ReducedValue 定义一个会"累积"的字段
const State = new StateSchema({
  topic: z.string(),
  sections: z.array(z.object({
    title: z.string(),
    description: z.string(),
  })),
  completedSections: new ReducedValue(
    z.array(z.object({ title: z.string(), content: z.string() })),
    // reducer: 把新结果追加到已有数组
    (existing, update) => [...(existing ?? []), ...update],
  ),
  finalReport: z.string(),
});

📄 完整代码：03_send_api.ts

编排模式代码

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

const State = new StateSchema({
  topic: z.string(),
  sections: z.array(z.object({ title: z.string(), description: z.string() })),
  completedSections: new ReducedValue(
    z.array(z.object({ title: z.string(), content: z.string() })),
    (existing, update) => [...(existing ?? []), ...update],
  ),
  finalReport: z.string(),
});

// 编排者：分析任务，拆分成子任务列表
const orchestrator: GraphNode<typeof State> = async (state) => {
  const structLlm = llm.withStructuredOutput(
    z.object({ sections: z.array(z.object({
      title: z.string().describe("Section title"),
      description: z.string().describe("What to cover"),
    })) })
  );
  const result = await structLlm.invoke([
    { role: "system", content: "Plan a report sections." },
    { role: "user", content: state.topic },
  ]);
  return { sections: result.sections };
};

// 工作者：根据子任务描述生成内容（会被 Send 动态创建多次）
const llmCall: GraphNode<typeof State> = async (state) => {
  const section = state.sections[0]; // Send 传入 { sections: [section] }，这里取第 0 个
  const msg = await llm.invoke(
    `Write a section: "${section.title}" covering: ${section.description}`
  );
  return { completedSections: [{ title: section.title, content: msg.content }] };
};

📄 完整代码：03_orchestrator_worker.ts

编排图构建

// 综合者：把所有工作者生成的内容合成最终报告
const synthesizer: GraphNode<typeof State> = async (state) => {
  const parts = state.completedSections
    .map((s) => `## ${s.title}\n${s.content}`)
    .join("\n\n");
  const msg = await llm.invoke(
    `Format into a final report:\n\n${parts}`
  );
  return { finalReport: msg.content as string };
};

// 条件边：编排者 → 动态分派工作者
const assignWorkers = (state) => {
  return state.sections.map(
    (section) => new Send("llmCall", { sections: [section] })
  );
};

// 构建图
const orchestratorWorkflow = new StateGraph(State)
  .addNode("orchestrator", orchestrator)
  .addNode("llmCall", llmCall)
  .addNode("synthesizer", synthesizer)
  .addEdge(START, "orchestrator")
  // 关键：条件边用 Send 动态创建多个 llmCall 实例
  .addConditionalEdges("orchestrator", assignWorkers, ["llmCall"])
  // 所有工作者完成后汇入综合者
  .addEdge("llmCall", "synthesizer")
  .addEdge("synthesizer", END)
  .compile();

const result = await orchestratorWorkflow.invoke({
  topic: "Create a report on the history of AI"
});

📄 完整代码：03_orchestrator_worker.ts

模式五：评估优化

一个 LLM 负责生成，另一个 LLM 负责评估。评估不合格就带着反馈回到生成者重新来过，直到合格为止。

评估优化代码

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

// 评估结果 Schema
const FeedbackSchema = z.object({
  grade: z.enum(["funny", "not funny"]).describe("Evaluation grade"),
  feedback: z.string().describe("Feedback for improvement"),
});

const State = new StateSchema({
  joke: z.string(),
  topic: z.string(),
  feedback: z.string(),
  funnyOrNot: z.string(),
});

// 生成者：写笑话（如果之前有反馈，就参考反馈改进）
const llmCallGenerator: GraphNode<typeof State> = async (state) => {
  const prompt = state.feedback
    ? `Write a joke about ${state.topic}. Improve based on feedback: ${state.feedback}`
    : `Write a joke about ${state.topic}`;
  const msg = await llm.invoke(prompt);
  return { joke: msg.content as string };
};

// 评估者：判断笑话好不好笑
const llmCallEvaluator: GraphNode<typeof State> = async (state) => {
  const evaluator = llm.withStructuredOutput(FeedbackSchema);
  const result = await evaluator.invoke(
    `Evaluate this joke: "${state.joke}". Is it funny? Give feedback.`
  );
  return {
    funnyOrNot: result.grade,
    feedback: result.feedback,
  };
};

// 条件边：合格 → 结束，不合格 → 带反馈重来
const routeJoke: ConditionalEdgeRouter<typeof State, "llmCallGenerator"> = (state) => {
  return state.funnyOrNot === "funny" ? "Accepted" : "Rejected";
};

📄 完整代码：03_evaluator_optimizer.ts

评估优化图构建

// 构建评估优化工作流
const optimizerWorkflow = new StateGraph(State)
  .addNode("llmCallGenerator", llmCallGenerator)
  .addNode("llmCallEvaluator", llmCallEvaluator)
  .addEdge(START, "llmCallGenerator")
  .addEdge("llmCallGenerator", "llmCallEvaluator")
  // 评估者之后：合格结束，不合格循环
  .addConditionalEdges("llmCallEvaluator", routeJoke, {
    Accepted: END,
    Rejected: "llmCallGenerator",    // 带反馈回到生成者
  })
  .compile();

// 测试
const result = await optimizerWorkflow.invoke({
  topic: "Cats",
  joke: "",
  feedback: "",
  funnyOrNot: "",
});
console.log(result.joke);
console.log("Feedback received:", result.feedback);

// 安全改进版：限制最大重试 3 次
const MAX_ATTEMPTS = 3;

const routeJokeSafe: ConditionalEdgeRouter<typeof State, "llmCallGenerator"> = (state) => {
  if (state.funnyOrNot === "funny" || (state.attempts ?? 0) >= MAX_ATTEMPTS) {
    return "Accepted";   // 合格或超限 → 结束
  }
  return "Rejected";     // 不合格 → 重试
};

📄 完整代码：03_evaluator_optimizer.ts

模式六：智能体

智能体（Agent）是一种自主决策的循环结构：LLM 自己决定要调用什么工具、调几次、什么时候结束。不需要你预先定义流程。

工具调用循环

核心机制：shouldContinue + ToolNode

import { ToolNode } from "@langchain/langgraph/prebuilt";

// 条件边：检查是否需要继续调用工具
const shouldContinue: ConditionalEdgeRouter<typeof State, "toolNode"> = (state) => {
  const lastMessage = state.messages.at(-1);
  // 最后一条消息有 tool_calls → 执行工具
  if (lastMessage?.tool_calls?.length) return "toolNode";
  // 没有 tool_calls → LLM 已经给出最终回答，结束
  return END;
};

📄 完整代码：03_agent_tool_calling.ts

智能体代码

import { StateGraph, StateSchema, GraphNode, ConditionalEdgeRouter, Messages, START, END } from "@langchain/langgraph";
import { ToolNode } from "@langchain/langgraph/prebuilt";
import { tool } from "@langchain/core/tools";
import * as z from "zod";

// 定义工具
const multiply = tool(({ a, b }) => a * b, {
  name: "multiply", description: "Multiply two numbers",
  schema: z.object({ a: z.number(), b: z.number() }),
});
const add = tool(({ a, b }) => a + b, {
  name: "add", description: "Add two numbers",
  schema: z.object({ a: z.number(), b: z.number() }),
});

// LLM 绑定工具
const llmWithTools = llm.bindTools([multiply, add]);

// State 只需要一个 messages 数组
const State = new StateSchema({
  messages: Messages({}),   // 内置 Messages 注解
});

// LLM 节点：调用带工具的 LLM
const llmCall: GraphNode<typeof State> = async (state) => {
  const response = await llmWithTools.invoke(state.messages);
  return { messages: [response] };
};

// 条件边：检查是否继续调用工具
const shouldContinue: ConditionalEdgeRouter<typeof State, "toolNode"> = (state) => {
  const lastMessage = state.messages.at(-1);
  if (lastMessage?.tool_calls?.length) return "toolNode";
  return END;
};

📄 完整代码：03_agent_tool_calling.ts

智能体图构建

// 工具执行节点
const toolNode = new ToolNode([multiply, add]);

// 构建智能体工作流
const agentWorkflow = new StateGraph(State)
  .addNode("llmCall", llmCall)
  .addNode("toolNode", toolNode)
  .addEdge(START, "llmCall")
  // LLM 之后检查是否有工具调用
  .addConditionalEdges("llmCall", shouldContinue, ["toolNode", END])
  // 工具执行完后回到 LLM 继续判断
  .addEdge("toolNode", "llmCall")
  .compile();

// 测试：让智能体计算 (3 + 5) * 2
const result = await agentWorkflow.invoke({
  messages: [{ role: "user", content: "What is (3 + 5) * 2?" }],
});
console.log(result.messages.at(-1).content);
// 智能体会自动：调用 add(3,5) → 得到 8 → 调用 multiply(8,2) → 得到 16 → 输出结果

📄 完整代码：03_agent_tool_calling.ts

六种模式对比

排错指南

动手练习

构建一个研究助手

需求：用户输入一个研究主题，助手自动搜索、评估搜索质量、综合成报告。

提示

• 搜索节点不需要真的调用搜索引擎——模拟一下，用 LLM 生成搜索结果即可
• 参考评估优化模式的 addConditionalEdges 写法
• 在条件边里检查 attempts 是否超过 3

参考答案

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

const MAX_ATTEMPTS = 3;

const EvalResult = z.object({
  sufficient: z.boolean().describe("Are the search results sufficient?"),
  feedback: z.string().describe("Feedback for improvement if insufficient."),
});

const State = new StateSchema({
  topic: z.string(),
  searchResults: z.string(),
  feedback: z.string(),
  sufficient: z.boolean(),
  attempts: z.number(),
  finalReport: z.string(),
});

// 搜索节点：模拟搜索（实际项目中替换为真实搜索 API）
const search: GraphNode<typeof State> = async (state) => {
  const prompt = state.feedback
    ? `Research "${state.topic}" more thoroughly. Previous feedback: ${state.feedback}`
    : `Research "${state.topic}" and summarize key findings.`;
  const msg = await llm.invoke(prompt);
  return {
    searchResults: msg.content as string,
    attempts: (state.attempts ?? 0) + 1,
  };
};

// 评估节点：判断搜索结果是否充分
const evaluate: GraphNode<typeof State> = async (state) => {
  const evaluator = llm.withStructuredOutput(EvalResult);
  const result = await evaluator.invoke(
    `Evaluate these research results for "${state.topic}":\n${state.searchResults}`
  );
  return { sufficient: result.sufficient, feedback: result.feedback };
};

// 综合节点：生成最终报告
const synthesize: GraphNode<typeof State> = async (state) => {
  const msg = await llm.invoke(
    `Write a research summary based on:\n${state.searchResults}`
  );
  return { finalReport: msg.content as string };
};

// 条件边：充分或超限 → 综合，否则 → 重试
const routeEvaluation: ConditionalEdgeRouter<typeof State, "search"> = (state) => {
  if (state.sufficient || (state.attempts ?? 0) >= MAX_ATTEMPTS) return "Pass";
  return "Fail";
};

const workflow = new StateGraph(State)
  .addNode("search", search)
  .addNode("evaluate", evaluate)
  .addNode("synthesize", synthesize)
  .addEdge(START, "search")
  .addEdge("search", "evaluate")
  .addConditionalEdges("evaluate", routeEvaluation, {
    Pass: "synthesize",
    Fail: "search",
  })
  .addEdge("synthesize", END)
  .compile();

本章回顾

延伸阅读