基于 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 || "hunyuan-turbos-latest",
  apiKey: process.env.TCB_API_KEY,
  configuration: {
    baseURL: `https://${process.env.TCB_ENV_ID}.api.tcloudbasegateway.com/v1/ai/hunyuan/v1`,
  },
});

// 定义节点
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 || "hunyuan-turbos-latest",
  apiKey: process.env.TCB_API_KEY,
  configuration: {
    baseURL: `https://${process.env.TCB_ENV_ID}.api.tcloudbasegateway.com/v1/ai/hunyuan/v1`,
  },
}).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 || "hunyuan-turbos-latest",
  apiKey: process.env.TCB_API_KEY,
  configuration: {
    baseURL: `https://${process.env.TCB_ENV_ID}.api.tcloudbasegateway.com/v1/ai/hunyuan/v1`,
  },
});

// 规划节点
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", "hunyuan-turbos-latest"),
    api_key=os.environ.get("TCB_API_KEY"),
    base_url=f"https://{os.environ.get('TCB_ENV_ID')}.api.tcloudbasegateway.com/v1/ai/hunyuan/v1",
)

# 定义节点
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", "hunyuan-turbos-latest"),
    api_key=os.environ.get("TCB_API_KEY"),
    base_url=f"https://{os.environ.get('TCB_ENV_ID')}.api.tcloudbasegateway.com/v1/ai/hunyuan/v1",
).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 云函数部署

云托管部署

参考 云托管部署

相关文档

• LangChain 开发指南
• CrewAI 开发指南
• AG-UI 协议