Skip to main content
本指南将向您展示如何在 LangSmith 中追踪 Google Agent Development Kit (ADK) 代理。您将为 ADK 应用配置自动追踪，以捕获代理调用、工具调用和 LLM 交互。

​
安装

使用您偏好的包管理器安装所需的包： 
uv add langsmith[google-adk]

​
设置

设置您的 API 密钥： 
export LANGSMITH_TRACING=true
export LANGSMITH_ENDPOINT=https://api.smith.langchain.com
export LANGSMITH_API_KEY=<your_langsmith_api_key>
export LANGSMITH_PROJECT=<your_langsmith_project>

export GOOGLE_API_KEY=<your_google_api_key>
要创建 Google API 密钥，请参阅 Google AI Studio。

​
配置追踪

要追踪 ADK 代理，请使用 LangSmith SDK 中的 configure_google_adk() 函数。在创建任何 ADK 代理之前，在应用程序启动时调用此函数一次： 
from langsmith.integrations.google_adk import configure_google_adk

configure_google_adk(
    project_name="my-adk-project",  # 可选：默认为 LANGSMITH_PROJECT 环境变量
)
该函数接受以下可选参数：
  • project_name：用于发送追踪数据的 LangSmith 项目。默认为 LANGSMITH_PROJECT 环境变量。
  • name：根追踪的名称。默认为 "google_adk.session"。
  • metadata：用于附加上下文的键值对字典。
  • tags：用于分类追踪的字符串列表。

​
示例

此示例创建一个带有工具的天气代理，然后在启用追踪的情况下运行它： 
import asyncio

from dotenv import load_dotenv  # 可选
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from langsmith.integrations.google_adk import configure_google_adk

load_dotenv()  # 可选


async def main():
    # 配置 LangSmith 追踪
    configure_google_adk()

    # 定义一个工具
    def get_weather(city: str) -> dict:
        """获取城市的天气信息。"""
        return {"city": city, "temperature": "72°F", "conditions": "Sunny"}

    # 创建代理
    agent = Agent(
        name="weather_agent",
        model="gemini-2.0-flash",
        description="提供天气信息。",
        instruction="使用 get_weather 工具回答天气问题。",
        tools=[get_weather],
    )

    # 设置会话和运行器
    session_service = InMemorySessionService()
    session = await session_service.create_session(
        app_name="weather_app",
        user_id="user_123",
        session_id="session_456",
    )

    runner = Runner(
        agent=agent,
        app_name="weather_app",
        session_service=session_service,
    )

    # 运行代理
    async for event in runner.run_async(
        user_id="user_123",
        session_id=session.id,
        new_message=types.Content(
            role="user",
            parts=[types.Part(text="旧金山的天气怎么样？")],
        ),
    ):
        if event.is_final_response():
            print(event.content.parts[0].text)


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

​
在 LangSmith 中查看追踪

运行应用程序后，您可以在 LangSmith UI 中查看追踪，其中包含：
  • 代理调用：通过 ADK 代理的完整流程
  • 工具调用：代理进行的单个函数调用
  • LLM 交互：来自 Gemini 模型的请求和响应
  • 多代理工作流：来自顺序和并行代理组合的追踪

​
自定义元数据和标签

在配置追踪时添加元数据和标签，以分类和过滤追踪： 
from langsmith.integrations.google_adk import configure_google_adk

configure_google_adk(
    project_name="production-agents",
    metadata={
        "environment": "production",
        "team": "ml-platform",
    },
    tags=["adk", "weather", "v2"],
)

​
多代理工作流

该集成会自动追踪多代理工作流，包括顺序和并行代理组合： 
import asyncio

from dotenv import load_dotenv  # 可选
from google.adk.agents import Agent, SequentialAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from langsmith.integrations.google_adk import configure_google_adk

load_dotenv()  # 可选


async def main():
    # 配置 LangSmith 追踪
    # 默认情况下，追踪数据会发送到 LANGSMITH_PROJECT 环境变量指定的项目。
    # 可通过传递 project_name="my-project" 来覆盖。
    configure_google_adk()

    # 创建子代理
    translator = Agent(
        name="translator",
        model="gemini-2.0-flash",
        description="将文本翻译成英语。",
    )

    summarizer = Agent(
        name="summarizer",
        model="gemini-2.0-flash",
        description="简洁地总结文本。",
    )

    # 创建一个按顺序运行子代理的顺序代理
    pipeline = SequentialAgent(
        name="translate_and_summarize",
        sub_agents=[translator, summarizer],
        description="先翻译文本，然后进行总结。",
    )

    # 设置并运行
    session_service = InMemorySessionService()
    session = await session_service.create_session(
        app_name="pipeline_app",
        user_id="user_123",
        session_id="session_456",
    )

    runner = Runner(
        agent=pipeline,
        app_name="pipeline_app",
        session_service=session_service,
    )

    events = runner.run_async(
        user_id="user_123",
        session_id=session.id,
        new_message=types.Content(
            role="user",
            parts=[types.Part(text="Quelle est la plus haute tour de Paris?")],
        ),
    )

    async for event in events:
        if event.is_final_response():
            print(event.content.parts[0].text)


if __name__ == "__main__":
    asyncio.run(main())
将这些文档连接到 Claude、VSCode 等，通过 MCP 获取实时答案。
在 GitHub 上编辑此页面 或 提交问题。