基于 LangGraph 一步步实现 Claude

• 实现偏底层，上层无过多抽象，用户可精准控制大模型的输入，这对llm上下文优化至关重要。
• 规则驱动和llm驱动相融合，可以在大模型的自主性和不确定性中间寻找平衡，让我们的系统具备较高的可靠性、可观测性。

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


const toolNodeForGraph = new ToolNode(tools)


const shouldContinue = (state: typeof agentState.State) => {
  const { messages } = state;
  const lastMessage = messages;
  if ("tool_calls" in lastMessage && Array.isArray(lastMessage.tool_calls) && lastMessage.tool_calls?.length) {
      return"tools";
  }
  return END;
}


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




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


const agent = workflow.compile()

const config = {
  configurable: { thread_id: this.sessionId },
  streamMode: ,
  version: 'v2' as const,
}
const stream = agent.streamEvents({messages: }, config);

// 添加人工审查节点的workflow
const workflowWithReview = new StateGraph(MessagesAnnotation)
  .addNode("agent", callModel)
  .addNode("tools", toolNodeForGraph)
  .addNode("human_review", humanReviewNode)  // 新增：人工审查节点
  .addEdge(START, "agent")
  .addConditionalEdges("agent", shouldContinueWithReview, )  // 修改：路由到审查节点
  .addConditionalEdges("human_review", checkUserApproval, )  // 新增：审查结果路由
  .addEdge("tools", "agent");

2.1 人工审查节点实现

// 人工审查节点的具体实现
const humanReviewNode = async (state: typeof MessagesAnnotation.State) => {
  const lastMessage = state.messages;
  const humanAnswer = interrupt("确认是否执行");
  if(humanAnswer === "同意"){
    returnnew Command({
      goto: "tool",
      update: {
        messages: 
      }
    })
  }
  
  if(humanAnswer === "拒绝"){
    returnnew Command({
      goto: "agent",
      update: {
        messages: 
      }
    })
  }
};

2.2 由大模型驱动决策是否需要人机协同

function createHumanLoopTool(){
    const executor = (arg, config) => {
      const msgs: BaseMessage = ;
      const state = getCurrentTaskInput();
      returnnew Command({
        goto: 'askHuman',
        graph: Command.PARENT,
        update: {
          messages: state.messages.concat(msgs),
        },
      });
    };
    return tool(executor, {
      name: 'ashHuman',
      description: '当需要向用户确认需求、询问补充信息、向用户提问时, 务必调用此工具',
      schema,
    });
  }

2.3 检查点持久化支持

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


const app = workflow.compile({
  checkpointer: new MemorySaver(),  // 状态持久化
});


// 使用示例
const config = { configurable: { thread_id: "review-session-1" } };
await app.invoke(initialState, config);


// 在用户审查后继续执行
await app.invoke(null, config);  // 从中断点继续

// Task工具对象结构 (p_2)
p_2 = {
    name: "Task",
    async call({ prompt }, context, globalConfig, parentMessage) {
        // 核心：通过Task工具内部创建和管理SubAgent
        if (config.parallelTasksCount > 1) {
            // 多Agent并发执行模式
            const agentTasks = Array(config.parallelTasksCount)
                .fill(`${prompt}

Provide a thorough and complete analysis.`)
                .map((taskPrompt, index) => I2A(taskPrompt, index, executionContext, parentMessage, globalConfig));
        } else {
            // 单Agent执行模式
        }
    }
}

// SubAgent启动函数
async function* I2A(taskPrompt, agentIndex, parentContext, globalConfig, options = {}){
    // 1. 生成唯一Agent ID
    const agentId = VN5();
    
    // 2. 创建隔离的执行上下文
    const executionContext = {
        abortController: parentContext.abortController,
        options: { ...parentContext.options },
        getToolPermissionContext: parentContext.getToolPermissionContext,
        readFileState: parentContext.readFileState,
        setInProgressToolUseIDs: parentContext.setInProgressToolUseIDs,
        tools: parentContext.tools.filter(tool => tool.name !== "Task") // 防止递归
    };
    
    // 3. 执行主Agent循环
    forawait(let agentResponse of nO(/* 主Agent循环 */)){
        // 处理Agent响应和工具调用
    }
    
    // 4. 返回最终结果
    yield { type: "result", data: { /* 结果数据 */ } };
}

// 并发执行调度器
async function* UH1(generators, maxConcurrency = Infinity){
    const wrapGenerator = (generator) => {
        const promise = generator.next().then(({ done, value }) => ({
            done, value, generator, promise
        }));
        return promise;
    };
    
    const remainingGenerators = ;
    const activePromises = new Set();
    
    // 启动初始并发任务
    while (activePromises.size < maxConcurrency && remainingGenerators.length > 0) {
        const generator = remainingGenerators.shift();
        activePromises.add(wrapGenerator(generator));
    }
    
    // 并发执行循环
    while (activePromises.size > 0) {
        const { done, value, generator, promise } = await Promise.race(activePromises);
        activePromises.delete(promise);
        
        if (!done) {
            activePromises.add(wrapGenerator(generator));
            if (value !== undefined) yield value;
        } elseif (remainingGenerators.length > 0) {
            const nextGenerator = remainingGenerators.shift();
            activePromises.add(wrapGenerator(nextGenerator));
        }
    }
}

// 多Agent结果合成器
function KN5(originalTask, agentResults){
    const sortedResults = agentResults.sort((a, b) => a.agentIndex - b.agentIndex);
    
    const agentResponses = sortedResults.map((result, index) => {
        const textContent = result.content
            .filter(content => content.type === "text")
            .map(content => content.text)
            .join("

");
        
        return `== AGENT ${index + 1} RESPONSE ==
${textContent}`;
    }).join("

");
    
    return `Original task: ${originalTask}


I've assigned multiple agents to tackle this task. Each agent has analyzed the problem and provided their findings.


${agentResponses}


Based on all the information provided by these agents, synthesize a comprehensive and cohesive response...`;
}

function createTaskTool(baseTools: any, model: any, subAgentConfigs: SubAgentConfig){
  return tool(
    async (args: { description: string; subagent_type: string }, config) => {
      // 具体工具内部实现，下文重点介绍
    },
    {
      name: "TaskTool",
      description: `Launch specialized SubAgents to autonomously handle complex multi-step tasks.


Available Agent types:
- general-purpose: General agent suitable forcomplex queries, file searches, and multi-step task execution (Tools: *)
- code-analyzer: Code analysis expert for code review, architecture analysis, performance optimization
- document-writer: Documentation expert for technical docs, user manuals, API documentation


Usage rules:
1. Use this tool when tasks are complexand require specialized handling
2. Each SubAgent invocation is independent and stateless  
3. Provide detailed task descriptions; SubAgents will complete autonomously
4. After SubAgent completion, summarize key information for the user


When to use SubAgents:
- Complex multi-step analysis tasks
- Tasks requiring specialized skills (code analysis, documentation)
- Large-scale file search and processing
- Independent tasks that can be processed in parallel`,
      schema: z.object({
        description: z.string().describe("Detailed task description for the SubAgent"),
        subagent_type: z.enum().describe("SubAgent type to use")
      })
    }
  );
}

function createTaskTool(baseTools: any, model: any, subAgentConfigs: SubAgentConfig){
  // 为每种类型创建专门的ReActAgent实例
  const agentInstances = new Map<string, any>();
  
  // 初始化预定义的SubAgent类型
  subAgentConfigs.forEach(config => {
    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 }, config) => {
      const { description, subagent_type } = args;
      
      // 获取指定类型的Agent
      const agent = agentInstances.get(subagent_type);
      if (!agent) {
        thrownew Error(`Unknown SubAgent type: ${subagent_type}`);
      }
      
      // 执行SubAgent
      try {
        console.log(` Launching SubAgent : ${description}`);
        
        // 创建隔离的执行上下文
        const subAgentState = {
          messages: 
        };
        
        // 执行Agent并获取结果
        const result = await agent.invoke(subAgentState);
        
        // 提取最终响应
        const finalMessage = result.messages;
        const responseContent = finalMessage.content;
        
        console.log(`✅ SubAgent  completed`);
        
        // 返回格式化结果 - 模拟claude-code的结果格式
        return `SubAgent  execution completed:


Task: ${description}


Result:
${responseContent}


Note: This result was generated by a specialized SubAgent. Please summarize key information for the user as needed.`;
        
      } catch (error) {
        console.error(`❌ SubAgent  failed:`, error);
        return `SubAgent  execution failed: ${error.message}`;
      }
    },
    {
      name: "TaskTool",
      description: `提示词 略过`,
      schema: z.object({
        description: z.string().describe("Detailed task description for the SubAgent"),
        subagentType: z.enum().describe("SubAgent type to use")
      })
    }
  );
}

// SubAgent配置定义
const subAgentConfigs: SubAgentConfig = 
  },
  {
    type: "document-writer",
    systemPrompt: `You are a technical writing expert focused on:
- Clear and accurate technical documentation
- User-friendly operation guides  
- Complete API documentation and examples
- Structured project documentation
Ensure readability and practicality of documentation`,
    allowedTools: 
  }
];