""" Google ADK 高级主题完整示例 展示安全、自定义 Agent、多模型等高级功能 对应教程：第10章 - 高级主题 """ # 导入 ADK 核心模块 from google.adk.agents import ( # 导入所有 Agent 类型 Agent, # LLM Agent LlmAgent, # LLM Agent（完整名） BaseAgent, # 基础 Agent ) from google.adk.events import Event, EventActions # 导入事件类型 from google.adk.agents.invocation_context import InvocationContext # 导入上下文 from google.adk.models.lite_llm import LiteLlm # LiteLLM 适配器 # 导入辅助模块 from typing import AsyncGenerator # 异步生成器 import re # 正则表达式 import random # 随机数 import asyncio # 异步编程 # ======================================== # 示例一：输入安全过滤 # ======================================== async def input_security_filter(cb_ctx, inv_ctx): """ 输入安全过滤回调 检测注入攻击和敏感信息 Args: cb_ctx: 回调上下文 inv_ctx: 调用上下文 """ # 获取用户消息 events = inv_ctx.session.events # 获取会话事件 user_msgs = [ # 筛选用户消息 e for e in events if hasattr(e, 'content') and e.content.role == 'user' ] if not user_msgs: # 如果没有用户消息 return # 直接返回 text = user_msgs[-1].content.parts[0].text # 提取文本 # 检测注入攻击 injection_patterns = [ # 注入模式 r"忽略.*指令", # 中文注入 r"ignore.*instruction", # 英文注入 r"你现在是", # 角色切换 r"system\s*:", # 系统提示伪造 ] for pattern in injection_patterns: # 遍历模式 if re.search(pattern, text, re.IGNORECASE): # 如果匹配 print(f"[安全] ⚠️ 检测到可能的注入攻击") # 记录警告 break # 跳出循环 # 检测敏感信息 sensitive_patterns = [ # 敏感信息模式 r"\b\d{3}-\d{2}-\d{4}\b", # SSN r"\b\d{16}\b", # 信用卡号 ] for pattern in sensitive_patterns: # 遍历模式 if re.search(pattern, text): # 如果匹配 print(f"[安全] ⚠️ 检测到可能的敏感信息") # 记录警告 break # 跳出循环 # 检查输入长度 if len(text) > 5000: # 如果过长 print(f"[安全] ⚠️ 输入过长: {len(text)} 字符") # 记录警告 # ======================================== # 示例二：自定义 Agent — 规则引擎 # ======================================== class RuleEngineAgent(BaseAgent): """ 规则引擎 Agent 基于预定义规则处理请求，不使用 LLM """ def __init__(self, rules: dict, **kwargs): """ 初始化规则引擎 Args: rules: 规则字典 {匹配模式: 响应} """ super().__init__(**kwargs) # 调用父类初始化 self.rules = rules # 保存规则 async def _run_async_impl( self, ctx: InvocationContext, # 调用上下文 ) -> AsyncGenerator[Event, None]: # 返回事件生成器 """执行规则匹配""" # 获取用户消息 events = ctx.session.events # 获取事件 user_events = [ # 筛选用户消息 e for e in events if hasattr(e, 'content') and e.content.role == 'user' ] if not user_events: # 如果没有消息 yield Event( # 生成提示 author=self.name, content="请输入您的问题。", ) return # 退出 text = user_events[-1].content.parts[0].text # 提取文本 # 匹配规则 for pattern, response in self.rules.items(): # 遍历规则 if pattern.lower() in text.lower(): # 如果匹配 yield Event( # 生成响应 author=self.name, content=response, ) return # 退出 # 没有匹配的规则 yield Event( # 生成默认响应 author=self.name, content="抱歉，我无法处理您的请求。请尝试其他问题。", ) # ======================================== # 示例三：自定义 Agent — 随机路由 # ======================================== class RandomRouter(BaseAgent): """ 随机路由 Agent 随机选择一个子 Agent 处理请求 """ async def _run_async_impl( self, ctx: InvocationContext, # 调用上下文 ) -> AsyncGenerator[Event, None]: # 返回事件生成器 """随机选择子 Agent""" sub_agents = self.sub_agents # 获取子 Agent if not sub_agents: # 如果没有子 Agent yield Event( # 生成错误 author=self.name, content="没有可用的子 Agent。", ) return # 退出 chosen = random.choice(sub_agents) # 随机选择 print(f"[随机路由] 选择了 {chosen.name}") # 打印选择 yield Event( # 生成委派事件 author=self.name, actions=EventActions( transfer_to_agent=chosen.name, # 委派 ), ) # ======================================== # 示例四：多模型混合策略 # ======================================== # 简单任务：使用快速模型 simple_agent = LlmAgent( name="SimpleAgent", # 简单任务 Agent model="gemini-2.0-flash", # 快速模型 description="处理简单的问答和翻译。", # 描述 instruction="快速准确地回答简单问题。", # 指令 ) # 复杂推理：使用强力模型 reasoning_agent = LlmAgent( name="ReasoningAgent", # 推理 Agent model="gemini-2.5-pro-preview-05-06", # 强力模型 description="处理需要深度推理的复杂任务。", # 描述 instruction="仔细分析，给出深入的推理过程。", # 指令 ) # 代码生成：使用 DeepSeek code_agent = LlmAgent( name="CodeAgent", # 代码 Agent model=LiteLlm( # 使用 LiteLLM model="deepseek/deepseek-coder", # DeepSeek Coder api_key="your_api_key", # API Key ), description="处理代码生成和调试。", # 描述 instruction="编写高质量的代码。", # 指令 ) # 协调器：使用快速模型 multi_model_coordinator = LlmAgent( name="MultiModelCoordinator", # 多模型协调器 model="gemini-2.0-flash", # 快速模型 instruction=( # 指令 "根据任务类型分配给合适的 Agent：\n" "- 简单问答 → SimpleAgent\n" "- 复杂推理 → ReasoningAgent\n" "- 代码相关 → CodeAgent" ), sub_agents=[simple_agent, reasoning_agent, code_agent], # 子 Agent ) # ======================================== # 示例五：完整实战 — 智能客服系统 # ======================================== def search_faq(query: str) -> dict: """搜索 FAQ""" faq_db = { # FAQ 数据库 "退款": "退款将在3-5个工作日内处理完成。", "配送": "标准配送3-5天，加急配送1-2天。", "退换货": "7天无理由退换货，请保持商品完好。", } for key, value in faq_db.items(): # 遍历 FAQ if key in query: # 如果匹配 return {"status": "success", "answer": value} return {"status": "not_found"} def create_ticket(category: str, description: str) -> dict: """创建工单""" ticket_id = f"TK{hash(description) % 10000:04d}" # 生成工单号 return { # 返回结果 "status": "success", "ticket_id": ticket_id, "message": f"工单 {ticket_id} 已创建。", } # FAQ Agent faq_agent = LlmAgent( name="FAQBot", # FAQ 机器人 model="gemini-2.0-flash", # 模型 description="处理常见问题。", # 描述 instruction="使用 search_faq 工具查找答案。", # 指令 tools=[search_faq], # FAQ 工具 ) # 工单 Agent ticket_agent = LlmAgent( name="TicketBot", # 工单机器人 model="gemini-2.0-flash", # 模型 description="创建客户工单。", # 描述 instruction="使用 create_ticket 创建工单。", # 指令 tools=[create_ticket], # 工单工具 ) # 客服协调器 customer_service = LlmAgent( name="CustomerService", # 客服系统 model="gemini-2.0-flash", # 模型 description="智能客服主协调器。", # 描述 instruction=( # 指令 "你是智能客服协调器。\n" "常见问题 → FAQBot\n" "无法解决 → TicketBot\n" "使用 transfer_to_agent 委派任务。" ), sub_agents=[faq_agent, ticket_agent], # 子 Agent before_model_callback=input_security_filter, # 安全过滤 ) # ======================================== # 打印信息 # ======================================== if __name__ == "__main__": print("=" * 60) # 分隔线 print("Google ADK 高级主题示例") # 标题 print("=" * 60) # 分隔线 print("\n🔒 安全过滤回调: input_security_filter") # 安全 print("🤖 自定义 Agent: RuleEngineAgent, RandomRouter") # 自定义 print("🔀 多模型策略: SimpleAgent + ReasoningAgent + CodeAgent") # 多模型 print("🏢 实战项目: 智能客服系统 (CustomerService)") # 实战 print("\n✅ 所有高级功能定义完成！") # 成功信息