整体测试下来ADK这个框架相当完整

ADK 能力清单包含

1. 普通Agent
2. 普通tools
3. 模型执行前后做拦截，做内容修改
4. 内置代码执行工具和google搜索工具，这俩工具限制了google模型，用处不大
5. MultiAgents模式，支持分批某个Agent示例做规划工作
6. 工作流代理-顺序执行
7. 工作流代理-循环执行
8. 工作流代理-并行执行
9. 自定义代理实现代码
10. 多代理可以共享会话状态
11. 多代理可以实现委托转移
12. agenttool ，允许将一个agent 作为另一个agent的tool使用
13. 其他不详细列了

最佳时间代码列几个

最最常见的智能分派

from google.adk.agents import LlmAgent

# 子智能体：负责处理账单相关问题
billing_agent = LlmAgent(
    name="Billing",
    description="处理账单相关的咨询请求。"
)

# 子智能体：负责处理技术支持问题
support_agent = LlmAgent(
    name="Support",
    description="处理技术支持类的问题。"
)

# 主协调智能体：根据用户问题转交给对应的子智能体
coordinator = LlmAgent(
    name="HelpDeskCoordinator",
    model="gemini-2.0-flash",  # 指定使用的 LLM 模型
    instruction="""
你是一个帮助台协调助手。
- 如果用户的问题是关于支付/账单，请将请求转交给 Billing 智能体。
- 如果用户的问题是技术类问题，请将请求转交给 Support 智能体。
""",
    description="主帮助台协调器，用于路由用户请求。",
    sub_agents=[billing_agent, support_agent]  # 注册子智能体
    # allow_transfer=True 一般在 AutoFlow 中自动生效
)

当用户说：“我的支付失败了” → 主 Agent 应调用：
transfer_to_agent(agent_name='Billing')

当用户说：“我无法登录” → 主 Agent 应调用：
transfer_to_agent(agent_name='Support')

顺序管道模式

from google.adk.agents import SequentialAgent, LlmAgent

# 步骤 1：输入校验智能体
validator = LlmAgent(
    name="ValidateInput",
    instruction="对输入数据进行合法性校验。",
    output_key="validation_status"  # 校验结果保存到状态中
)

# 步骤 2：数据处理智能体
processor = LlmAgent(
    name="ProcessData",
    instruction="仅当 state['validation_status'] 为 'valid' 时，执行数据处理逻辑。",
    output_key="result"  # 处理结果保存到状态中
)

# 步骤 3：结果报告智能体
reporter = LlmAgent(
    name="ReportResult",
    instruction="从 state['result'] 中读取处理结果，并将其汇报输出。"
)

# 构建顺序执行的智能体流水线
data_pipeline = SequentialAgent(
    name="DataPipeline",  # 整体流程命名为 DataPipeline
    sub_agents=[validator, processor, reporter]  # 顺序执行这三个子 Agent
)

并行收集资料，最后合成

# Conceptual Code: Parallel Information Gathering
from google.adk.agents import SequentialAgent, ParallelAgent, LlmAgent

fetch_api1 = LlmAgent(name="API1Fetcher", instruction="Fetch data from API 1.", output_key="api1_data")
fetch_api2 = LlmAgent(name="API2Fetcher", instruction="Fetch data from API 2.", output_key="api2_data")

gather_concurrently = ParallelAgent(
    name="ConcurrentFetch",
    sub_agents=[fetch_api1, fetch_api2]
)

synthesizer = LlmAgent(
    name="Synthesizer",
    instruction="Combine results from state keys 'api1_data' and 'api2_data'."
)

overall_workflow = SequentialAgent(
    name="FetchAndSynthesize",
    sub_agents=[gather_concurrently, synthesizer] # Run parallel fetch, then synthesize
)
# fetch_api1 and fetch_api2 run concurrently, saving to state.
# synthesizer runs afterwards, reading state['api1_data'] and state['api2_data'].

任务拆解执行

# Conceptual Code: Hierarchical Research Task
from google.adk.agents import LlmAgent
from google.adk.tools import agent_tool

# Low-level tool-like agents
web_searcher = LlmAgent(name="WebSearch", description="Performs web searches for facts.")
summarizer = LlmAgent(name="Summarizer", description="Summarizes text.")

# Mid-level agent combining tools
research_assistant = LlmAgent(
    name="ResearchAssistant",
    model="gemini-2.0-flash",
    description="Finds and summarizes information on a topic.",
    tools=[agent_tool.AgentTool(agent=web_searcher), agent_tool.AgentTool(agent=summarizer)]
)

# High-level agent delegating research
report_writer = LlmAgent(
    name="ReportWriter",
    model="gemini-2.0-flash",
    instruction="Write a report on topic X. Use the ResearchAssistant to gather information.",
    tools=[agent_tool.AgentTool(agent=research_assistant)]
    # Alternatively, could use LLM Transfer if research_assistant is a sub_agent
)
# User interacts with ReportWriter.
# ReportWriter calls ResearchAssistant tool.
# ResearchAssistant calls WebSearch and Summarizer tools.
# Results flow back up.

增加人工确认环节

# Conceptual Code: Using a Tool for Human Approval
import config
config.load_config()
from google.adk.agents import LlmAgent, SequentialAgent
from google.adk.tools import FunctionTool
from google.adk.models.lite_llm import LiteLlm


# --- Assume external_approval_tool exists ---
# This tool would:
# 1. Take details (e.g., request_id, amount, reason).
# 2. Send these details to a human review system (e.g., via API).
# 3. Poll or wait for the human response (approved/rejected).
# 4. Return the human's decision.
# async def external_approval_tool(amount: float, reason: str) -> str: ...
async def external_approval_tool(amount: float, reason: str) -> str:
    return "rejected"  # For demonstration purposes, always return 'approved'

approval_tool = FunctionTool(func=external_approval_tool)

# Agent that prepares the request

prepare_request = LlmAgent(
    name="PrepareApproval",
    model=LiteLlm(model="deepseek/deepseek-chat"),
    instruction="""
你是一个审批准备助手。
请根据用户输入生成审批请求的两个字段：
1. 审批金额（amount）
2. 审批事由（reason）

请将这两个字段存入会话状态中，键名分别为 'approval_amount' 和 'approval_reason'。
""",
    description="生成审批申请所需的金额与事由。",
    output_key="approval_reason"
)

# Agent that calls the human approval tool
request_approval = LlmAgent(
    name="RequestHumanApproval",
    model=LiteLlm(model="deepseek/deepseek-chat"),
    instruction="Use the external_approval_tool with amount from state['approval_amount'] and reason from state['approval_reason'].",
    tools=[approval_tool],
    output_key="human_decision"
)

# Agent that proceeds based on human decision
process_decision = LlmAgent(
    name="ProcessDecision",
    model=LiteLlm(model="deepseek/deepseek-chat"),
    instruction="Check state key 'human_decision'. If 'approved', proceed. If 'rejected', inform user.",
    description = "根据人工审批结果决定后续响应"
)

approval_workflow = SequentialAgent(
    name="HumanApprovalWorkflow",
    sub_agents=[prepare_request, request_approval, process_decision])


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

# --- 应用基本配置 ---
APP_NAME = "doc_writing_app"       # 应用名称
USER_ID = "dev_user_01"            # 用户 ID
SESSION_ID = "session_01"          # 会话 ID

# --- 状态键常量 ---

# --- 3. 会话服务与执行器 ---
# 用于管理用户会话与执行整个智能体流程
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)
runner = Runner(
    agent=approval_workflow,
    app_name=APP_NAME,
    session_service=session_service
)
import logging
logging.basicConfig(level=logging.INFO)
# --- 4. 执行入口函数 ---
# 接收用户请求并打印最终结果
def call_agent(query):
    content = types.Content(role='user', parts=[types.Part(text=query)])

    events = runner.run(
        user_id=USER_ID,
        session_id=SESSION_ID,
        new_message=content
    )

    for event in events:
        if event.is_final_response():
            final_response = event.content.parts[0].text
            print(" 智能体最终输出结果：\n", final_response)

# 示例请求：让智能体编写加法相关代码
call_agent("购买电脑")

最后注意如果使用其他大模型，不用google的模型，要更新
.venv/Lib/site-packages/google/adk/models/lite_llm.py用版本库里的最新代码更新，不然有bug