Langgraph logo Langgraph
登录
16. 集成与扩展能力

16.集成与扩展能力

当应用从实验环境走向生产部署，扩展能力往往决定了架构的天花板。LangGraph 的设计哲学从不假设开发者只需要一个封闭的黑盒，而是提供了一套完整的扩展机制，让自定义路由、中间件、异步通知、标准化协议和动态 UI 都能无缝融入已有的图结构。这套扩展体系既保持了核心运行时的简洁，又赋予了系统足够的弹性去应对真实世界的复杂需求。

自定义 API 路由扩展

部署到 LangGraph Platform 后，系统会自动暴露一套标准的 RESTful API，涵盖线程管理、运行调用、状态查询等核心功能。但真实场景总有些特殊需求：可能需要提供一个健康检查端点供监控系统调用，或者暴露内部指标给 Prometheus，甚至实现一套完整的 OAuth 回调流程。这时候就需要在现有服务上扩展自定义路由。

LangGraph Platform 的解决思路很直接——它允许开发者提供一个完整的 Starlette 应用对象（包括 FastAPI、FastHTML 等兼容框架），与内置 API 共存。这种方式的优势在于完全掌控：既能复用成熟的 Web 框架生态，又能保持与原生 API 的一致性。

实现自定义路由需要两步。首先创建一个 Web 应用文件，定义路由和处理函数：

# ./src/agent/webapp.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/hello")
def read_root():
    return {"Hello": "World"}

@app.get("/health")
def health_check():
    return {"status": "healthy", "timestamp": "2024-01-01T00:00:00Z"}

这里创建了一个 FastAPI 实例，添加了两个简单端点。/hello 返回问候信息，/health 提供健康检查能力。关键点在于这个 app 对象必须是符合 ASGI 规范的应用实例，FastAPI 和 Starlette 都满足这个要求。

接下来在 langgraph.json 中声明这个应用：

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./src/agent/graph.py:graph"
  },
  "env": ".env",
  "http": {
    "app": "./src/agent/webapp.py:app"
  }
}

http.app 字段指向刚才创建的 FastAPI 实例。启动服务后，访问 localhost:2024/hello 和 localhost:2024/health 就能看到自定义端点的响应。值得注意的是，自定义路由优先级高于系统默认路由，这意味着必要时可以完全重写内置行为——虽然这种情况很少见，但在某些特殊场景下（比如统一鉴权逻辑）会非常有用。

更复杂的场景可能需要完整的 CRUD 接口。比如为内部运营人员提供一个管理界面，可以查询当前活跃线程、查看系统负载、甚至手动触发某些维护任务。这时候可以在同一个 FastAPI 应用中组织多个路由模块，利用依赖注入管理数据库连接、配置读取等共享资源。由于整个应用运行在 LangGraph Server 的上下文中，还可以直接导入编译好的图实例，在自定义路由中调用 graph.invoke() 或 graph.astream()，实现与主业务逻辑的深度集成。

中间件生命周期管理

中间件是 Web 服务的横切关注点解决方案。日志记录、请求追踪、CORS 处理、认证鉴权……这些功能往往需要在每个请求处理前后执行，但又不属于具体业务逻辑。LangGraph Platform 允许通过自定义应用对象注入中间件，实现请求级别的全局控制。

添加中间件的方式与标准 FastAPI/Starlette 应用完全一致。创建一个中间件类，继承 BaseHTTPMiddleware，实现 dispatch 方法：

# ./src/agent/webapp.py
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI()

class CustomHeaderMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # 请求处理前的逻辑
        response = await call_next(request)
        # 响应返回前的逻辑
        response.headers['X-Custom-Header'] = 'Hello from middleware!'
        response.headers['X-Request-ID'] = request.headers.get('x-request-id', 'unknown')
        return response

app.add_middleware(CustomHeaderMiddleware)

这个中间件在每个响应中注入了两个自定义头。call_next(request) 是关键调用，它将请求传递给下一个中间件或最终的路由处理函数，返回响应对象后还能继续修改。这种模式非常适合实现请求追踪：从请求头中提取或生成唯一 ID，贯穿整个调用链，最终再返回给客户端。

除了请求级别的中间件，服务级别的生命周期管理同样重要。服务启动时可能需要初始化数据库连接池、加载机器学习模型、预热缓存；关闭时需要优雅释放资源。FastAPI 提供了 lifespan 上下文管理器来处理这类场景：

# ./src/agent/webapp.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncpg

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时执行
    app.state.db_pool = await asyncpg.create_pool("postgresql://...")
    app.state.model = load_large_model()
    yield
    # 关闭时执行
    await app.state.db_pool.close()
    del app.state.model

app = FastAPI(lifespan=lifespan)

在 lifespan 上下文中初始化的资源可以通过 app.state 访问，这种方式比全局变量更优雅，也更容易测试。结合 LangGraph 的持久化层，可以在服务启动时执行数据迁移，或在关闭前将内存中的状态刷入检查点。

配置中间件和生命周期事件后，langgraph.json 的写法与自定义路由完全相同，都是通过 http.app 字段指定应用对象。这种统一的设计降低了学习成本，也让扩展行为更加可预测。

Webhook 异步通知

长时间运行的任务在 AI 应用中很常见：生成一份深度研究报告可能需要调用多个工具、执行数十次 LLM 调用，耗时数分钟甚至更久。让客户端一直等待响应显然不现实。Webhook 机制允许在任务完成后主动通知外部系统，将结果推送到指定的回调地址。

LangGraph Platform 的多个 API 端点支持 webhook 参数。当任务完成时，平台会向这个 URL 发送 POST 请求，携带执行结果和元数据。支持的端点包括创建运行、流式运行、后台任务、定时任务等，覆盖了大部分异步场景。

使用 webhook 的流程分为三步。首先准备接收端，这是一个能够处理 POST 请求的标准 HTTP 端点：

# 假设这是外部服务的接收端点
@app.post("/webhook-handler")
async def handle_webhook(payload: dict):
    run_id = payload["run_id"]
    status = payload["status"]
    result = payload.get("result", {})
    
    if status == "success":
        await save_result_to_database(run_id, result)
        await send_notification_to_user(run_id, "任务已完成")
    else:
        await log_error(run_id, payload.get("error"))
    
    return {"acknowledged": True}

这个端点需要验证请求来源（可以通过签名或白名单 IP），解析 payload，根据状态执行相应业务逻辑。Payload 的结构包含运行 ID、状态、结果、错误信息等关键字段。

然后在调用 LangGraph API 时指定 webhook 参数：

from langgraph_sdk import get_client

client = get_client(url="https://your-deployment.url", api_key="your-api-key")

# 创建线程
thread = await client.threads.create()
thread_id = thread["thread_id"]

# 启动流式运行并指定 webhook
input_data = {"messages": [{"role": "user", "content": "分析过去一周的销售数据"}]}

async for chunk in client.runs.stream(
    thread_id,
    "agent",
    input=input_data,
    stream_mode="updates",
    webhook="https://my-server.app/webhook-handler"
):
    # 可以实时处理中间结果
    print(f"收到更新: {chunk.data}")

# 客户端连接可以断开，最终结果会通过 webhook 推送

这里的关键是 webhook 参数。即使客户端在任务完成前断开连接，平台仍会保证 webhook 的投递。对于完全无状态的客户端（比如浏览器页面刷新后），webhook 是唯一可靠的结果获取方式。

Webhook 的 payload 结构会根据端点类型略有差异。流式运行的 webhook 会包含最终聚合结果，后台任务的 webhook 会包含任务状态转换信息。开发接收端时最好打印完整 payload 观察结构，再提取所需字段。

生产环境中还需要考虑失败重试。LangGraph Platform 会在 webhook 返回非 2xx 状态码时自动重试，指数退避策略避免对故障服务造成更大压力。接收端应该实现幂等性，因为重试可能导致多次接收相同通知。

MCP 协议服务暴露

Model Context Protocol（MCP）是 Anthropic 推出的开放协议，旨在标准化 LLM 应用与外部工具、数据源的交互方式。它定义了一套统一的接口，让任何兼容 MCP 的客户端都能发现和使用远程服务提供的工具。LangGraph 从 0.2.3 版本开始内置 MCP 服务端实现，可以将部署的 agent 直接暴露为 MCP 工具。

这种能力的意义在于生态互操作性。假设团队 A 用 LangGraph 开发了一个专业的金融分析 agent，团队 B 使用 Claude Desktop 或其他 MCP 客户端构建助手，通过 MCP 协议，B 可以直接调用 A 的 agent 作为工具，无需了解其内部实现，也无需集成专用 SDK。

启用 MCP 功能非常简单。首先确保使用足够新的版本：

pip install "langgraph-api>=0.2.3" "langgraph-sdk>=0.1.61"

然后在 langgraph.json 中为 agent 添加描述信息：

{
  "graphs": {
    "finance_agent": {
      "path": "./src/finance/agent.py:graph",
      "description": "专业的金融数据分析助手，支持股票查询、财报解读、市场趋势分析"
    }
  }
}

description 字段会被 MCP 客户端用作用户提示，帮助 LLM 理解这个工具的用途。部署后，每个 graph 会自动暴露为 MCP 工具，工具名就是 graph ID，输入输出模式由 graph 的状态模式推导。

为了让 LLM 更好地理解工具用途，建议为 graph 定义明确的输入输出模式，而不是使用通用的 MessagesState：

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class InputState(TypedDict):
    question: str  # 明确的问题字段

class OutputState(TypedDict):
    answer: str    # 明确的答案字段
    sources: list[str]  # 引用来源

class OverallState(InputState, OutputState):
    pass

def analyze_node(state: InputState):
    # 实际的分析逻辑
    return {
        "answer": f"基于分析，{state['question']} 的答案是...",
        "sources": ["财报数据", "市场分析"]
    }

builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)
builder.add_node("analyze", analyze_node)
builder.add_edge(START, "analyze")
builder.add_edge("analyze", END)
graph = builder.compile()

显式定义模式后，MCP 客户端能生成更精确的工具调用参数，避免 LLM 猜测数据结构。input_schema 和 output_schema 参数让 graph 的接口契约更加清晰。

从客户端连接 LangGraph 提供的 MCP 服务，可以使用 langchain-mcp-adapters：

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent

# 配置远程 MCP 服务
server_params = {
    "url": "https://finance-agent.your-org.langgraph.app/mcp",
    "headers": {"X-Api-Key": "lsv2_pt_your_api_key"}
}

async def main():
    async with streamablehttp_client(**server_params) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # 加载远程工具
            tools = await load_mcp_tools(session)
            
            # 创建本地 agent，使用远程工具
            agent = create_react_agent("anthropic:claude-3-7-sonnet-latest", tools)
            
            # 用户提问时，agent 会自动决定是否调用远程金融分析工具
            result = await agent.ainvoke({
                "messages": "帮我分析特斯拉最新财报，重点关注现金流和毛利率"
            })
            
            print(result["messages"][-1].content)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

这个例子展示了 MCP 的真正价值：本地 agent 不需要知道金融分析工具的实现细节，它只需要通过 MCP 协议发现工具、理解工具用途，然后在适当的时候调用。远程的 LangGraph agent 处理复杂逻辑，返回结构化结果，本地 agent 再整合到对话中。

MCP 端点使用与 LangGraph API 相同的认证机制，通过 X-Api-Key 头传递凭证。如果需要禁用 MCP 功能，可以在配置中设置：

{
  "http": {
    "disable_mcp": true
  }
}

目前 LangGraph 的 MCP 实现基于 Streamable HTTP 传输，每个请求都是独立的，不维护会话状态。这意味着每次工具调用都是无状态的，远程 agent 的线程状态不会被自动保留。如果需要跨调用保持上下文，可以在输入模式中设计 thread_id 字段，由客户端传递。

生成式 UI 组件流

文本对话是 LLM 应用的基础交互形式，但许多场景需要更丰富的界面：数据可视化、交互式表单、多媒体展示。生成式 UI（Generative UI）让 agent 能够动态生成 React 组件并推送到前端，实现对话驱动的界面适配。

LangGraph Platform 的生成式 UI 方案将组件定义与图代码同仓管理，通过配置关联。组件只在需要时加载，避免初始 bundle 过大。支持 Tailwind CSS 和 shadcn/ui，让组件风格统一且美观。

实现生成式 UI 分为三步。第一步定义组件，创建 UI 模块文件：

// src/agent/ui.tsx
import "./styles.css";

// 天气展示组件
const WeatherComponent = (props: { city: string; temperature: number; condition: string }) => {
  return (
    <div className="bg-blue-50 border border-blue-200 rounded-lg p-4 my-4">
      <h3 className="text-lg font-semibold text-blue-900">{props.city} 天气</h3>
      <div className="mt-2 text-3xl font-bold text-blue-700">{props.temperature}°C</div>
      <div className="text-blue-600">{props.condition}</div>
    </div>
  );
};

// 导出组件映射，key 是组件标识符
export default {
  weather: WeatherComponent,
};

组件接收 props 渲染，与普通 React 组件无异。可以导入 CSS 文件，使用 Tailwind 类名，甚至引入第三方组件库。LangGraph Platform 在构建时会将这些组件打包成独立资源，不会污染主应用。

第二步在 langgraph.json 中配置 UI 组件路径：

{
  "node_version": "20",
  "graphs": {
    "agent": "./src/agent/graph.py:graph"
  },
  "ui": {
    "agent": "./src/agent/ui.tsx"
  }
}

ui 字段的 key 对应 graph 名称，value 是 UI 组件文件路径。这种设计支持多 graph 场景，每个 graph 可以有独立的 UI 组件集。

第三步在 graph 中发送 UI 组件。需要扩展状态模式，加入 UI 消息管理：

# src/agent/graph.py
import uuid
from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import AIMessage, BaseMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.graph.ui import AnyUIMessage, ui_message_reducer, push_ui_message

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    ui: Annotated[Sequence[AnyUIMessage], ui_message_reducer]  # UI 消息状态

async def weather_node(state: AgentState):
    # 提取城市信息（简化示例）
    class WeatherOutput(TypedDict):
        city: str
    
    # 使用结构化输出获取参数
    weather: WeatherOutput = (
        await ChatOpenAI(model="gpt-4o-mini")
        .with_structured_output(WeatherOutput)
        .ainvoke(state["messages"])
    )
    
    # 创建文本消息
    message = AIMessage(
        id=str(uuid.uuid4()),
        content=f"这是 {weather['city']} 的天气信息",
    )
    
    # 推送 UI 组件，关联到消息
    push_ui_message(
        "weather",  # 组件标识符
        {"city": weather["city"], "temperature": 25, "condition": "晴朗"},
        message=message
    )
    
    return {
        "messages": [message],
        # ui_message_reducer 会自动处理 UI 消息的追加
    }

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("weather", weather_node)
workflow.add_edge(START, "weather")
workflow.add_edge("weather", END)
graph = workflow.compile()

push_ui_message 函数将 UI 组件与特定消息关联。第一个参数是组件标识符（对应 UI 模块中的 key），第二个参数是 props，第三个参数是关联的消息对象。当这条消息被发送到前端时，UI 组件会随消息一起渲染。

前端使用 useStream hook 接收流式响应时，会自动处理 UI 消息：

import { useStream } from "@langchain/langgraph-sdk/react";

function ChatInterface() {
  const { messages, submit } = useStream({
    apiUrl: "https://your-agent.url",
    assistantId: "agent",
  });

  return (
    <div>
      {messages.map((msg) => (
        <div key={msg.id}>
          {/* 文本消息渲染 */}
          <p>{msg.content}</p>
          
          {/* UI 组件自动渲染 */}
          {msg.ui?.map((uiItem, idx) => (
            <UIComponentRenderer key={idx} ui={uiItem} />
          ))}
        </div>
      ))}
      
      <button onClick={() => submit("北京天气怎么样？")}>
        查询天气
      </button>
    </div>
  );
}

UIComponentRenderer 是框架提供的组件，负责动态加载和渲染从服务器推送的 UI 组件。它会在运行时从 LangGraph Platform 加载组件代码，缓存后复用，避免重复网络请求。

生成式 UI 的强大之处在于动态性。同一个 agent 可以根据不同场景推送不同组件：查询天气时推送 WeatherComponent，生成报告时推送 ChartComponent，收集用户反馈时推送 RatingComponent。组件的 props 由 agent 运行时决定，实现了真正的对话驱动界面。

总结

从最初的状态图定义，到复杂的记忆系统，再到生产部署和性能优化，这本教程涵盖了 LangGraph 的完整技术栈。最后一章的集成与扩展能力，为整个体系画上了开放的句号——它告诉我们，框架的价值不仅在于提供了什么，更在于能如何与现有系统共生。

自定义 API 路由和中间件让 LangGraph Server 不再是孤立的服务，而是能融入企业现有技术栈的有机组件。Webhook 机制解决了异步任务的通知难题，让长时运行不再意味着长时等待。MCP 协议的拥抱展现了生态思维，让 LangGraph 构建的 agent 能成为更大智能系统的积木。生成式 UI 则突破了文本对话的边界，让 agent 的输出形式随场景而变。

这些扩展机制的共同特点是：它们都不侵入核心运行时。无论是添加路由还是实现 MCP，都是在标准接口之上的叠加。这种分层设计让 LangGraph 既能保持底层的简洁高效，又能满足上层的多样化需求。

回顾全书，LangGraph 的核心设计哲学贯穿始终：状态驱动、显式控制、持久化优先。从最简单的 StateGraph 开始，我们学会了用类型化的状态管理数据流；通过节点和边，我们掌握了将业务逻辑拆解为可组合单元的方法；记忆系统让我们理解了短期上下文与长期知识的区别；人工介入控制让我们看到了自动化与人性化的平衡；流式交互让用户体验从等待变为共创；时间旅行能力让调试不再是黑盒猜测；平台功能将本地开发平滑过渡到生产部署；性能优化让理论架构经得起流量考验。

每一章都在回答一个核心问题：如何构建可靠、可控、可扩展的智能体系统。答案不是某个单一技术，而是一整套协同工作的机制。状态管理提供基础，节点系统赋予结构，记忆注入上下文，人工介入确保边界，流式反馈提升体验，持久化保障可靠，平台化简化运维，扩展性适应未来。

掌握这些概念和工具后，面对真实的业务场景，我们能做的不再是拼凑提示词和 API 调用，而是设计状态模式、规划节点流程、配置记忆策略、设置检查点、设计人工审核点、优化流式输出、规划部署架构。这种系统化的思维方式，正是复杂 AI 应用从 demo 走向产品的关键。

技术总在演进，但好的设计原则具有持久生命力。LangGraph 从 Pregel 和 Apache Beam 中汲取的图计算思想，从网络协议中借鉴的显式状态管理，都经过了大规模系统的验证。理解这些原理，不仅能用好当下的工具，也能在未来面对新框架时快速迁移和判断。

至此，这本教程抵达终点，但实践之路才刚刚开始。建议从简单的对话机器人入手，逐步加入工具调用，然后引入记忆系统，接着设置人工介入点，最后尝试部署到平台并扩展自定义功能。每个阶段都深入理解其设计意图，而不是简单复制代码。当这些概念内化为本能，就能真正释放 LangGraph 的潜力，构建出既智能又可靠的 AI 应用。

上一章

性能优化与扩展