用 LangGraph 搭建 Claude Code 风格的代码智能体：ReAct、SubAgent、Todo 与上下文压缩

import { Annotation, END, START, StateGraph } from "@langchain/langgraph";
import { ToolNode } from "@langchain/langgraph/prebuilt";
import type { BaseMessage } from "@langchain/core/messages";

const agentState = Annotation.Root({
  messages: Annotation<BaseMessage[]>({
    reducer: safeMessagesStateReducer,
    default: () => [],
  }),
});

const toolNode = new ToolNode(tools);

const shouldContinue = (state: typeof agentState.State) => {
  const { messages } = state;
  const lastMessage = messages[messages.length - 1];

  if (
    "tool_calls" in lastMessage &&
    Array.isArray(lastMessage.tool_calls) &&
    lastMessage.tool_calls.length > 0
  ) {
    return "tools";
  }

  return END;
};

const callModel = async (state: typeof agentState.State) => {
  const response = await modelWithTools.invoke(state.messages);

  return {
    messages: [response],
  };
};

const workflow = new StateGraph(agentState)
  .addNode("llm", callModel)
  .addNode("tools", toolNode)
  .addEdge(START, "llm")
  .addConditionalEdges("llm", shouldContinue, ["tools", END])
  .addEdge("tools", "llm");

const agent = workflow.compile();

const config = {
  configurable: {
    thread_id: sessionId,
  },
  streamMode: ["values", "messages", "updates"],
  version: "v2" as const,
};

const stream = agent.streamEvents(
  {
    messages: [inputMessage],
  },
  config,
);

for await (const event of stream) {
  // 根据 event 类型推送给前端
}

import { Command, interrupt } from "@langchain/langgraph";
import { HumanMessage } from "@langchain/core/messages";

const humanReviewNode = async (state: typeof agentState.State) => {
  const lastMessage = state.messages[state.messages.length - 1];

  const answer = interrupt({
    question: "确认是否执行该工具调用？",
    toolCall: lastMessage,
    options: ["approve", "reject", "modify"],
  });

  if (answer.type === "approve") {
    return new Command({
      goto: "tools",
      update: {
        messages: [new HumanMessage("用户已确认执行该工具调用。")],
      },
    });
  }

  if (answer.type === "reject") {
    return new Command({
      goto: "llm",
      update: {
        messages: [
          new HumanMessage(
            `用户拒绝执行该工具调用，原因：${answer.reason ?? "未提供"}`,
          ),
        ],
      },
    });
  }

  return new Command({
    goto: "llm",
    update: {
      messages: [
        new HumanMessage(`用户修改了要求：${answer.content}`),
      ],
    },
  });
};

const workflowWithReview = new StateGraph(agentState)
  .addNode("llm", callModel)
  .addNode("human_review", humanReviewNode)
  .addNode("tools", toolNode)
  .addEdge(START, "llm")
  .addConditionalEdges("llm", shouldContinueWithReview, [
    "human_review",
    END,
  ])
  .addEdge("human_review", "tools")
  .addEdge("tools", "llm");

import { tool } from "@langchain/core/tools";
import { ToolMessage } from "@langchain/core/messages";
import { getCurrentTaskInput, Command } from "@langchain/langgraph";
import { z } from "zod";

function createAskHumanTool() {
  const executor = async (
    args: { question: string },
    config: any,
  ) => {
    const state = getCurrentTaskInput() as typeof agentState.State;

    const toolResult = new ToolMessage({
      content: `已发起人工询问：${args.question}`,
      name: "askHuman",
      tool_call_id: config.toolCall.id,
    });

    return new Command({
      graph: Command.PARENT,
      goto: "askHuman",
      update: {
        messages: state.messages.concat(toolResult),
      },
    });
  };

  return tool(executor, {
    name: "askHuman",
    description:
      "当需求不明确、需要用户确认方案、需要用户补充信息时，调用该工具向用户提问。",
    schema: z.object({
      question: z.string().describe("需要询问用户的问题"),
    }),
  });
}

import { MemorySaver } from "@langchain/langgraph";

const app = workflowWithReview.compile({
  checkpointer: new MemorySaver(),
});

const config = {
  configurable: {
    thread_id: "review-session-1",
  },
};

await app.invoke(
  {
    messages: [inputMessage],
  },
  config,
);

// 用户确认后继续
await app.invoke(null, config);

interface SubAgentConfig {
  type: "general-purpose" | "code-analyzer" | "document-writer";
  systemPrompt: string;
  allowedTools: string[] | null;
}

const subAgentConfigs: SubAgentConfig[] = [
  {
    type: "general-purpose",
    systemPrompt: `
你是一个通用任务 Agent，适合处理复杂搜索、内容分析和多步骤任务。
处理问题时要拆解步骤，给出清晰、可验证的结论。
    `.trim(),
    allowedTools: null,
  },
  {
    type: "code-analyzer",
    systemPrompt: `
你是代码分析 Agent，重点关注：
- 代码质量问题
- 架构设计问题
- 性能瓶颈
- 安全风险
输出必须具体到文件、函数或代码片段，并给出可执行修改建议。
    `.trim(),
    allowedTools: ["Read", "Grep", "Glob", "LS"],
  },
  {
    type: "document-writer",
    systemPrompt: `
你是技术文档 Agent，重点生成：
- API 文档
- 使用说明
- 项目结构说明
- 面向开发者的教程
文档要结构清晰，并尽量包含示例。
    `.trim(),
    allowedTools: ["Read", "Write", "Edit"],
  },
];

import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { tool } from "@langchain/core/tools";
import { z } from "zod";

function createTaskTool(
  baseTools: any[],
  model: any,
  subAgentConfigs: SubAgentConfig[],
) {
  const agentInstances = new Map<string, any>();

  for (const config of subAgentConfigs) {
    const filteredTools = config.allowedTools
      ? baseTools.filter((tool) => config.allowedTools!.includes(tool.name))
      : baseTools.filter((tool) => tool.name !== "TaskTool");

    agentInstances.set(
      config.type,
      createReactAgent({
        llm: model,
        tools: filteredTools,
        systemMessage: config.systemPrompt,
      }),
    );
  }

  return tool(
    async (args: { description: string; subagent_type: string }) => {
      const { description, subagent_type } = args;

      const agent = agentInstances.get(subagent_type);
      if (!agent) {
        throw new Error(`未知 SubAgent 类型：${subagent_type}`);
      }

      const subAgentState = {
        messages: [
          {
            role: "user",
            content: description,
          },
        ],
      };

      const result = await agent.invoke(subAgentState);
      const finalMessage = result.messages[result.messages.length - 1];

      return `
SubAgent 执行完成。

类型：${subagent_type}

任务：
${description}

结果：
${finalMessage.content}

请主 Agent 根据该结果提取关键信息，并面向用户给出最终回答。
      `.trim();
    },
    {
      name: "TaskTool",
      description: `
启动专门的 SubAgent 来处理复杂多步骤任务。

可用类型：
- general-purpose：通用任务，适合搜索、分析和多步骤执行
- code-analyzer：代码分析，适合架构、质量、性能、安全问题定位
- document-writer：文档编写，适合 API 文档、使用说明和项目文档

使用规则：
- 复杂任务、跨文件分析、大范围搜索时优先使用
- 每次调用都是独立执行环境
- 任务描述必须具体，包含目标、范围和输出要求
- SubAgent 完成后，由主 Agent 负责总结给用户
      `.trim(),
      schema: z.object({
        description: z.string().describe("给 SubAgent 的详细任务描述"),
        subagent_type: z
          .enum(["general-purpose", "code-analyzer", "document-writer"])
          .describe("要使用的 SubAgent 类型"),
      }),
    },
  );
}

const safeConcurrencyTools = [
  taskTool,
  readFileTool,
  grepTool,
  globTool,
  lsTool,
];

const unsafeConcurrencyTools = [
  writeFileTool,
  editFileTool,
  bashTool,
];

const safeConcurrencyToolNode = new ToolNode(safeConcurrencyTools, {
  maxConcurrency: 10,
  name: "safeConcurrencyTools",
});

const unsafeConcurrencyToolNode = new ToolNode(unsafeConcurrencyTools, {
  maxConcurrency: 1,
  name: "unsafeConcurrencyTools",
});

function getToolCalls(message: any) {
  return Array.isArray(message.tool_calls) ? message.tool_calls : [];
}

function isSafeConcurrencyTool(name: string) {
  return safeConcurrencyTools.some((tool) => tool.name === name);
}

const shouldContinue = (state: typeof agentState.State) => {
  const { messages } = state;
  const lastMessage = messages[messages.length - 1];
  const toolCalls = getToolCalls(lastMessage);

  if (toolCalls.length === 0) {
    return END;
  }

  const allSafe = toolCalls.every((call) => isSafeConcurrencyTool(call.name));

  return allSafe ? "safeConcurrencyTools" : "unsafeConcurrencyTools";
};

const graph = new StateGraph(agentState)
  .addNode("llm", callModel)
  .addNode("safeConcurrencyTools", safeConcurrencyToolNode)
  .addNode("unsafeConcurrencyTools", unsafeConcurrencyToolNode)
  .addEdge(START, "llm")
  .addConditionalEdges("llm", shouldContinue, [
    "safeConcurrencyTools",
    "unsafeConcurrencyTools",
    END,
  ])
  .addEdge("safeConcurrencyTools", "llm")
  .addEdge("unsafeConcurrencyTools", "llm");

export enum TaskStatus {
  Pending = "pending",
  InProgress = "in_progress",
  Completed = "completed",
  Failed = "failed",
  Blocked = "blocked",
}

export interface TodoItem {
  id: string;
  name: string;
  desc: string;
  status: TaskStatus;
  priority?: "high" | "medium" | "low";
  startTime?: string;
  endTime?: string;
  error?: string;
}

const agentState = Annotation.Root({
  messages: Annotation<BaseMessage[]>({
    reducer: safeMessagesStateReducer,
    default: () => [],
  }),

  todoList: Annotation<TodoItem[]>({
    reducer: (_oldValue, newValue) => newValue,
    default: () => [],
  }),
});

import { Command, getCurrentTaskInput } from "@langchain/langgraph";
import { ToolMessage } from "@langchain/core/messages";
import { tool } from "@langchain/core/tools";
import { z } from "zod";

export function createTodoWriteTool() {
  const executor = async (
    args: { todoList: TodoItem[] },
    config: any,
  ) => {
    const state = getCurrentTaskInput() as typeof agentState.State;

    const normalizedTodoList = args.todoList.map((item) => ({
      ...item,
      id: item.id || crypto.randomUUID(),
    }));

    const responseMsg = new ToolMessage({
      content: `任务列表已更新，共 ${normalizedTodoList.length} 个任务。`,
      name: "TodoWrite",
      tool_call_id: config.toolCall.id,
    });

    return new Command({
      update: {
        todoList: normalizedTodoList,
        messages: state.messages.concat(responseMsg),
      },
    });
  };

  return tool(executor, {
    name: "TodoWrite",
    description: `
创建和更新当前会话的任务列表。

适合使用：
- 用户请求包含 3 个以上步骤
- 任务需要跨文件修改或多轮操作
- 用户明确要求制定计划或跟踪进度
- 执行过程中发现新的后续任务
- 开始任务前需要标记 in_progress
- 完成任务后需要标记 completed

不适合使用：
- 单步、简单、可立即完成的请求
- 纯解释型问题
- 无需执行动作的信息查询

规则：
- 同一时间通常只保持一个任务为 in_progress
- 只有确认完成后才能标记 completed
- 遇到阻塞要标记 blocked，并说明原因
- 执行失败要标记 failed，并写入 error
    `.trim(),
    schema: z.object({
      todoList: z.array(
        z.object({
          id: z.string().describe("任务唯一标识"),
          name: z.string().describe("任务名称"),
          desc: z.string().describe("任务描述"),
          status: z.nativeEnum(TaskStatus).describe("任务状态"),
          priority: z.enum(["high", "medium", "low"]).optional(),
          startTime: z.string().optional(),
          endTime: z.string().optional(),
          error: z.string().optional(),
        }),
      ),
    }),
  });
}

export function createTodoReadTool() {
  const executor = async (_args: {}, config: any) => {
    const state = getCurrentTaskInput() as typeof agentState.State;
    const currentTasks = state.todoList ?? [];

    return new ToolMessage({
      content: `
请继续使用 TodoRead 和 TodoWrite 跟踪任务进度。

当前任务列表：
${JSON.stringify(currentTasks, null, 2)}
      `.trim(),
      name: "TodoRead",
      tool_call_id: config.toolCall.id,
    });
  };

  return tool(executor, {
    name: "TodoRead",
    description:
      "读取当前会话的任务列表。开始复杂任务、完成任务、遇到阻塞或不确定下一步时，应主动读取任务状态。",
    schema: z.object({}),
  });
}

const todoPrompt = `
你具备任务管理能力，需要遵守这些规则：

1. 对复杂请求使用 TodoWrite 创建任务列表：
   - 任务包含 3 个以上步骤
   - 需要跨文件搜索、修改、验证
   - 用户给出多个目标
   - 用户明确要求制定计划

2. 执行前使用 TodoRead 查看当前任务状态。

3. 开始处理某个任务前，将它标记为 in_progress。

4. 完成任务后，将它标记为 completed。

5. 遇到无法继续的情况，将任务标记为 blocked，并说明阻塞原因。

6. 执行失败时，将任务标记为 failed，并写明错误信息。

7. 不要为简单单步问题创建任务列表。
`.trim();

const systemPrompt = `
你是一个代码智能体，可以阅读文件、搜索代码、修改文件、执行安全命令，并通过 Todo 工具管理复杂任务。

${todoPrompt}
`.trim();

interface CompressionRecord {
  beforeTokens: number;
  afterTokens: number;
  createdAt: string;
  summary: string;
}

const agentStateWithCompression = Annotation.Root({
  messages: Annotation<BaseMessage[]>({
    reducer: safeMessagesStateReducer,
    default: () => [],
  }),

  todoList: Annotation<TodoItem[]>({
    reducer: (_oldValue, newValue) => newValue,
    default: () => [],
  }),

  compressionHistory: Annotation<CompressionRecord[]>({
    reducer: (oldValue, newValue) => [
      ...(oldValue ?? []),
      ...(newValue ?? []),
    ],
    default: () => [],
  }),
});

const workflow = new StateGraph(agentStateWithCompression)
  .addNode("compression", compressionNode)
  .addNode("llm", callModel)
  .addNode("tools", toolNode)
  .addEdge(START, "compression")
  .addEdge("compression", "llm")
  .addConditionalEdges("llm", shouldContinue, ["tools", END])
  .addEdge("tools", "compression");

import { AIMessage, BaseMessage } from "@langchain/core/messages";

function isAIMessage(msg: BaseMessage): msg is AIMessage {
  return msg.getType() === "ai";
}

export function getLatestTokenUsage(messages: BaseMessage[]): number {
  for (let i = messages.length - 1; i >= 0; i--) {
    const msg = messages[i];

    if (isAIMessage(msg) && msg.response_metadata?.usage) {
      const usage = msg.response_metadata.usage;

      return (
        (usage.total_tokens ?? 0) +
        (usage.cache_creation_tokens ?? 0) +
        (usage.cache_read_tokens ?? 0)
      );
    }
  }

  return estimateTokens(messages);
}

function needsCompress(params: {
  usedTokens: number;
  maxContextTokens: number;
  reservedOutputTokens: number;
  thresholdRatio: number;
}) {
  const usableInputTokens =
    params.maxContextTokens - params.reservedOutputTokens;

  return params.usedTokens >= usableInputTokens * params.thresholdRatio;
}

const shouldCompress = needsCompress({
  usedTokens,
  maxContextTokens: 200_000,
  reservedOutputTokens: 16_000,
  thresholdRatio: 0.92,
});

import { AIMessage, RemoveMessage } from "@langchain/core/messages";

async function compressionNode(
  state: typeof agentStateWithCompression.State,
) {
  const usedTokens = getLatestTokenUsage(state.messages);

  const shouldCompress = needsCompress({
    usedTokens,
    maxContextTokens: 200_000,
    reservedOutputTokens: 16_000,
    thresholdRatio: 0.92,
  });

  if (!shouldCompress) {
    return {};
  }

  const compressionPrompt = generateCompressionPrompt();

  const compressedSummary = await compressionModel.invoke([
    {
      role: "system",
      content: compressionPrompt,
    },
    ...state.messages,
  ]);

  const messagesToRemove = selectMessagesToRemove(state.messages);

  const summaryMessage = new AIMessage({
    content: `
以下是此前对话和工作的压缩摘要。后续继续执行任务时，必须把它当作已有上下文。

${compressedSummary.content}
    `.trim(),
  });

  return {
    messages: [
      ...messagesToRemove.map((message) => new RemoveMessage({ id: message.id! })),
      summaryMessage,
    ],
    compressionHistory: [
      {
        beforeTokens: usedTokens,
        afterTokens: estimateTokens([summaryMessage]),
        createdAt: new Date().toISOString(),
        summary: String(compressedSummary.content),
      },
    ],
  };
}

你的任务是为当前代码智能体会话生成详细摘要，用于在压缩上下文后继续开发工作。

摘要必须保留技术细节、文件路径、代码结构、架构决策、用户明确要求和仍未完成的任务。

生成最终摘要前，请先使用 <analysis> 标签梳理信息，检查是否遗漏关键内容。

最终摘要必须包含：

1. 主要请求和意图
2. 关键技术概念
3. 文件和代码段
4. 错误和修复
5. 已解决问题
6. 用户消息
7. 待处理任务
8. 当前工作
9. 可选下一步

不要泛泛而谈。涉及代码时保留必要代码片段、函数名、文件名和修改原因。

const config = {
  configurable: {
    thread_id: sessionId,
  },
  streamMode: ["values", "messages", "updates"] as any,
  version: "v2" as const,
  signal: abortSignal,
};

const stream = agent.streamEvents(
  {
    messages: [inputMessage],
  },
  config,
);

for await (const event of stream) {
  sendToClient(event);
}

const abortController = new AbortController();

const response = await fetch("/api/agent", {
  method: "POST",
  body: JSON.stringify({
    sessionId,
    message,
  }),
  headers: {
    "Content-Type": "application/json",
  },
  signal: abortController.signal,
});

// 用户输入新指令时
abortController.abort();

await agent.invoke(
  {
    messages: [
      {
        role: "user",
        content: newUserMessage,
      },
    ],
  },
  {
    configurable: {
      thread_id: sessionId,
    },
  },
);

const filteredTools = baseTools.filter((tool) => tool.name !== "TaskTool");