pip install langchain langgraph google-cloud-aiplatform langchain-google-genai google-adk deprecated pydantic

# Copyright (c) 2025 Marco Fago
# https://www.linkedin.com/in/marco-fago/
#
# 此代码在MIT许可证下许可。
# 有关完整许可证文本，请参阅存储库中的LICENSE文件。

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableBranch

# --- 配置 ---
# 确保设置了API密钥环境变量（如GOOGLE_API_KEY）
try:
    llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", temperature=0)
    print(f"语言模型已初始化: {llm.model}")
except Exception as e:
    print(f"初始化语言模型时出错: {e}")
    llm = None

# --- 定义模拟子智能体处理器（相当于ADK sub_agents） ---
def booking_handler(request: str) -> str:
    """模拟预订智能体处理请求。"""
    print("\n--- 委托给预订处理器 ---")
    return f"预订处理器处理请求: '{request}'。结果: 模拟预订操作。"

def info_handler(request: str) -> str:
    """模拟信息智能体处理请求。"""
    print("\n--- 委托给信息处理器 ---")
    return f"信息处理器处理请求: '{request}'。结果: 模拟信息检索。"

def unclear_handler(request: str) -> str:
    """处理无法委托的请求。"""
    print("\n--- 处理不清楚的请求 ---")
    return f"协调器无法委托请求: '{request}'。请澄清。"

# --- 定义协调器路由器链（相当于ADK协调器的指令） ---
# 此链决定委托给哪个处理器。
coordinator_router_prompt = ChatPromptTemplate.from_messages([
    ("system", """分析用户的请求并确定应该处理它的专业处理器。
    - 如果请求与预订航班或酒店相关，
      输出'booker'。
    - 对于所有其他一般信息问题，输出'info'。
    - 如果请求不清楚或不符合任一类别，
      输出'unclear'。
    仅输出一个词: 'booker'、'info'或'unclear'。"""),
    ("user", "{request}")
])

if llm:
    coordinator_router_chain = coordinator_router_prompt | llm | StrOutputParser()

# --- 定义委托逻辑（相当于ADK基于sub_agents的自动流程） ---
# 使用RunnableBranch根据路由器链的输出进行路由。
# 为RunnableBranch定义分支
branches = {
    "booker": RunnablePassthrough.assign(output=lambda x: booking_handler(x['request']['request'])),
    "info": RunnablePassthrough.assign(output=lambda x: info_handler(x['request']['request'])),
    "unclear": RunnablePassthrough.assign(output=lambda x: unclear_handler(x['request']['request'])),
}

# 创建RunnableBranch。它获取路由器链的输出
# 并将原始输入（'request'）路由到相应的处理器。
delegation_branch = RunnableBranch(
    (lambda x: x['decision'].strip() == 'booker', branches["booker"]), # 添加了.strip()
    (lambda x: x['decision'].strip() == 'info', branches["info"]),     # 添加了.strip()
    branches["unclear"] # 'unclear'或任何其他输出的默认分支
)

# 将路由器链和委托分支组合成单个可运行对象
# 路由器链的输出（'decision'）与原始输入（'request'）一起传递
# 到delegation_branch。
coordinator_agent = {
    "decision": coordinator_router_chain,
    "request": RunnablePassthrough()
} | delegation_branch | (lambda x: x['output']) # 提取最终输出

# --- 示例用法 ---
def main():
    if not llm:
        print("\n由于LLM初始化失败，跳过执行。")
        return
    
    print("--- 使用预订请求运行 ---")
    request_a = "为我预订到伦敦的航班。"
    result_a = coordinator_agent.invoke({"request": request_a})
    print(f"最终结果A: {result_a}")
    
    print("\n--- 使用信息请求运行 ---")
    request_b = "意大利的首都是什么？"
    result_b = coordinator_agent.invoke({"request": request_b})
    print(f"最终结果B: {result_b}")
    
    print("\n--- 使用不清楚的请求运行 ---")
    request_c = "告诉我关于量子物理学。"
    result_c = coordinator_agent.invoke({"request": request_c})
    print(f"最终结果C: {result_c}")

if __name__ == "__main__":
    main()

# Copyright (c) 2025 Marco Fago
#
# 此代码在MIT许可证下许可。
# 有关完整许可证文本，请参阅存储库中的LICENSE文件。

import uuid
from typing import Dict, Any, Optional
from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.adk.tools import FunctionTool
from google.genai import types
from google.adk.events import Event

# --- 定义工具函数 ---
# 这些函数模拟专业智能体的行动。
def booking_handler(request: str) -> str:
    """
    处理航班和酒店的预订请求。
    参数:
        request: 用户的预订请求。
    返回:
        确认预订已处理的确认消息。
    """
    print("-------------------------- 预订处理器被调用 ----------------------------")
    return f"'{request}'的预订操作已被模拟。"

def info_handler(request: str) -> str:
    """
    处理一般信息请求。
    参数:
        request: 用户的问题。
    返回:
        表示信息请求已处理的消息。
    """
    print("-------------------------- 信息处理器被调用 ----------------------------")
    return f"'{request}'的信息请求。结果: 模拟信息检索。"

def unclear_handler(request: str) -> str:
    """处理无法委托的请求。"""
    return f"协调器无法委托请求: '{request}'。请澄清。"

# --- 从函数创建工具 ---
booking_tool = FunctionTool(booking_handler)
info_tool = FunctionTool(info_handler)

# 定义配备各自工具的专门子智能体
booking_agent = Agent(
    name="Booker",
    model="gemini-2.0-flash",
    description="通过调用预订工具处理所有航班和酒店预订请求的专门智能体。",
    tools=[booking_tool]
)

info_agent = Agent(
    name="Info",
    model="gemini-2.0-flash",
    description="通过调用信息工具提供一般信息和回答用户问题的专门智能体。",
    tools=[info_tool]
)

# 定义具有显式委托指令的父智能体
coordinator = Agent(
    name="Coordinator",
    model="gemini-2.0-flash",
    instruction=(
        "你是主协调器。你的唯一任务是分析 "
        "传入的用户请求 "
        "并将它们委托给适当的专业智能体。 "
        "不要尝试直接回答用户。\n"
        "- 对于与预订航班或酒店相关的任何请求， "
        "  委托给'Booker'智能体。\n"
        "- 对于所有其他一般信息问题，委托给'Info'智能体。"
    ),
    description="将用户请求路由到正确专业智能体的协调器。",
    # sub_agents的存在默认启用LLM驱动的委托（自动流程）。
    sub_agents=[booking_agent, info_agent]
)

# --- 执行逻辑 ---
async def run_coordinator(runner: InMemoryRunner, request: str):
    """使用给定请求运行协调器智能体并委托。"""
    print(f"\n--- 使用请求运行协调器: '{request}' ---")
    final_result = ""
    try:
        user_id = "user_123"
        session_id = str(uuid.uuid4())
        await runner.session_service.create_session(
            app_name=runner.app_name, user_id=user_id, session_id=session_id
        )
        for event in runner.run(
            user_id=user_id,
            session_id=session_id,
            new_message=types.Content(
                role='user',
                parts=[types.Part(text=request)]
            ),
        ):
            if event.is_final_response() and event.content:
                # 尝试直接从event.content获取文本
                # 避免迭代parts
                if hasattr(event.content, 'text') and event.content.text:
                    final_result = event.content.text
                elif event.content.parts:
                    # 后备: 迭代parts并提取文本（可能触发警告）
                    text_parts = [part.text for part in event.content.parts if part.text]
                    final_result = "".join(text_parts)
                # 假设循环应该在最终响应后中断
                break
        print(f"协调器最终响应: {final_result}")
        return final_result
    except Exception as e:
        print(f"处理您的请求时发生错误: {e}")
        return f"处理您的请求时发生错误: {e}"

async def main():
    """运行ADK示例的主函数。"""
    print("--- Google ADK路由示例（ADK自动流程样式） ---")
    print("注意: 这需要安装Google ADK并进行身份验证。")
    runner = InMemoryRunner(coordinator)
    
    # 示例用法
    result_a = await run_coordinator(runner, "为我在巴黎预订一家酒店。")
    print(f"最终输出A: {result_a}")
    
    result_b = await run_coordinator(runner, "世界上最高的山是什么？")
    print(f"最终输出B: {result_b}")
    
    result_c = await run_coordinator(runner, "告诉我一个随机事实。") # 应该去Info
    print(f"最终输出C: {result_c}")
    
    result_d = await run_coordinator(runner, "查找下个月到东京的航班。") # 应该去Booker
    print(f"最终输出D: {result_d}")

if __name__ == "__main__":
    import nest_asyncio
    nest_asyncio.apply()
    await main()