LangGraph adapter for AG-Kit agents
将 LangGraph 工作流转换为符合 AG-UI 协议 的 Agent。
``bash`
npm install @cloudbase/agent-adapter-langgraph
- LanggraphAgent:将编译后的 LangGraph 工作流包装为 AG-UI 兼容的 Agent
- ClientStateAnnotation:工作流状态定义,包含 messages 和 client tools
`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,
});
`
将编译后的 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 文档
创建 LangGraph 工作流时使用的状态定义,已包含 AG-UI 需要的字段:
- messages:消息历史client.tools
- :客户端传来的工具列表
`typescript`
const workflow = new StateGraph(ClientStateAnnotation)
.addNode("chat_node", chatNode)
// ...
AG-UI 支持客户端工具(Client Tools):客户端定义工具,Agent 调用后由客户端执行并返回结果。
适用场景:
- 需要访问客户端 API(如获取地理位置、访问剪贴板)
- 需要用户确认的操作(如发送邮件前确认)
- 需要展示 UI 交互(如让用户选择文件)
`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] };
}
`
当 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 是基于腾讯云开发(CloudBase)的 LangGraph 检查点存储实现,支持多租户隔离。
`bash`
npm install @cloudbase/node-sdk
`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 });
`
| 参数 | 类型 | 必填 | 默认值 | 说明 |
|------|------|------|--------|------|
| db | CloudBaseDb | ✅ | - | CloudBase 数据库实例,通过 app.database() 获取 |userId
| | string | ✅ | - | 用户 ID,用于多租户数据隔离 |agentId
| | string | ❌ | "default" | Agent ID,用于区分同一用户的不同 Agent/工作流 |checkpointsCollection
| | string | ❌ | "checkpoints" | 存储检查点的集合名称 |writesCollection
| | string | ❌ | "checkpoint_writes" | 存储待处理写入的集合名称 |
setup() 方法用于自动创建所需的数据库集合。该方法是幂等的,可以安全地多次调用:
`typescript`
// 在应用启动时调用一次即可
await checkpointer.setup();
如果集合已存在,setup() 会静默跳过,不会抛出错误。你也可以选择在 CloudBase 控制台手动创建集合。
在实际应用中,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 验证。
在 CloudBase 控制台创建以下集合:
- checkpoints(或自定义名称)checkpoint_writes
- (或自定义名称)
`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` 等)