BytesRouter

主题

LangSmith 文档版本 跟随上游 源快照 5db1f00b 更新于 2026-10-18

部署 Google ADK agent

使用 deployments-wrap-sdk 包将 Google Agent Development Kit (ADK) agent 部署到 LangSmith Agent Server。

本指南展示如何部署一个 Google Agent Development Kit (ADK) agent 到 LangSmith Agent Server，使用 deployments-wrap-sdk 包。

deployments-wrap-sdk 提供了一个轻量 wrapper，把一个配置好的 ADK Runner 转换为 LangGraph 兼容的图，因此你无需自己编写 Functional API 胶水代码就能部署 ADK agent。该 wrapper：

前置条件

安装

使用 google-adk extra 安装该包。该 extra 会引入 google-adk 以及 wrapper 所需的其他依赖：

bash
pip install "deployments-wrap-sdk[google-adk]"

NOTE

PyPI 发行名是 deployments-wrap-sdk，但 Python 导入路径是 saf_sdk。两者指的是同一个包。

快速开始

这个最小示例构建了一个将输入原样作为响应返回、且不需要模型 API 密钥的 agent。该 agent 绕过了 LLM 调用，以便你在连接真实模型之前验证部署是否正常工作。

创建 agent.py：

python
from google.adk.agents import Agent
from google.adk.models.llm_response import LlmResponse
from google.adk.runners import Runner
from google.genai.types import Content, Part
from saf_sdk.adk import LangsmithSessionService, wrap

def echo_callback(callback_context, llm_request):
    """Return the user's message instead of calling a real model."""
    user_text = ""
    if callback_context.user_content and callback_context.user_content.parts:
        for part in callback_context.user_content.parts:
            if part.text:
                user_text += part.text
    return LlmResponse(
        content=Content(role="model", parts=[Part(text=f"echo: {user_text}")])
    )

agent = wrap(
    Runner(
        agent=Agent(
            name="echo_agent",
            model="gemini-2.5-flash",
            instruction="Echo the user message.",
            before_model_callback=echo_callback,
        ),
        app_name="adk_echo",
        session_service=LangsmithSessionService(),
    )
)

有两点至关重要：

  1. 将 LangsmithSessionService() 作为 runner 的 session_service 传入。如果你忘记传入，wrap() 会抛出 TypeError。Agent Server 需要这个钩子来通过其检查点器加载和保存 ADK 会话状态。
  2. 将包装后的 agent 作为模块级变量导出。Agent Server 在服务该图时会导入此符号。

对于真实 agent，请去掉 before_model_callback 并直接配置一个模型。例如，通过设置 model="gemini-2.5-flash" 并设置 GOOGLE_API_KEY 来使用 Gemini，或通过 ADK 的 LiteLLM 适配器（google.adk.models.lite_llm.LiteLlm，通过 google-adk[extensions] 提供）使用 Claude/OpenAI。

能力与限制

wrap() 将 ADK 运行时一个已定义的子集桥接到 Agent Server。在移植现有 ADK agent 之前，请审阅以下边界，因为某些 ADK 功能会原样透传，而另一些则有意不予支持。

已支持

  • Agent 原语：Agent、SequentialAgent 和 ParallelAgent，包括通过 sub_agents 参数实现的嵌套子 agent 委派。
  • 工具：Python 函数工具和 LongRunningFunctionTool。
  • 模型：直接使用 Gemini 模型，以及 ADK 的 LiteLLM 适配器（google.adk.models.lite_llm.LiteLlm，通过 google-adk[extensions] 提供）支持的任何模型。在部署上设置提供商的 API 密钥。
  • token 流式传输：ADK 部分事件通过 LangGraph 的异步回调管理器转发，因此 token 块能到达消费 stream_mode="messages" 的客户端和 Studio 聊天视图。
  • 结构化输出：配置了 output_schema 和 output_key 的 agent，除了 messages 之外还会在图的响应上暴露类型化的值。
  • 会话持久化：LangsmithSessionService 将 ADK 会话状态存储在部署的检查点存储中。状态能在重启后存续，并在同一线程的每个后续轮次被加载。
  • 追踪：当 LANGSMITH_TRACING=true 时，wrapper 会自动调用 configure_google_adk()（参阅启用追踪）。
  • 身份验证：如果启用了 Agent Server 身份验证，已认证用户 id 会成为 ADK 的 user_id。否则用户 id 为 "anonymous"。

未支持

  • 多模态输入：wrapper 仅转发 messages[-1].content 作为单个文本部分。入站的图像、文件、音频或内联二进制块都不会传递给 ADK runner。
  • 每轮多条新消息：只有 messages 中的最后一项被视为新的用户消息。对话历史是从 ADK 会话状态重建的，而非从 LangGraph 消息列表。
  • 双向 / 实时流式传输：wrapper 硬编码了 RunConfig(streaming_mode=StreamingMode.SSE)。ADK 的 Runner.run_live() 以及用于音频或语音 agent 的双向流式模式不会被调用，因此无法通过 wrap() 部署实时音频和语音 agent。
  • 非文本输出部分：只从 ADK 事件收集 part.text 值。agent 生成的内联图像、音频或文件不会在图的 messages 输出上呈现。
  • 将中间事件作为消息：响应作为一条包含拼接文本的 AIMessage 发出。工具调用、工具结果和中间子 agent 轮次不会作为图的 messages 字段中独立条目暴露。请改为在 LangSmith 追踪中检查它们。
  • 替代的 ADK 会话服务：runner.session_service 必须是一个 LangsmithSessionService。ADK 的 InMemorySessionService、DatabaseSessionService 和 VertexAiSessionService 会被拒绝并抛出 TypeError，因为会话状态保存在 LangGraph 检查点中。
  • 原生 LangGraph 中断：wrapper 不暴露 LangGraph 的 interrupt 或 Command(resume=...) 机制。基于 LongRunningFunctionTool 构建的人在回路流程遵循 ADK 自己的模式：工具返回一个诸如 pending_approval 的状态，agent 做出回复，随后一个跟进轮次解决该待处理调用。

项目布局

一个可部署的项目需要三个文件：

my-adk-agent/
├── agent.py              # exports the wrapped agent
├── langgraph.json        # Agent Server config
└── pyproject.toml        # Python dependencies

langgraph.json 将 Agent Server 指向导出的符号：

json
{
  "$schema": "https://langgra.ph/schema.json",
  "dependencies": ["."],
  "graphs": {
    "adk_echo": "./agent.py:agent"
  },
  "env": ".env"
}

pyproject.toml 声明依赖：

toml
[project]
name = "my-adk-agent"
version = "0.0.1"
requires-python = ">=3.11"
dependencies = [
    "deployments-wrap-sdk[google-adk]>=0.0.1",
]

安装依赖

bash
pip install -e .

本地运行

使用 LangGraph CLI 启动本地 Agent Server：

bash
langgraph dev

这会在 http://127.0.0.1:2024 服务该 agent，并打开 LangSmith Studio 让你能与 agent 聊天。直接用 curl 发送一个请求：

bash
# Create a thread
THREAD=$(curl -s -X POST http://127.0.0.1:2024/threads \
  -H "Content-Type: application/json" -d '{}' | python -c "import sys, json; print(json.load(sys.stdin)['thread_id'])")

# Run the agent and wait for the final response
curl -s -X POST "http://127.0.0.1:2024/threads/$THREAD/runs/wait" \
  -H "Content-Type: application/json" \
  -d '{
    "assistant_id": "adk_echo",
    "input": {"messages": [{"type": "human", "content": "Hello"}]}
  }'

部署到 LangSmith

一旦 agent 在本地运行，就用 langgraph deploy 将其部署到 LangSmith：

bash
langgraph deploy --name my-adk-agent

有关环境配置、部署类型和修订管理，参阅部署到 Cloud。对于自托管设置，参阅自托管部署。

启用追踪

每当启用 LangSmith 追踪时，wrap() 都会自动调用 langsmith.integrations.google_adk.configure_google_adk()，因此你需要做的只是在部署上设置环境变量：

bash
LANGSMITH_API_KEY=your-langsmith-api-key
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=my-adk-agent     # optional
GOOGLE_API_KEY=your-google-api-key

追踪 会在 LangSmith UI 中显示 agent 调用、工具调用和 LLM 交互。有关底层追踪集成的更多信息，参阅追踪 Google ADK 应用。

API 参考

wrap(runner)

包装一个配置好的 google.adk.runners.Runner，并返回一个可从你的模块导出、并由 Agent Server 服务的 LangGraph Pregel 图。

参数类型描述
runnergoogle.adk.runners.Runner一个配置好的 ADK Runner。其 session_service 必须是一个 LangsmithSessionService。

返回： 一个名字为 runner.app_name 的 Pregel 图。

抛出： 如果 runner.session_service 不是 LangsmithSessionService，则抛出 TypeError。

如果 runner.agent 定义了 output_key，该键的值除了 messages 之外也会在图的输出上暴露。正是这一点让 ADK 结构化输出 agent（output_schema=...、output_key=...）能够配合 Studio 和 /runs/wait 响应工作。

LangsmithSessionService

一个由 Agent Server 检查点存储支撑的 google.adk.sessions.BaseSessionService 实现。wrapper 自动管理会话生命周期。它在线程的第一轮创建一个会话，在后续轮次从检查点加载它，并在运行完成时将更新后的会话写回。

为每个 Runner 使用一个全新实例：

python
session_service = LangsmithSessionService()

你应当不需要直接调用它的方法；wrap() 会通过 ADK 正常的会话生命周期来驱动它们。

ADKInput

一个被包装 agent 的默认输入 schema。

字段类型描述
messageslist[AnyMessage]（必需）对话消息；wrapper 将 messages[-1].content 作为新用户消息发送给 ADK runner。
state_deltadict[str, Any] | None（可选）透传给 runner.run_async(state_delta=...)，以在本轮修改 ADK 会话状态。

ADKOutput

一个被包装 agent 的默认输出 schema。

字段类型描述
messageslist[AnyMessage]agent 的响应消息，通过 LangGraph 的 add_messages reducer 追加到线程。

将 messages 作为一个类型化字段（而非普通 dict）暴露，正是这一点让 Studio 将该图识别为聊天兼容并启用聊天模式切换。

工作原理

当一个运行到达时：

  1. 被包装的图从运行配置中读取 thread_id，并将其用作 ADK 的 session_id。如果启用了身份验证，已认证用户的 id 会成为 ADK 的 user_id；否则用户 id 为 "anonymous"。
  2. wrapper 从 LangGraph 检查点将先前的会话（如有）加载到 LangsmithSessionService，然后请求 runner 处理最新消息。
  3. runner 发出 ADK 事件。wrapper 通过 LangGraph 的异步回调管理器转发部分 token 事件，使它们通过 stream_mode="messages" 流出，并为响应消息收集最终文本。
  4. 当运行完成时，wrapper 序列化 ADK 会话并通过 entrypoint.final(save=...) 将其保存到检查点。同一线程上的下一次运行会从该状态恢复。

这意味着 ADK 自身的会话/状态语义被端到端地保留，同时部署获得标准的 Agent Server 特性：持久运行、流式传输、多线程持久化和追踪。

友情链接

  • Agents.md面向 AI 编码助手的仓库说明文件规范
  • LangChainLangChain 是一个开放的Agent平台
  • LangSmithAgent 工程平台：可观测与评估、部署、Fleet、Engine 等
  • EmbabelEmbabel 是一个JVM平台Agent框架
  • SpringAISpringAI 是一个基于 Spring 的 AI 框架
  • MCPModel Context Protocol，连接 AI 应用与外部数据/工具的开放标准
  • LangChain4jJava 生态的 LLM 应用开发框架
  • LlamaIndex把数据接入 LLM 的框架与文档平台
  • LlamaParse智能体式 OCR 文档解析：把 PDF、扫描件变成 LLM 可用的文本与结构化数据
  • LiteParse本地优先、轻量的文档解析库，不需要 API key
  • CrewAI多智能体编排框架：以 Crew/Flow 组织自主协作的 AI Agent
  • MAFMicrosoft Agent Framework，企业级 .NET/Python 智能体框架

BytesRouter 是社区翻译站点，所有译文仅供参考，一切以官方英文原文为准。

© 2026 BytesRouter · 译文采用 CC BY 4.0 许可