294 lines
11 KiB
Python
294 lines
11 KiB
Python
"""
|
||
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✅ 所有高级功能定义完成！") # 成功信息
|