@cloudbase/agent-adapter-langgraph

@cloudbase/agent-adapter-langgraph

将 LangGraph 工作流转换为符合 AG-UI 协议 的 Agent。

安装

npm install @cloudbase/agent-adapter-langgraph

功能

• LanggraphAgent：将编译后的 LangGraph 工作流包装为 AG-UI 兼容的 Agent
• ClientStateAnnotation：工作流状态定义，包含 messages 和 client tools

快速开始

配合 @cloudbase/agent-server 使用

import { run } from "@cloudbase/agent-server";
import { StateGraph, START, END } from "@langchain/langgraph";
import { ClientStateAnnotation, LanggraphAgent } from "@cloudbase/agent-adapter-langgraph";

// 创建 LangGraph 工作流
const workflow = new StateGraph(ClientStateAnnotation)
  .addNode("chat_node", chatNode)
  .addEdge(START, "chat_node")
  .addEdge("chat_node", END);

const compiledWorkflow = workflow.compile();

// 部署为 HTTP 服务
run({
  createAgent: () => ({
    agent: new LanggraphAgent({ compiledWorkflow }),
  }),
  port: 9000,
});

API 参考

LanggraphAgent

将编译后的 LangGraph 工作流转换为 AG-UI 兼容的 Agent。

type LanggraphAgentConfig = AgentConfig & {
  compiledWorkflow: CompiledStateGraph; // 编译后的 LangGraph 工作流
  logger?: Logger;                      // 可选，日志实例
};

const agent = new LanggraphAgent(config);

• AgentConfig：来自 AG-UI 协议
• Logger：日志接口，详见 @cloudbase/agent-server 文档

ClientStateAnnotation

创建 LangGraph 工作流时使用的状态定义，已包含 AG-UI 需要的字段：

• messages：消息历史
• client.tools：客户端传来的工具列表

const workflow = new StateGraph(ClientStateAnnotation)
  .addNode("chat_node", chatNode)
  // ...

使用客户端工具

AG-UI 支持客户端工具（Client Tools）：客户端定义工具，Agent 调用后由客户端执行并返回结果。

适用场景：

• 需要访问客户端 API（如获取地理位置、访问剪贴板）
• 需要用户确认的操作（如发送邮件前确认）
• 需要展示 UI 交互（如让用户选择文件）

在 chatNode 中使用客户端工具

import { ClientState } from "@cloudbase/agent-adapter-langgraph";

async function chatNode(state: ClientState) {
  const model = new ChatOpenAI({ model: "gpt-4o" });

  // 合并服务端工具和客户端工具
  const modelWithTools = model.bindTools([
    ...serverTools,              // 服务端定义的工具
    ...(state.client?.tools || []) // 客户端传来的工具
  ]);

  const response = await modelWithTools.invoke([...state.messages]);
  return { messages: [response] };
}

区分服务端工具和客户端工具

当 Agent 调用工具时，需要判断是服务端执行还是交给客户端：

const serverToolNames = new Set(serverTools.map(t => t.name));

function shouldContinue(state: ClientState): "tools" | "end" {
  const lastMessage = state.messages.at(-1) as AIMessage;

  if (lastMessage.tool_calls?.length) {
    // 如果是服务端工具，继续执行
    const hasServerTool = lastMessage.tool_calls.some(
      tc => serverToolNames.has(tc.name)
    );
    if (hasServerTool) return "tools";
  }

  // 客户端工具或无工具调用，结束并返回给客户端
  return "end";
}

CloudBaseSaver

CloudBaseSaver 是基于腾讯云开发（CloudBase）的 LangGraph 检查点存储实现，支持多租户隔离。

安装依赖

npm install @cloudbase/node-sdk

基本用法

import { CloudBaseSaver } from "@cloudbase/agent-adapter-langgraph";
import cloudbase from "@cloudbase/node-sdk";

// 初始化 CloudBase
const app = cloudbase.init({
  env: process.env.CLOUDBASE_ENV_ID,
  secretId: process.env.CLOUDBASE_SECRET_ID,
  secretKey: process.env.CLOUDBASE_SECRET_KEY,
});

const db = app.database();

// 创建检查点存储
const checkpointer = new CloudBaseSaver({
  db,
  userId: "user-123",  // 用于多租户隔离
  agentId: "my-agent",                       // 可选，用于区分不同 Agent，默认 "default"
  checkpointsCollection: "checkpoints",      // 可选，默认 "checkpoints"
  writesCollection: "checkpoint_writes",     // 可选，默认 "checkpoint_writes"
});

// 首次使用前，调用 setup() 创建数据库集合（可选，也可手动在控制台创建）
await checkpointer.setup();

// 在 LangGraph 工作流中使用
const compiledWorkflow = workflow.compile({ checkpointer });

配置选项

参数	类型	必填	默认值	说明
db	CloudBaseDb	✅	-	CloudBase 数据库实例，通过 app.database() 获取
userId	string	✅	-	用户 ID，用于多租户数据隔离
agentId	string	❌	"default"	Agent ID，用于区分同一用户的不同 Agent/工作流
checkpointsCollection	string	❌	"checkpoints"	存储检查点的集合名称
writesCollection	string	❌	"checkpoint_writes"	存储待处理写入的集合名称

setup() 方法

setup() 方法用于自动创建所需的数据库集合。该方法是幂等的，可以安全地多次调用：

// 在应用启动时调用一次即可
await checkpointer.setup();

如果集合已存在，setup() 会静默跳过，不会抛出错误。你也可以选择在 CloudBase 控制台手动创建集合。

获取 userId

在实际应用中，userId 通常从请求的认证信息中获取。一种常见方式是解析 JWT token 的 sub claim：

function getUserIdFromRequest(request: Request): string {
  const authHeader = request.headers.get("Authorization");
  if (!authHeader?.startsWith("Bearer ")) {
    throw new Error("Missing Authorization header");
  }

  const token = authHeader.slice(7);
  const parts = token.split(".");
  if (parts.length !== 3) {
    throw new Error("Invalid JWT format");
  }

  // 解码 payload（base64url）
  const payload = parts[1]!;
  const decoded = atob(payload.replace(/-/g, "+").replace(/_/g, "/"));
  const claims = JSON.parse(decoded);

  if (typeof claims.sub !== "string" || !claims.sub) {
    throw new Error("Invalid JWT: missing 'sub' claim");
  }

  return claims.sub;
}

// 在 createAgent 中使用
createExpressRoutes({
  createAgent: async (context) => {
    const userId = getUserIdFromRequest(context.request);
    const checkpointer = new CloudBaseSaver({ db, userId });
    // ...
  },
  express: app,
});

注意：上述 JWT 解析仅做 base64 解码，未验证签名。生产环境建议使用 jose 等库进行完整的 JWT 验证。

创建数据库集合

在 CloudBase 控制台创建以下集合：

• checkpoints（或自定义名称）
• checkpoint_writes（或自定义名称）

环境变量

CLOUDBASE_ENV_ID=your-env-id
CLOUDBASE_SECRET_ID=your-secret-id
CLOUDBASE_SECRET_KEY=your-secret-key

依赖

• @langchain/langgraph：LangGraph 框架
• @langchain/core：LangChain 核心工具
• @cloudbase/node-sdk：腾讯云开发 SDK（peerDependency，使用 CloudBaseSaver 时需安装）

文档

📚 完整文档请参阅 云开发 Agent 开发指南

相关资源

• AG-UI 协议
• LangGraph 文档
• LangGraph Checkpointer 概念 - 了解 checkpointer 的更多用法（如 getTuple、list、deleteThread 等）