追踪 Google ADK 应用

本指南介绍如何在 LangSmith 中追踪 Google Agent Development Kit (ADK) 智能体。你将为自己的 ADK 应用配置自动追踪，以捕获智能体调用、工具调用和 LLM 交互。

NOTE

本指南涵盖 ADK 的标准执行路径（文本智能体、工具和多智能体工作流）。要追踪使用 ADK Runner.run_live 流式循环的 Gemini Live 语音智能体，请参阅追踪 Gemini Live 应用。

安装

使用你偏好的包管理器安装所需依赖包：

uv

uv add "langsmith[google-adk]"

pip

pip install langsmith[google-adk]

设置

设置你的 API key：

shell

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>

.env

LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT=https://api.smith.langchain.com
LANGSMITH_API_KEY=<your_langsmith_api_key>
LANGSMITH_PROJECT=<your_langsmith_project>

GOOGLE_API_KEY=<your_google_api_key>

要创建 Google API key，请参阅 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",  # Optional: defaults to LANGSMITH_PROJECT env var
)

该函数接受以下可选参数：

• project_name：要向其发送追踪的 LangSmith 项目。默认为 LANGSMITH_PROJECT 环境变量。
• name：根追踪的名称。默认为 "google_adk.session"。
• metadata：用于提供附加上下文的键值对字典。
• tags：用于对追踪进行分类的字符串列表。

示例

以下示例创建一个带工具的天气智能体，然后在启用追踪的情况下运行它：

import asyncio

from dotenv import load_dotenv  # Optional
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()  # Optional

async def main():
    # Configure LangSmith tracing
    configure_google_adk()

    # Define a tool
    def get_weather(city: str) -> dict:
        """Get weather for a city."""
        return {"city": city, "temperature": "72°F", "conditions": "Sunny"}

    # Create the agent
    agent = Agent(
        name="weather_agent",
        model="gemini-2.5-flash",
        description="Provides weather information.",
        instruction="Use the get_weather tool to answer weather questions.",
        tools=[get_weather],
    )

    # Set up session and runner
    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,
    )

    # Run the agent
    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="What's the weather in San Francisco?")],
        ),
    ):
        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  # Optional
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()  # Optional

async def main():
    # Configure LangSmith tracing
    # Traces go to LANGSMITH_PROJECT env var by default.
    # Pass project_name="my-project" to override.
    configure_google_adk()

    # Create sub-agents
    translator = Agent(
        name="translator",
        model="gemini-2.5-flash",
        description="Translates text to English.",
    )

    summarizer = Agent(
        name="summarizer",
        model="gemini-2.5-flash",
        description="Summarizes text concisely.",
    )

    # Create a sequential agent that runs sub-agents in order
    pipeline = SequentialAgent(
        name="translate_and_summarize",
        sub_agents=[translator, summarizer],
        description="Translates text then summarizes it.",
    )

    # Set up and run
    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())

---