Google ADK：官方 Agent 开发套件

from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# 定义 Agent
root_agent = Agent(
    name="assistant",
    model="gemini-2.0-flash",
    description="一个通用助手",
    instruction="你是一个有帮助的助手，用中文回答所有问题。",
)

# 初始化 Runner
session_service = InMemorySessionService()
runner = Runner(agent=root_agent, session_service=session_service, app_name="demo")

# 创建 Session
session = session_service.create_session(app_name="demo", user_id="user1")

# 发送消息
response = runner.run(
    user_id="user1",
    session_id=session.id,
    new_message=types.Content(
        role="user",
        parts=[types.Part(text="介绍一下 Google ADK")]
    ),
)

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

def get_weather(city: str) -> dict:
    """获取指定城市的当前天气信息。

    Args:
        city: 城市名称，例如：北京、上海、广州

    Returns:
        包含温度、天气状况的字典
    """
    data = {
        "北京": {"temperature": 25, "condition": "晴天"},
        "上海": {"temperature": 22, "condition": "多云"},
        "广州": {"temperature": 30, "condition": "小雨"},
    }
    return data.get(city, {"error": f"暂无 {city} 天气数据"})

def search_web(query: str) -> str:
    """搜索网络获取最新信息。

    Args:
        query: 搜索关键词

    Returns:
        搜索结果摘要
    """
    # 实际接入搜索 API
    return f"关于 '{query}' 的搜索结果：..."

# 把工具绑定到 Agent
weather_agent = Agent(
    name="weather_assistant",
    model="gemini-2.0-flash",
    instruction="你可以查询天气和搜索信息。",
    tools=[get_weather, search_web],
)

from google.adk.agents import SequentialAgent

pipeline = SequentialAgent(
    name="research_pipeline",
    sub_agents=[
        fetch_agent,    # 第一步：获取数据
        analyze_agent,  # 第二步：分析
        report_agent,   # 第三步：生成报告
    ],
    description="顺序执行三步研究流程",
)

from google.adk.agents import ParallelAgent

parallel = ParallelAgent(
    name="parallel_research",
    sub_agents=[
        web_search_agent,
        db_search_agent,
        api_agent,
    ],
    description="并行从三个来源搜集信息",
)

from google.adk.agents import Agent

router = Agent(
    name="router",
    model="gemini-2.0-flash",
    instruction="""根据用户请求，转交给合适的专家 Agent：
- 技术问题 -> tech_agent
- 账单问题 -> billing_agent
- 投诉 -> support_agent""",
    sub_agents=[tech_agent, billing_agent, support_agent],
)

from google.adk.tools import google_search, code_execution

# Google 搜索（grounding）
search_agent = Agent(
    name="search_agent",
    model="gemini-2.0-flash",
    tools=[google_search],  # 使用 Google 搜索的真实结果
)

# 代码执行
code_agent = Agent(
    name="code_agent",
    model="gemini-2.0-flash",
    tools=[code_execution],  # 在沙箱中执行 Python 代码
)

async for event in runner.run_async(
    user_id="user1",
    session_id=session.id,
    new_message=types.Content(role="user", parts=[types.Part(text="写一首诗")]),
):
    if hasattr(event, "content") and event.content:
        for part in event.content.parts:
            if hasattr(part, "text"):
                print(part.text, end="", flush=True)

from google.adk.runners import VertexAIRunner

# 直接部署到 Vertex AI Agent Engine
runner = VertexAIRunner(
    agent=root_agent,
    project="your-gcp-project",
    location="us-central1",
)