基于 LangGraph 开发

LangGraph 是 LangChain 团队开发的图结构工作流框架，提供了更精细的流程控制能力。云开发同时支持 TypeScript 和 Python 版本的 LangGraph 适配器。

前置条件

Node.js 18+ 或 Python 3.9+
已开通云开发环境
已配置大模型（见接入大模型）
已创建 API Key（获取地址）

TypeScript 版本

安装依赖

npm install @cloudbase/agent-adapter-langgraph @langchain/langgraph @langchain/openai

快速开始

import { LanggraphAgent, createAgentServer } from "@cloudbase/agent-adapter-langgraph";
import { StateGraph, Annotation, END } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";

// 定义状态
const StateAnnotation = Annotation.Root({
  messages: Annotation<string[]>({
    reducer: (x, y) => x.concat(y),
    default: () => [],
  }),
  response: Annotation<string>(),
});

// 使用云开发内置的大模型端点
const model = new ChatOpenAI({
  model: process.env.TCB_AI_MODEL || "deepseek-v4-flash",
  apiKey: process.env.TCB_API_KEY,
  configuration: {
    baseURL: `https://${process.env.TCB_ENV_ID}.api.tcloudbasegateway.com/v1/ai/cloudbase`,
  },
});

// 定义节点
async function chatNode(state: typeof StateAnnotation.State) {
  const lastMessage = state.messages[state.messages.length - 1];
  const response = await model.invoke(lastMessage);
  return { response: response.content };
}

// 构建图
const graph = new StateGraph(StateAnnotation)
  .addNode("chat", chatNode)
  .addEdge("__start__", "chat")
  .addEdge("chat", END);

// 编译图
const compiledGraph = graph.compile();

// 创建 Agent
const agent = new LanggraphAgent({
  name: "ChatBot",
  description: "基于 LangGraph 的对话助手",
  graph: compiledGraph,
});

// 导出
module.exports = createAgentServer(agent);

带工具的 Agent

import { LanggraphAgent, createAgentServer, ClientStateAnnotation } from "@cloudbase/agent-adapter-langgraph";
import { StateGraph, Annotation, END } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";
import { ToolNode } from "@langchain/langgraph/prebuilt";
import { tool } from "@langchain/core/tools";
import { z } from "zod";

// 定义工具
const weatherTool = tool(
  async ({ city }) => {
    return JSON.stringify({
      city,
      temperature: "25°C",
      weather: "晴天",
    });
  },
  {
    name: "get_weather",
    description: "获取指定城市的天气信息",
    schema: z.object({
      city: z.string().describe("城市名称"),
    }),
  }
);

const calculatorTool = tool(
  async ({ expression }) => {
    try {
      // 注意：实际项目中应使用安全的表达式计算库
      const result = eval(expression);
      return String(result);
    } catch {
      return "计算错误";
    }
  },
  {
    name: "calculator",
    description: "计算数学表达式",
    schema: z.object({
      expression: z.string().describe("数学表达式"),
    }),
  }
);

const tools = [weatherTool, calculatorTool];

// 定义状态
const StateAnnotation = Annotation.Root({
  ...ClientStateAnnotation.spec,
  messages: Annotation<any[]>({
    reducer: (x, y) => x.concat(y),
    default: () => [],
  }),
});

// 使用云开发内置的大模型端点
const model = new ChatOpenAI({
  model: process.env.TCB_AI_MODEL || "deepseek-v4-flash",
  apiKey: process.env.TCB_API_KEY,
  configuration: {
    baseURL: `https://${process.env.TCB_ENV_ID}.api.tcloudbasegateway.com/v1/ai/cloudbase`,
  },
}).bindTools(tools);

// 定义节点
async function agentNode(state: typeof StateAnnotation.State) {
  const response = await model.invoke(state.messages);
  return { messages: [response] };
}

// 路由函数
function shouldContinue(state: typeof StateAnnotation.State) {
  const lastMessage = state.messages[state.messages.length - 1];
  if (lastMessage.tool_calls?.length > 0) {
    return "tools";
  }
  return END;
}

// 构建图
const graph = new StateGraph(StateAnnotation)
  .addNode("agent", agentNode)
  .addNode("tools", new ToolNode(tools))
  .addEdge("__start__", "agent")
  .addConditionalEdges("agent", shouldContinue)
  .addEdge("tools", "agent");

const compiledGraph = graph.compile();

// 创建 Agent
const agent = new LanggraphAgent({
  name: "ToolAgent",
  description: "带工具调用的智能助手",
  graph: compiledGraph,
});

module.exports = createAgentServer(agent);

多步骤工作流

import { LanggraphAgent, createAgentServer } from "@cloudbase/agent-adapter-langgraph";
import { StateGraph, Annotation, END } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";

// 定义状态
const StateAnnotation = Annotation.Root({
  input: Annotation<string>(),
  plan: Annotation<string[]>({
    default: () => [],
  }),
  currentStep: Annotation<number>({
    default: () => 0,
  }),
  results: Annotation<string[]>({
    reducer: (x, y) => x.concat(y),
    default: () => [],
  }),
  finalOutput: Annotation<string>(),
});

const model = new ChatOpenAI({
  model: process.env.TCB_AI_MODEL || "deepseek-v4-flash",
  apiKey: process.env.TCB_API_KEY,
  configuration: {
    baseURL: `https://${process.env.TCB_ENV_ID}.api.tcloudbasegateway.com/v1/ai/cloudbase`,
  },
});

// 规划节点
async function planNode(state: typeof StateAnnotation.State) {
  const response = await model.invoke(
    `请将以下任务分解为具体步骤：${state.input}\n返回 JSON 数组格式的步骤列表。`
  );
  const plan = JSON.parse(response.content as string);
  return { plan };
}

// 执行节点
async function executeNode(state: typeof StateAnnotation.State) {
  const currentStep = state.plan[state.currentStep];
  const response = await model.invoke(`请执行以下步骤：${currentStep}`);
  return {
    results: [response.content as string],
    currentStep: state.currentStep + 1,
  };
}

// 总结节点
async function summarizeNode(state: typeof StateAnnotation.State) {
  const response = await model.invoke(
    `请总结以下执行结果：\n${state.results.join("\n")}`
  );
  return { finalOutput: response.content as string };
}

// 路由函数
function shouldContinueExecution(state: typeof StateAnnotation.State) {
  if (state.currentStep < state.plan.length) {
    return "execute";
  }
  return "summarize";
}

// 构建图
const graph = new StateGraph(StateAnnotation)
  .addNode("plan", planNode)
  .addNode("execute", executeNode)
  .addNode("summarize", summarizeNode)
  .addEdge("__start__", "plan")
  .addEdge("plan", "execute")
  .addConditionalEdges("execute", shouldContinueExecution)
  .addEdge("summarize", END);

const compiledGraph = graph.compile();

const agent = new LanggraphAgent({
  name: "WorkflowAgent",
  description: "多步骤工作流 Agent",
  graph: compiledGraph,
});

module.exports = createAgentServer(agent);

Python 版本

安装依赖

pip install cloudbase-agent-langgraph cloudbase-agent-server langgraph langchain-openai

快速开始

from cloudbase_agent.langgraph import LangGraphAgent
from cloudbase_agent.server import create_agent_server
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated
import operator
import os

# 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    response: str

# 使用云开发内置的大模型端点
model = ChatOpenAI(
    model=os.environ.get("TCB_AI_MODEL", "deepseek-v4-flash"),
    api_key=os.environ.get("TCB_API_KEY"),
    base_url=f"https://{os.environ.get('TCB_ENV_ID')}.api.tcloudbasegateway.com/v1/ai/cloudbase",
)

# 定义节点
def chat_node(state: AgentState) -> dict:
    last_message = state["messages"][-1]
    response = model.invoke(last_message)
    return {"response": response.content}

# 构建图
graph = StateGraph(AgentState)
graph.add_node("chat", chat_node)
graph.set_entry_point("chat")
graph.add_edge("chat", END)

compiled_graph = graph.compile()

# 创建 Agent
agent = LangGraphAgent(
    name="ChatBot",
    description="基于 LangGraph 的对话助手",
    graph=compiled_graph,
)

# 创建服务
app = create_agent_server(agent)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=3000)

带工具的 Agent

from cloudbase_agent.langgraph import LangGraphAgent
from cloudbase_agent.server import create_agent_server
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from typing import TypedDict, Annotated
import operator

# 定义工具
@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息"""
    return f"{city}：25°C，晴天"

@tool
def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return str(result)
    except:
        return "计算错误"

tools = [get_weather, calculator]

# 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]

# 使用云开发内置的大模型端点
model = ChatOpenAI(
    model=os.environ.get("TCB_AI_MODEL", "deepseek-v4-flash"),
    api_key=os.environ.get("TCB_API_KEY"),
    base_url=f"https://{os.environ.get('TCB_ENV_ID')}.api.tcloudbasegateway.com/v1/ai/cloudbase",
).bind_tools(tools)

# 定义节点
def agent_node(state: AgentState) -> dict:
    response = model.invoke(state["messages"])
    return {"messages": [response]}

# 路由函数
def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return END

# 构建图
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(tools))
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")

compiled_graph = graph.compile()

# 创建 Agent
agent = LangGraphAgent(
    name="ToolAgent",
    description="带工具调用的智能助手",
    graph=compiled_graph,
)

app = create_agent_server(agent)

核心概念

状态（State）

状态是 LangGraph 的核心，定义了工作流中传递的数据结构：

// TypeScript
const StateAnnotation = Annotation.Root({
  messages: Annotation<any[]>({
    reducer: (x, y) => x.concat(y),  // 定义如何合并状态
    default: () => [],                // 默认值
  }),
});

// Python
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]

节点（Node）

节点是工作流中的处理单元：

// TypeScript
async function myNode(state: StateType) {
  // 处理逻辑
  return { key: newValue };  // 返回状态更新
}

graph.addNode("myNode", myNode);

边（Edge）

边定义了节点之间的连接关系：

// 普通边：固定流向
graph.addEdge("nodeA", "nodeB");

// 条件边：动态路由
graph.addConditionalEdges("nodeA", routingFunction);

客户端状态

使用 ClientStateAnnotation 支持与客户端的状态同步：

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

const StateAnnotation = Annotation.Root({
  ...ClientStateAnnotation.spec,
  // 其他状态字段
});

最佳实践

1. 状态设计

保持状态简洁，只存储必要信息
使用合适的 reducer 函数处理状态合并
考虑状态的可序列化性

2. 节点设计

每个节点只做一件事
节点应该是纯函数（给定相同输入，产生相同输出）
避免在节点中产生副作用

3. 错误处理

async function safeNode(state: StateType) {
  try {
    // 处理逻辑
    return { result: value };
  } catch (error) {
    return { error: error.message };
  }
}

4. 调试

// 添加检查点
const compiledGraph = graph.compile({
  checkpointer: new MemorySaver(),
});

// 获取执行历史
const history = await compiledGraph.getStateHistory(threadId);

部署

云函数部署

参考 HTTP 云函数部署

云托管部署

参考云托管部署

可观测性

云开发为 LangGraph 应用提供内置的可观测性能力，支持通过环境变量或代码配置开启。

环境变量方式（推荐）

# 开启控制台输出
AUTO_TRACES_STDOUT=true

无需修改代码，SDK 会自动读取环境变量并开启追踪。

代码配置方式

Python
TypeScript

from cloudbase_agent.server import AgentServiceApp
from cloudbase_agent.observability.server import ConsoleTraceConfig

app = AgentServiceApp(observability=ConsoleTraceConfig())

import { createExpressRoutes } from "@cloudbase/agent-server";
import { ExporterType } from "@cloudbase/agent-observability/server";

createExpressRoutes({
  createAgent,
  express: app,
  observability: { type: ExporterType.Console }
});

更多配置选项（OTLP 导出、Serverless 环境注意事项等）请参考 Agent 可观测性指南。

前置条件​

TypeScript 版本​

安装依赖​

快速开始​

带工具的 Agent​

多步骤工作流​

Python 版本​

安装依赖​

快速开始​

带工具的 Agent​

核心概念​

状态（State）​

节点（Node）​

边（Edge）​

客户端状态​

最佳实践​

1. 状态设计​

2. 节点设计​

3. 错误处理​

4. 调试​

部署​

云函数部署​

云托管部署​

可观测性​

环境变量方式（推荐）​

代码配置方式​

相关文档​

前置条件

TypeScript 版本

安装依赖

快速开始

带工具的 Agent

多步骤工作流

Python 版本

安装依赖

快速开始

带工具的 Agent

核心概念

状态（State）

节点（Node）

边（Edge）

客户端状态

最佳实践

1. 状态设计

2. 节点设计

3. 错误处理

4. 调试

部署

云函数部署

云托管部署

可观测性

环境变量方式（推荐）

代码配置方式

相关文档