@cloudbase/agent-adapter-langgraph

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

安装

``bash
npm install @cloudbase/agent-adapter-langgraph
`

功能

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

快速开始

$3

`typescript
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 参考

$3

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

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

const agent = new LanggraphAgent(config);
`

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

$3

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

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

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

使用客户端工具

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

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

$3

`typescript
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] };
}
`

$3

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

`typescript
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 检查点存储实现，支持多租户隔离。

$3

`bash
npm install @cloudbase/node-sdk
`

$3

`typescript
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 });
`

$3

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

$3

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

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

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

$3

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

`typescript
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 验证。

$3

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

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

$3

`bash
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` 等）