部署 Google ADK 智能体 - 以编程方式使用文档

本指南介绍如何部署 Google Agent Development Kit (ADK) 智能体 LangSmith Agent Server 使用 deployments-wrap-sdk package.

deployments-wrap-sdk 提供了一个轻量级包装器，将配置好的 ADK Runner 转换为 LangGraph 兼容的图，因此您可以部署 ADK 智能体而无需编写 Functional API 粘合代码。包装器：

- 将 ADK 会话桥接到 Agent Server 的检查点持久化，以便会话状态在重启后仍然保留，并在多次运行间恢复。
- 通过 LangGraph 的流式管道转发 ADK token 事件，以便部分 token 出现在 stream_mode="messages" 和 LangSmith Studio.
- 自动启用 LangSmith 追踪用于 ADK（当 LANGSMITH_TRACING 设置时）。

前提条件

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

安装

使用以下命令安装包 google-adk 额外选项。该额外选项会拉取 google-adk 以及其他包装器所需的依赖项：

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

快速入门

这个最小示例构建了一个返回输入内容作为响应的智能体，不需要模型 API 密钥。该智能体绕过 LLM 调用，这样您可以在连接真实模型之前验证部署是否正常工作。

创建 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):
    """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(),
    )
)

两件事至关重要：

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

对于真实智能体，去掉 before_model_callback 并直接配置模型。例如，通过设置 model="gemini-2.5-flash" 来使用 Gemini GOOGLE_API_KEY set, or use Claude/OpenAI via ADK's LiteLLM adapter (google.adk.models.lite_llm.LiteLlm，可通过 google-adk[extensions]).

获取

wrap() 将 ADK 运行时的一个定义子集桥接到 Agent Server。在移植现有 ADK 代理之前，请查看以下边界，因为某些 ADK 功能会原封不动地传递，而其他功能则有意不被支持。

支持

代理原语: Agent, SequentialAgent，以及 ParallelAgent，包括通过以下方式进行的嵌套子代理委托 sub_agents parameter.
工具：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 运行器。
每轮多条新消息：中的最后一项 messages 被视为新的用户消息。对话历史从 ADK 会话状态重建，而非从 LangGraph 消息列表重建。
Bidirectional / live streaming：包装器硬编码 RunConfig(streaming_mode=StreamingMode.SSE)。ADK 的 Runner.run_live() 以及用于音频或语音代理的双向流模式未被调用，因此无法通过以下方式部署实时音频和语音代理 wrap().
非文本输出部分：仅 part.text 值从 ADK 事件中收集。代理生成的内联图像、音频或文件不会显示在图的 messages output.
中间事件作为消息：响应作为一条 AIMessage 发出，包含串联的文本。工具调用、工具结果和中间子代理轮次不会作为单独的项显示在图的 messages 字段中。在以下位置检查它们 LangSmith 追踪 instead.
替代 ADK 会话服务: runner.session_service 必须是 LangsmithSessionService。ADK 的 InMemorySessionService, DatabaseSessionService和 VertexAiSessionService 将被拒绝并返回 TypeError，因为会话状态保存在 LangGraph 检查点中。
原生 LangGraph 中断：该包装器不公开 LangGraph 的 interrupt or Command(resume=...) 机制。基于 LongRunningFunctionTool 构建的人机交互流程遵循 ADK 自身的模式：工具返回状态（如 pending_approval），代理进行回复，后续轮次解决待处理的调用。

项目布局

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

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

langgraph.json 指向导出的符号：

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

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:

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:

langgraph deploy --name my-adk-agent

将其部署到 LangSmith。环境配置、部署类型和版本管理，请参阅部署到云端。关于自托管设置，请参阅自托管部署.

启用追踪

wrap() 调用 langsmith.integrations.google_adk.configure_google_adk() 只要启用 LangSmith 追踪就会自动进行，因此您只需在部署上设置环境变量：

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 图，该图可从模块导出并由代理服务器提供服务。

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

Returns: A Pregel 其名称为 runner.app_name.

Raises: TypeError if runner.session_service 不是 LangsmithSessionService.

If runner.agent 定义了一个 output_key，该键的值也会在图的输出上公开，除了 messages。这使得 ADK 结构化输出代理（output_schema=..., output_key=...）能够与 Studio 和 /runs/wait response.

`LangsmithSessionService`

A google.adk.sessions.BaseSessionService 实现配合使用，后者由代理服务器的检查点存储提供支持。包装器自动管理会话生命周期。它在线程第一轮时创建会话，在后续轮次中从检查点加载会话，并在运行完成时将更新后的会话写回。

每个 Runner:

session_service = LangsmithSessionService()

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

`ADKInput`

包装代理的默认输入 schema。

字段	类型	描述
`messages`	`list[AnyMessage]`	（必填）对话消息；包装器将 `messages[-1].content` 作为新的用户消息发送给 ADK 运行器。
`state_delta`	`dict[str, Any] \	None`

`ADKOutput`

包装代理的默认输出 schema。

字段	类型	描述
`messages`	`list[AnyMessage]`	代理的响应消息，通过 LangGraph 的 `add_messages` reducer 添加到对话线程中。

将 messages 作为类型化字段（而不是普通 dict）使 Studio 能够将图检测为聊天兼容并启用聊天模式开关。

工作原理

当运行到达时：

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

This means ADK's own session/state semantics are preserved end-to-end while the deployment gets the standard Agent Server features: durable runs, streaming, multi-thread persistence, and tracing.