课 2 · 最小 StateGraph 实现

本课目标

从零实现一个最小 StateGraph，覆盖 State、Node、Edge、Reducer 四个核心概念，理解 LangGraph 的底层原理。

这节课写的代码不用于生产，目的是拆解 LangGraph 的核心机制，让你能在面试中讲清楚原理，而不只是"用了一个库"。

四个核心概念

概念	类比	职责
State	数据库表的一行	保存整个任务的当前状态
Node	函数	读取状态、执行操作、返回状态更新
Edge	路由规则	决定当前 Node 执行完后走哪个 Node
Reducer	Redux reducer	把 Node 的输出合并到当前 State

最小实现

// packages/agent-runtime/src/state-graph.ts

// State 类型：任务的全局状态
export type StateValue = unknown
export type State = Record<string, StateValue>

// Node：接收当前 State，返回部分状态更新
export type NodeFn<S extends State> = (state: S) => Promise<Partial<S>>

// Edge：普通边（固定跳转）或条件边（根据状态决定）
export type Edge<S extends State> =
  | string                             // 固定跳转到某个 Node
  | ((state: S) => string | '__end__') // 条件边

// Reducer：合并 Node 输出到 State（默认浅合并）
export type Reducer<S extends State> = (current: S, update: Partial<S>) => S

const defaultReducer = <S extends State>(current: S, update: Partial<S>): S => ({
  ...current,
  ...update,
})

export class StateGraph<S extends State> {
  private nodes = new Map<string, NodeFn<S>>()
  private edges = new Map<string, Edge<S>>()
  private reducer: Reducer<S>

  constructor(options?: { reducer?: Reducer<S> }) {
    this.reducer = options?.reducer ?? defaultReducer
  }

  addNode(name: string, fn: NodeFn<S>): this {
    this.nodes.set(name, fn)
    return this
  }

  addEdge(from: string, to: Edge<S>): this {
    this.edges.set(from, to)
    return this
  }

  // 执行图，从 startNode 开始
  async invoke(initialState: S, startNode: string): Promise<S> {
    let state = { ...initialState }
    let currentNode: string | '__end__' = startNode

    while (currentNode !== '__end__') {
      const nodeFn = this.nodes.get(currentNode)
      if (!nodeFn) throw new Error(`节点 "${currentNode}" 不存在`)

      // 执行 Node，获取状态更新
      const update = await nodeFn(state)

      // Reducer 合并更新
      state = this.reducer(state, update)

      // 决定下一个 Node
      const edge = this.edges.get(currentNode)
      if (!edge) {
        currentNode = '__end__'
      } else if (typeof edge === 'string') {
        currentNode = edge
      } else {
        currentNode = edge(state)
      }
    }

    return state
  }
}

用最小 StateGraph 实现知识库 Agent

// packages/agent-runtime/src/knowledge-agent.ts
import { StateGraph } from './state-graph'
import { generateText } from 'ai'

interface AgentState {
  question: string
  rewrittenQuery?: string
  retrievedChunks?: Array<{ id: string; content: string }>
  answer?: string
  needsRetrieval: boolean
}

const graph = new StateGraph<AgentState>()

// Node 1: Query Rewrite
graph.addNode('rewrite', async (state) => {
  const { text } = await generateText({
    model: getModel(),
    prompt: `将以下问题改写为更适合语义检索的形式：\n${state.question}`,
  })
  return { rewrittenQuery: text }
})

// Node 2: 路由决定
graph.addNode('router', async (state) => {
  // 简单规则：包含"你好"等闲聊词就不需要检索
  const isChitchat = /^(你好|谢谢|再见)/.test(state.question)
  return { needsRetrieval: !isChitchat }
})

// Node 3: 检索
graph.addNode('retrieve', async (state) => {
  const chunks = await retrieve(state.rewrittenQuery ?? state.question)
  return { retrievedChunks: chunks }
})

// Node 4: 生成答案
graph.addNode('generate', async (state) => {
  const context = state.retrievedChunks
    ?.map((c) => c.content)
    .join('\n\n') ?? ''
  const { text } = await generateText({
    model: getModel(),
    system: '你是知识库问答助手，根据提供的文档回答问题，没有依据就说不知道。',
    prompt: context ? `文档：\n${context}\n\n问题：${state.question}` : state.question,
  })
  return { answer: text }
})

// 边定义
graph.addEdge('router', (state) => state.needsRetrieval ? 'rewrite' : 'generate')
graph.addEdge('rewrite', 'retrieve')
graph.addEdge('retrieve', 'generate')
graph.addEdge('generate', '__end__')

// 导出执行函数
export async function runKnowledgeAgent(question: string) {
  const finalState = await graph.invoke(
    { question, needsRetrieval: true },
    'router',
  )
  return finalState.answer!
}

Reducer 的作用

默认 Reducer 是浅合并（Object.assign）。有时候需要自定义 Reducer，比如 messages 字段应该追加而不是覆盖：

const agentReducer = (current: AgentState, update: Partial<AgentState>) => ({
  ...current,
  ...update,
  // messages 字段追加而不是覆盖
  messages: [
    ...(current.messages ?? []),
    ...(update.messages ?? []),
  ],
})

const graph = new StateGraph<AgentState>({ reducer: agentReducer })

这正是 LangGraph 的 Annotated 机制背后在做的事。

本节产物

packages/agent-runtime/src/
  state-graph.ts          # 最小 StateGraph 实现
  knowledge-agent.ts      # 基于 StateGraph 的知识库 Agent

面试追问

LangGraph 的核心实现原理是什么？

StateGraph 本质上是一个状态机执行引擎：维护一个全局 State 对象，每次执行把当前 State 传给当前 Node，Node 返回 Partial State 更新，Reducer 把更新合并到 State，然后根据 Edge 规则决定下一个 Node。整个流程就是一个 while (currentNode !== '__end__') 的循环，直到没有后续节点为止。Checkpoint 在每次 Reducer 合并后持久化 State，支持断点恢复。

Reducer 在 LangGraph 里解决什么问题？

Agent 的多个 Node 可能都要更新同一个字段（比如 messages），如果简单覆盖会丢失历史。Reducer 定义了"如何合并更新"——messages 字段用 append，其他字段用 replace。这让每个 Node 只需要关心自己要修改什么，不需要知道其他 Node 的状态。