部署 Google ADK 智能体

本指南介绍如何使用 deployments-wrap-sdk 包，将一个 Google Agent Development Kit (ADK) 智能体部署到 LangSmith Agent Server。 deployments-wrap-sdk 提供了一个轻量级封装，可将配置好的 ADK Runner 转换为兼容 LangGraph 的图，因此你无需自行编写 Functional API glue 代码即可部署 ADK 智能体。该封装可以：

将 ADK 会话桥接到 Agent Server 的检查点持久化，使会话状态能够跨重启保留，并在多次运行之间恢复。
通过 LangGraph 的流式传输管道转发 ADK token 事件，使部分 token 显示在 stream_mode="messages" 和 LangSmith Studio 中。
当设置 LANGSMITH_TRACING 时，自动为 ADK 启用 LangSmith 跟踪。

前置条件

Python 3.11+
用于本地开发和部署的 LangGraph CLI
LangSmith API 密钥，参见创建账户和 API 密钥
如果使用 Gemini 模型，则需要 Google AI API 密钥，参见 Google AI Studio

安装

使用 google-adk extra 安装该包。此 extra 会拉取 google-adk 以及封装所需的其他依赖：

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

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

快速开始

下面这个最小示例会构建一个将输入作为响应返回的智能体，不需要模型 API 密钥。该智能体会绕过 LLM 调用，因此你可以在连接真实模型之前，先验证部署是否正常工作。创建 agent.py：

agent.py

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):
    """返回用户消息，而不是调用真实模型。"""
    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"回显: {user_text}")])
    )


agent = wrap(
    Runner(
        agent=Agent(
            name="echo_agent",
            model="gemini-2.5-flash",
            instruction="回显用户消息。",
            before_model_callback=echo_callback,
        ),
        app_name="adk_echo",
        session_service=LangsmithSessionService(),
    )
)

有两点非常重要：

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

对于真实智能体，请移除 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 智能体之前，请先查看以下边界，因为某些 ADK 功能会原样传递，而另一些功能则明确不受支持。

支持的功能

智能体原语：Agent、SequentialAgent 和 ParallelAgent，包括通过 sub_agents 参数进行嵌套子智能体委派。
工具：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 的智能体，会在图的响应中通过 messages 之外的字段暴露类型化值。
会话持久化：LangsmithSessionService 会将 ADK 会话状态存储在部署的检查点存储中。状态会跨重启保留，并在同一线程的后续轮次中加载。
跟踪：当 LANGSMITH_TRACING=true 时，封装会自动调用 configure_google_adk()（参见启用跟踪）。
身份验证：如果启用了 Agent Server身份验证，经过身份验证的用户 ID 会成为 ADK 的 user_id。否则，用户 ID 为 "anonymous"。

不支持的功能

多模态输入：封装只会将 messages[-1].content 作为单个文本部分转发。传入的图像、文件、音频或内联二进制块不会传递给 ADK runner。
每轮多个新消息：只有 messages 中的最后一项会被视为新的用户消息。对话历史会从 ADK 会话状态中重建，而不是从 LangGraph 消息列表中重建。
双向/实时流式传输：封装会固定使用 RunConfig(streaming_mode=StreamingMode.SSE)。不会调用 ADK 的 Runner.run_live()，也不会使用音频或语音智能体所需的双向流式模式，因此无法通过 wrap() 部署实时音频和语音智能体。
非文本输出部分：只会收集 ADK 事件中的 part.text 值。智能体生成的内联图像、音频或文件不会出现在图的 messages 输出中。
将中间事件作为消息：响应会作为一个包含拼接文本的 AIMessage 发出。工具调用、工具结果以及中间子智能体轮次不会作为图 messages 字段中的独立项目公开。如需查看这些内容，请改用 LangSmith 跟踪。
其他 ADK 会话服务：runner.session_service 必须是 LangsmithSessionService。由于会话状态由 LangGraph 检查点保存，ADK 的 InMemorySessionService、DatabaseSessionService 和 VertexAiSessionService 会被 TypeError 拒绝。
原生 LangGraph 中断：封装不会公开 LangGraph 的 interrupt 或 Command(resume=...) 机制。基于 LongRunningFunctionTool 构建的人工介入流程遵循 ADK 自身的模式：工具返回类似 pending_approval 的状态，智能体进行回复，然后由后续轮次解决待处理的调用。

项目结构

一个可部署的项目需要包含以下三个文件：

my-adk-agent/
├── agent.py              # 导出封装后的智能体
├── langgraph.json        # Agent Server 配置
└── pyproject.toml        # Python 依赖

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

langgraph.json

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

pyproject.toml 声明依赖：

pyproject.toml

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

安装依赖

pip install -e .

本地运行

使用 LangGraph CLI 启动本地 Agent Server：

langgraph dev

该命令会在 http://127.0.0.1:2024 提供智能体服务，并打开 LangSmith Studio，以便你与智能体聊天。使用 curl 直接发送请求：

# 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

智能体在本地运行后，使用 langgraph deploy 将其部署到 LangSmith：

langgraph deploy --name my-adk-agent

有关环境配置、部署类型和修订版本管理，请参阅部署到云端。有关自托管设置，请参阅自托管部署。

启用跟踪

只要启用了 LangSmith 跟踪，wrap() 就会自动调用 langsmith.integrations.google_adk.configure_google_adk()，因此你只需在部署中设置环境变量：

.env

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 中显示智能体调用、工具调用和 LLM 交互。有关底层跟踪集成的更多信息，请参阅跟踪 Google ADK 应用。

API 参考

`wrap(runner)`

封装一个已配置的 google.adk.runners.Runner，并返回一个 LangGraph Pregel 图。该图可以从模块中导出，并由 Agent Server 提供服务。

参数	类型	描述
`runner`	`google.adk.runners.Runner`	已配置的 ADK Runner。其 `session_service` 必须是 `LangsmithSessionService`。

返回值： 一个名称为 runner.app_name 的 Pregel 图。 抛出异常： 如果 runner.session_service 不是 LangsmithSessionService，则抛出 TypeError。如果 runner.agent 定义了 output_key，则该键对应的值也会暴露在图的输出中，同时保留 messages。这正是 ADK 结构化输出智能体（output_schema=...、output_key=...）能够与 Studio 和 /runs/wait 响应配合使用的原因。

`LangsmithSessionService`

一个由 Agent Server 检查点存储支持的 google.adk.sessions.BaseSessionService 实现。该封装会自动管理会话生命周期。它会在线程的第一个轮次创建会话，在后续轮次中从检查点加载会话，并在运行完成后将更新后的会话写回。每个 Runner 使用一个新实例：

session_service = LangsmithSessionService()

通常不需要直接调用其方法；wrap() 会通过 ADK 的正常会话生命周期驱动这些方法。

`ADKInput`

封装后智能体的默认输入模式。

字段	类型	描述
`messages`	`list[AnyMessage]`	（必填）对话消息；封装会将 `messages[-1].content` 作为新的用户消息发送给 ADK runner。
`state_delta`	`dict[str, Any] \| None`	（可选）传递给 `runner.run_async(state_delta=...)`，用于修改本轮的 ADK 会话状态。

`ADKOutput`

封装后智能体的默认输出模式。

字段	类型	描述
`messages`	`list[AnyMessage]`	智能体的响应消息，会通过 LangGraph 的 `add_messages` reducer 追加到线程中。

将 messages 暴露为类型化字段（而不是普通的 dict），Studio 才能检测出该图兼容聊天功能，并启用聊天模式切换开关。

工作原理

当收到一次运行请求时：

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

这意味着 ADK 自身的会话/状态语义能够端到端保留，同时部署也可以获得标准 Agent Server 功能：持久化运行、流式传输、多线程持久化和跟踪。

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

Deployment components

Develop & test

Frameworks and platforms

Reference

前置条件

安装

快速开始

功能与限制

支持的功能

不支持的功能

项目结构

安装依赖

本地运行

部署到 LangSmith

启用跟踪

API 参考

`wrap(runner)`

`LangsmithSessionService`

`ADKInput`

`ADKOutput`

工作原理

​前置条件

​安装

​快速开始

​功能与限制

​支持的功能

​不支持的功能

​项目结构

​安装依赖

​本地运行

​部署到 LangSmith

​启用跟踪

​API 参考

​wrap(runner)

​LangsmithSessionService

​ADKInput

​ADKOutput

​工作原理

前置条件

安装

快速开始

功能与限制

支持的功能

不支持的功能

项目结构

安装依赖

本地运行

部署到 LangSmith

启用跟踪

API 参考

`wrap(runner)`

`LangsmithSessionService`

`ADKInput`

`ADKOutput`

工作原理