23 KiB
23 KiB
第10章：高级主题
📌 本章目标
- 掌握 ADK 的安全最佳实践
- 了解 A2A（Agent-to-Agent）协议
- 学习多模型混合使用策略
- 掌握自定义 Agent 的开发方法
- 了解性能优化技巧
10.1 安全最佳实践
10.1.1 输入安全
"""
输入安全：防止注入攻击和不当输入
"""
from google.adk.agents import Agent # 导入 Agent 类
from google.adk.agents.invocation_context import InvocationContext # 导入上下文
import re # 导入正则表达式模块
# ========================================
# 输入过滤器
# ========================================
async def input_filter_callback(
callback_context, # 回调上下文
invocation_context: InvocationContext, # 调用上下文
):
"""
输入安全过滤回调
在模型调用前检查用户输入的安全性
"""
# 获取用户消息
# 从调用上下文中获取最近的用户消息
events = invocation_context.session.events # 获取会话事件
user_messages = [ # 筛选用户消息
e for e in events
if hasattr(e, 'content') and e.content.role == 'user'
]
if not user_messages: # 如果没有用户消息
return # 直接返回
last_message = user_messages[-1] # 获取最后一条用户消息
user_text = last_message.content.parts[0].text # 提取文本
# ========================================
# 检查1：提示注入检测
# ========================================
injection_patterns = [ # 注入攻击模式
r"忽略.*指令", # "忽略之前的指令"
r"ignore.*instruction", # 英文注入
r"你现在是", # 角色切换
r"pretend.*you are", # 英文角色切换
r"system\s*:", # 系统提示伪造
]
for pattern in injection_patterns: # 遍历模式
if re.search(pattern, user_text, re.IGNORECASE): # 如果匹配
print(f"[安全] 检测到可能的注入攻击: {pattern}") # 记录警告
# 在实际应用中，可以拒绝处理或返回安全提示
# ========================================
# 检查2：输入长度限制
# ========================================
MAX_INPUT_LENGTH = 5000 # 最大输入长度
if len(user_text) > MAX_INPUT_LENGTH: # 如果超过限制
print(f"[安全] 输入过长: {len(user_text)} > {MAX_INPUT_LENGTH}") # 记录警告
# ========================================
# 检查3：敏感信息检测
# ========================================
sensitive_patterns = [ # 敏感信息模式
r"\b\d{3}-\d{2}-\d{4}\b", # SSN 格式
r"\b\d{16}\b", # 信用卡号格式
r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", # 邮箱
]
for pattern in sensitive_patterns: # 遍历模式
if re.search(pattern, user_text): # 如果匹配
print(f"[安全] 检测到可能的敏感信息") # 记录警告
# 不要在日志中记录实际的敏感信息
# ========================================
# 创建安全 Agent
# ========================================
safe_agent = Agent(
name="safe_agent", # Agent 名称
model="gemini-2.0-flash", # 模型
instruction="你是一个安全的助手。遵循安全准则。", # 指令
before_model_callback=input_filter_callback, # 🔑 安全过滤回调
)
10.1.2 输出安全
"""
输出安全：过滤不当输出
"""
async def output_filter_callback(
callback_context, # 回调上下文
invocation_context: InvocationContext, # 调用上下文
):
"""
输出安全过滤回调
在模型返回响应后检查输出内容
"""
response = callback_context.response # 获取模型响应
if not response: # 如果没有响应
return # 直接返回
# 检查响应中是否有文本内容
if response.candidates: # 如果有候选响应
for candidate in response.candidates: # 遍历候选
if candidate.content: # 如果有内容
text = candidate.content.parts[0].text # 提取文本
# 检查不当内容
forbidden_words = [ # 禁止词列表
"暴力", "违法", # 中文禁止词
]
for word in forbidden_words: # 遍历禁止词
if word in text: # 如果包含禁止词
print(f"[安全] 输出包含不当内容: {word}") # 记录警告
agent = Agent(
name="output_safe_agent", # Agent 名称
model="gemini-2.0-flash", # 模型
instruction="你是一个安全的助手。", # 指令
after_model_callback=output_filter_callback, # 🔑 输出过滤回调
)
10.2 A2A 协议（Agent-to-Agent）
10.2.1 A2A 概述
A2A（Agent-to-Agent）协议是 Google 提出的开放标准，用于实现不同 Agent 系统之间的远程通信。
┌──────────────┐ ┌──────────────┐
│ Agent A │ │ Agent B │
│ (ADK) │◄────────►│ (其他框架) │
│ │ A2A 协议 │ │
│ - 发送任务 │ │ - 接收任务 │
│ - 接收结果 │ │ - 返回结果 │
└──────────────┘ └──────────────┘
10.2.2 A2A 集成示例
"""
A2A 协议集成示例
让 ADK Agent 与远程 Agent 通信
"""
from google.adk.agents import Agent # 导入 Agent 类
from google.adk.tools import AgentTool # 导入 AgentTool
# ========================================
# 创建远程 Agent 客户端
# ========================================
# 假设有一个远程 Agent 服务运行在 http://remote-agent:3000
# 可以通过 A2A 协议与之通信
# 在实际使用中，需要安装 A2A 客户端库
# pip install a2a-sdk
# 创建远程 Agent 的代理
# remote_agent_proxy = create_remote_agent_proxy(
# url="http://remote-agent:3000", # 远程 Agent 地址
# agent_name="RemoteExpert", # 远程 Agent 名称
# )
# 将远程 Agent 包装为工具
# remote_tool = AgentTool(agent=remote_agent_proxy)
# ========================================
# 在本地 Agent 中使用远程 Agent
# ========================================
local_agent = Agent(
name="local_agent", # 本地 Agent 名称
model="gemini-2.0-flash", # 模型
instruction=( # 指令
"你是一个协调助手。\n"
"对于本地可以处理的问题，直接回答。\n"
"对于需要专业知识的问题，使用远程 Agent 工具。"
),
# tools=[remote_tool], # 注册远程 Agent 工具
)
10.3 自定义 Agent 开发
10.3.1 继承 BaseAgent
"""
自定义 Agent 开发
通过继承 BaseAgent 实现独特的 Agent 逻辑
"""
from google.adk.agents import BaseAgent # 导入基础 Agent 类
from google.adk.agents.invocation_context import InvocationContext # 导入调用上下文
from google.adk.events import Event, EventActions # 导入事件类型
from typing import AsyncGenerator # 导入异步生成器
import json # 导入 JSON 模块
import random # 导入随机数模块
# ========================================
# 示例一：随机决策 Agent
# ========================================
class RandomDecisionAgent(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 # 退出
# 随机选择一个子 Agent
chosen = random.choice(sub_agents) # 随机选择
print(f"[随机] 选择了 {chosen.name}") # 打印选择结果
# 将选择结果保存到状态
ctx.session.state["temp:chosen_agent"] = chosen.name # 保存选择
# 生成事件，指示委派给选中的 Agent
yield Event( # 生成委派事件
author=self.name, # 作者
actions=EventActions( # 事件动作
transfer_to_agent=chosen.name, # 委派给选中的 Agent
),
)
# ========================================
# 示例二：规则引擎 Agent
# ========================================
class RuleEngineAgent(BaseAgent):
"""
规则引擎 Agent
基于预定义规则处理请求，不使用 LLM
"""
def __init__(self, rules: dict, **kwargs):
"""
初始化规则引擎
Args:
rules: 规则字典，key 为匹配模式，value 为响应
"""
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 # 退出
user_text = user_events[-1].content.parts[0].text # 提取文本
# 匹配规则
matched_response = None # 匹配的响应
for pattern, response in self.rules.items(): # 遍历规则
if pattern.lower() in user_text.lower(): # 如果匹配
matched_response = response # 保存响应
break # 跳出循环
if matched_response: # 如果匹配到规则
yield Event( # 生成响应事件
author=self.name, # 作者
content=matched_response, # 响应内容
)
else: # 如果没有匹配
yield Event( # 生成默认响应
author=self.name,
content="抱歉，我无法处理您的请求。请尝试其他问题。",
)
# ========================================
# 使用自定义 Agent
# ========================================
# 创建规则引擎 Agent
faq_agent = RuleEngineAgent(
name="FAQBot", # Agent 名称
rules={ # 规则字典
"价格": "我们的产品价格请参考官网定价页面。", # 价格相关
"地址": "我们的地址是北京市海淀区xxx路xxx号。", # 地址相关
"电话": "客服电话：400-xxx-xxxx。", # 电话相关
"营业时间": "营业时间：周一至周五 9:00-18:00。", # 营业时间
},
)
# 创建随机决策 Agent
random_agent = RandomDecisionAgent(
name="RandomRouter", # Agent 名称
sub_agents=[ # 子 Agent 列表
faq_agent, # FAQ Agent
],
)
10.4 多模型混合策略
"""
多模型混合使用
根据任务类型选择不同的模型
"""
from google.adk.agents import LlmAgent # 导入 LlmAgent
from google.adk.models.lite_llm import LiteLlm # 导入 LiteLLM
# ========================================
# 策略一：不同 Agent 使用不同模型
# ========================================
# 简单任务使用快速模型
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="仔细分析问题，给出深入的推理过程。", # 指令
)
# 代码生成使用专用模型
code_agent = LlmAgent(
name="CodeAgent", # 代码 Agent
model=LiteLlm( # 使用 LiteLLM
model="deepseek/deepseek-coder", # DeepSeek Coder
api_key="your_api_key", # API Key
),
description="处理代码生成和调试任务。", # 描述
instruction="编写高质量的代码。", # 指令
)
# 协调器使用快速模型
coordinator = LlmAgent(
name="Coordinator", # 协调器
model="gemini-2.0-flash", # 快速模型
instruction=( # 指令
"根据任务复杂度分配给合适的 Agent：\n"
"- 简单问答 → SimpleAgent\n"
"- 复杂推理 → ReasoningAgent\n"
"- 代码相关 → CodeAgent"
),
sub_agents=[simple_agent, reasoning_agent, code_agent], # 子 Agent
)
10.5 性能优化技巧
10.5.1 指令优化
"""
指令优化技巧
减少 token 使用，提高响应速度
"""
# ❌ 冗长的指令（浪费 token）
verbose_instruction = """
你是一个非常专业的、经验丰富的、知识渊博的助手。
你擅长回答各种各样的问题，包括但不限于技术问题、
生活问题、工作问题、学习问题等等。
当用户问你问题时，你应该：
1. 首先理解用户的问题
2. 然后分析问题
3. 最后给出答案
请始终使用中文回答。
"""
# ✅ 精简的指令（节省 token）
concise_instruction = """
你是专业助手。用中文简洁回答。
"""
# ✅ 使用结构化指令
structured_instruction = """
## 角色
技术助手
## 规则
1. 使用中文
2. 简洁回答（<200字）
3. 不确定时说"不确定"
"""
10.5.2 工具优化
"""
工具优化技巧
减少不必要的工具调用
"""
# ❌ 工具描述过于详细（浪费 token）
def bad_tool(query: str) -> dict:
"""
这是一个非常详细的工具描述，包含了大量的信息，
但是大部分信息对 LLM 来说是不必要的。
LLM 需要读取所有这些文本来理解工具的功能，
这会消耗大量的 token 并降低响应速度。
"""
return {"result": "done"}
# ✅ 精简的工具描述
def good_tool(query: str) -> dict:
"""搜索知识库并返回相关文档片段。"""
return {"result": "done"}
# ✅ 合并相似工具
# 如果有多个功能相似的工具，考虑合并
def unified_search(
query: str, # 搜索关键词
search_type: str = "all", # 搜索类型：all/doc/faq
) -> dict:
"""
统一搜索工具
Args:
query: 搜索关键词
search_type: 搜索类型，可选 "all"(全部)、"doc"(文档)、"faq"(FAQ)
"""
return {"status": "success", "results": []}
10.6 完整实战项目：智能客服系统
"""
完整实战项目：多智能体客服系统
整合本教程所有知识点
"""
from google.adk.agents import ( # 导入所有 Agent 类型
Agent, # LLM Agent
LlmAgent, # LLM Agent（完整名）
SequentialAgent, # 顺序工作流
ParallelAgent, # 并行工作流
LoopAgent, # 循环工作流
BaseAgent, # 基础 Agent
)
from google.adk.events import Event, EventActions # 导入事件类型
from google.adk.agents.invocation_context import InvocationContext # 导入上下文
from google.adk.tools import google_search # 导入 Google 搜索
from typing import AsyncGenerator # 导入异步生成器
import asyncio # 导入异步库
# ========================================
# 第一步：定义工具
# ========================================
def search_faq(query: str) -> dict:
"""搜索常见问题解答"""
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", "answer": "未找到相关FAQ。"} # 未找到
def create_ticket(
category: str, # 工单类别
description: str, # 问题描述
priority: str = "normal", # 优先级
) -> dict:
"""创建客户工单"""
ticket_id = f"TK{len(description)}" # 生成工单号
return { # 返回工单信息
"status": "success",
"ticket_id": ticket_id,
"message": f"工单 {ticket_id} 已创建，我们会尽快处理。",
}
# ========================================
# 第二步：定义子 Agent
# ========================================
# FAQ Agent：处理常见问题
faq_agent = LlmAgent(
name="FAQBot", # FAQ 机器人
model="gemini-2.0-flash", # 快速模型
description="处理常见问题，如退款、配送、退换货等。", # 描述
instruction="你是FAQ助手，使用 search_faq 工具查找答案。", # 指令
tools=[search_faq], # FAQ 搜索工具
)
# 搜索 Agent：处理需要联网查询的问题
search_agent = LlmAgent(
name="SearchBot", # 搜索机器人
model="gemini-2.0-flash", # 快速模型
description="处理需要搜索互联网的问题。", # 描述
instruction="使用 Google Search 搜索最新信息。", # 指令
tools=[google_search], # Google 搜索工具
)
# 工单 Agent：处理需要人工介入的问题
ticket_agent = LlmAgent(
name="TicketBot", # 工单机器人
model="gemini-2.0-flash", # 快速模型
description="创建客户工单，转交人工处理。", # 描述
instruction="收集用户问题描述，使用 create_ticket 创建工单。", # 指令
tools=[create_ticket], # 创建工单工具
)
# ========================================
# 第三步：定义协调器
# ========================================
coordinator = LlmAgent(
name="CustomerServiceCoordinator", # 客服协调器
model="gemini-2.0-flash", # 快速模型
description="智能客服主协调器，负责路由用户请求。", # 描述
instruction=( # 系统指令
"你是一个智能客服协调器。\n"
"根据用户问题类型，分配给合适的子 Agent：\n"
"- 退款、配送、退换货等常见问题 → FAQBot\n"
"- 需要最新信息的问题 → SearchBot\n"
"- 无法自动解决的问题 → TicketBot\n"
"使用 transfer_to_agent 进行任务分配。"
),
sub_agents=[ # 注册子 Agent
faq_agent, # FAQ 机器人
search_agent, # 搜索机器人
ticket_agent, # 工单机器人
],
)
# ========================================
# 第四步：添加监控回调
# ========================================
async def monitoring_callback(cb_ctx, inv_ctx):
"""监控回调：记录所有操作"""
agent_name = inv_ctx.agent.name # 获取 Agent 名称
print(f"[监控] Agent '{agent_name}' 正在处理请求") # 打印日志
coordinator.before_model_callback = monitoring_callback # 添加监控
# ========================================
# 第五步：设置 root_agent
# ========================================
root_agent = coordinator # 设置根 Agent
print("✅ 智能客服系统构建完成！") # 打印成功信息
print(f"根 Agent: {root_agent.name}") # 打印根 Agent 名称
print(f"子 Agent 数量: {len(root_agent.sub_agents)}") # 打印子 Agent 数量
📌 本章小结
- 输入安全：检测注入攻击、限制输入长度、过滤敏感信息
- 输出安全：过滤不当内容、记录审计日志
- A2A 协议：实现不同 Agent 系统间的远程通信
- 自定义 Agent：继承 BaseAgent 实现独特逻辑
- 多模型策略：根据任务类型选择合适的模型
- 性能优化：精简指令、合并工具、合理选择模型
🎉 教程完结
恭喜你完成了 Google ADK 完整教程的学习！
学习路径回顾
第01章 环境搭建 → 第02章 Hello World
↓
第03章 LLM Agent → 第04章 自定义工具
↓
第05章 多智能体 → 第06章 会话状态
↓
第07章 回调机制 → 第08章 评估
↓
第09章 部署 → 第10章 高级主题
后续学习建议
- 实践项目：基于教程知识，构建自己的 Agent 应用
- 阅读源码：深入研究 adk-python 源码
- 社区参与：关注 ADK 的更新和社区讨论
- 探索生态：了解 ADK 的 Java、Go 版本和 Web 版本