LangGraph 提供了两个不同的 API 来构建智能体工作流： **Graph API** 和 **Functional API**。两个 API 共享相同的底层运行时，可以在同一应用程序中一起使用，但它们是为不同的用例和开发偏好而设计的。

本指南将帮助您根据您的具体需求了解何时使用每个 API。

快速决策指南

使用 **Graph API** 当您需要： - **复杂的工作流可视化** 用于调试和文档 - **显式状态管理** 跨多个节点共享数据 - **条件分支** 具有多个决策点 - **并行执行路径** 需要稍后合并 - **团队协作** 视觉表示有助于理解

使用 **Functional API** 当您想要： - **最少的代码改动** 对现有过程式代码 - **标准控制流** (if/else, loops, function calls) - **函数作用域的状态** 无需显式状态管理 - **快速原型开发** 更少的样板代码 - **线性工作流** 配合简单的分支逻辑

详细对比

何时使用图 API

当 图 API 使用声明式方法，您可以在其中定义节点、边和共享状态来创建可视化图形结构。

1. 复杂的决策树和分支逻辑

当您的工作流有多个依赖于各种条件的决策点时，图 API 使这些分支变得清晰可见，便于可视化。

  StateGraph,
  StateSchema,
  MessagesValue,
  START,
  END,
  type GraphNode,
  type ConditionalEdgeRouter,
} from "@langchain/langgraph";

// Graph API: Clear visualization of decision paths
const AgentState = new StateSchema({
  messages: MessagesValue,
  currentTool: z.string(),
  retryCount: z.number().default(0),
});

const shouldContinue: ConditionalEdgeRouter<typeof AgentState> = (state) => {
  if (state.retryCount > 3) {
    return END;
  } else if (state.currentTool === "search") {
    return "processSearch";
  } else {
    return "callLlm";
  }
};

const workflow = new StateGraph(AgentState)
  .addNode("callLlm", callLlmNode)
  .addNode("processSearch", searchNode)
  .addConditionalEdges("callLlm", shouldContinue);

2. 跨多个组件的状态管理

当您需要在工作流的不同部分之间共享和协调状态时，图 API 的显式状态管理非常有益。

// Multiple nodes can access and modify shared state
const WorkflowState = new StateSchema({
  userInput: z.string(),
  searchResults: z.array(z.string()).default([]),
  generatedResponse: z.string().optional(),
  validationStatus: z.string().optional(),
});

const searchNode: GraphNode<typeof WorkflowState> = async (state) => {
  // Access shared state
  const results = await search(state.userInput);
  return { searchResults: results };
};

const validationNode: GraphNode<typeof WorkflowState> = async (state) => {
  // Access results from previous node
  const isValid = await validate(state.generatedResponse);
  return { validationStatus: isValid ? "valid" : "invalid" };
};

3. 带同步的并行处理

当您需要并行运行多个操作然后合并其结果时，图 API 可以自然地处理这种情况。

// Parallel processing of multiple data sources
workflow
  .addNode("fetchNews", fetchNews)
  .addNode("fetchWeather", fetchWeather)
  .addNode("fetchStocks", fetchStocks)
  .addNode("combineData", combineAllData)
  // All fetch operations run in parallel
  .addEdge(START, "fetchNews")
  .addEdge(START, "fetchWeather")
  .addEdge(START, "fetchStocks")
  // Combine waits for all parallel operations to complete
  .addEdge("fetchNews", "combineData")
  .addEdge("fetchWeather", "combineData")
  .addEdge("fetchStocks", "combineData");

4. 团队协作开发和文档

图 API 的可视化特性使团队更容易理解、记录和维护复杂的工作流。

// Clear separation of concerns - each team member can work on different nodes
workflow
  .addNode("dataIngestion", dataTeamFunction)
  .addNode("mlProcessing", mlTeamFunction)
  .addNode("businessLogic", productTeamFunction)
  .addNode("outputFormatting", frontendTeamFunction);

何时使用函数式 API

当 函数式 API 使用命令式方法，将 LangGraph 功能集成到标准过程式代码中。

1. 现有过程式代码

当你拥有使用标准控制流的现有代码，并希望以最少的重构添加 LangGraph 功能时。

// Functional API: Minimal changes to existing code
const processUserInput = task(
  "processUserInput",
  async (userInput: string) => {
    // Existing function with minimal changes
    return { processed: userInput.toLowerCase().trim() };
  }
);

const workflow = entrypoint(
  { checkpointer },
  async (userInput: string) => {
    // Standard control flow
    const processed = await processUserInput(userInput);

    let response: string;
    if (processed.processed.includes("urgent")) {
      response = await handleUrgentRequest(processed);
    } else {
      response = await handleNormalRequest(processed);
    }

    return response;
  }
);

2. 具有简单逻辑的线性工作流

当你的工作流主要是顺序的，具有简单的条件逻辑时。

const essayWorkflow = entrypoint(
  { checkpointer },
  async (topic: string) => {
    // Linear flow with simple branching
    let outline = await createOutline(topic);

    if (outline.points.length < 3) {
      outline = await expandOutline(outline);
    }

    const draft = await writeDraft(outline);

    // Human review checkpoint
    const feedback = interrupt({ draft, action: "Please review" });

    let finalEssay: string;
    if (feedback === "approve") {
      finalEssay = draft;
    } else {
      finalEssay = await reviseEssay(draft, feedback);
    }

    return { essay: finalEssay };
  }
);

3. 快速原型开发

当你想快速测试想法，而无需定义状态模式和图结构的开销时。

const quickPrototype = entrypoint(
  { checkpointer },
  async (data: Record<string, unknown>) => {
    // Fast iteration - no state schema needed
    const step1Result = await processStep1(data);
    const step2Result = await processStep2(step1Result);

    return { finalResult: step2Result };
  }
);

4. 函数作用域状态管理

当你的状态自然地限定在单个函数中，不需要广泛共享时。

const analyzeDocument = task("analyzeDocument", async (document: string) => {
  // Local state management within function
  const sections = extractSections(document);
  const summaries = await Promise.all(sections.map(summarize));
  const keyPoints = extractKeyPoints(summaries);

  return {
    sections: sections.length,
    summaries,
    keyPoints,
  };
});

const documentProcessor = entrypoint(
  { checkpointer },
  async (document: string) => {
    const analysis = await analyzeDocument(document);
    // State is passed between functions as needed
    return await generateReport(analysis);
  }
);

结合使用两种 API

你可以在同一应用中同时使用两种 API。这在系统的不同部分有不同需求时非常有用。

  StateGraph,
  StateSchema,
  entrypoint,
  type GraphNode,
} from "@langchain/langgraph";

// Define state for complex multi-agent coordination
const CoordinationState = new StateSchema({
  rawData: z.record(z.string(), z.unknown()),
  processedData: z.record(z.string(), z.unknown()).optional(),
});

// Simple data processing using Functional API
const dataProcessor = entrypoint({}, async (rawData: Record<string, unknown>) => {
  const cleaned = await cleanData(rawData);
  const transformed = await transformData(cleaned);
  return transformed;
});

// Use the functional API result in the graph
const orchestratorNode: GraphNode<typeof CoordinationState> = async (state) => {
  const processedData = await dataProcessor.invoke(state.rawData);
  return { processedData };
};

// Complex multi-agent coordination using Graph API
const coordinationGraph = new StateGraph(CoordinationState)
  .addNode("orchestrator", orchestratorNode)
  .addNode("agentA", agentANode)
  .addNode("agentB", agentBNode);

API 之间的迁移

从函数式 API 到图 API

当你的函数式工作流变得复杂时，可以迁移到图 API：

// Before: Functional API
const complexWorkflow = entrypoint(
  { checkpointer },
  async (inputData: Record<string, unknown>) => {
    const step1 = await processStep1(inputData);

    let result: unknown;
    if (step1.needsAnalysis) {
      const analysis = await analyzeData(step1);
      if (analysis.confidence > 0.8) {
        result = await highConfidencePath(analysis);
      } else {
        result = await lowConfidencePath(analysis);
      }
    } else {
      result = await simplePath(step1);
    }

    return result;
  }
);

// After: Graph API

  StateGraph,
  StateSchema,
  type GraphNode,
  type ConditionalEdgeRouter,
} from "@langchain/langgraph";

const WorkflowState = new StateSchema({
  inputData: z.record(z.string(), z.unknown()),
  step1Result: z.record(z.string(), z.unknown()).optional(),
  analysis: z.record(z.string(), z.unknown()).optional(),
  finalResult: z.unknown().optional(),
});

const shouldAnalyze: ConditionalEdgeRouter<typeof WorkflowState> = (state) => {
  return state.step1Result?.needsAnalysis ? "analyze" : "simplePath";
};

const confidenceCheck: ConditionalEdgeRouter<typeof WorkflowState> = (state) => {
  return (state.analysis?.confidence as number) > 0.8
    ? "highConfidence"
    : "lowConfidence";
};

const workflow = new StateGraph(WorkflowState)
  .addNode("step1", processStep1Node)
  .addConditionalEdges("step1", shouldAnalyze)
  .addNode("analyze", analyzeDataNode)
  .addConditionalEdges("analyze", confidenceCheck);
// ... add remaining nodes and edges

从图 API 到函数式 API

当你的图对于简单的线性流程变得过于复杂时：

// Before: Over-engineered Graph API
const SimpleState = new StateSchema({
  input: z.string(),
  step1: z.string().optional(),
  step2: z.string().optional(),
  result: z.string().optional(),
});

// After: Simplified Functional API
const simpleWorkflow = entrypoint(
  { checkpointer },
  async (inputData: string) => {
    const step1 = await processStep1(inputData);
    const step2 = await processStep2(step1);
    return await finalizeResult(step2);
  }
);

总结

选择 **图 API** 当你需要对工作流结构、复杂分支、并行处理或团队协作进行显式控制时。

选择 **函数式 API** 当你想以最少的更改向现有代码添加 LangGraph 功能，拥有简单的线性工作流，或需要快速原型开发功能时。

两种 API 都提供相同的核心 LangGraph 功能（持久化、流式处理、人机交互、记忆），但以不同的范式打包它们，以适应不同的开发风格和使用场景。