566 lines
20 KiB
Markdown
566 lines
20 KiB
Markdown
# 第07章：回调机制与事件系统
|
||
|
||
## 📌 本章目标
|
||
|
||
- 理解 ADK 的事件系统和工作原理
|
||
- 掌握四种回调函数的使用方法
|
||
- 学习如何通过回调实现自定义逻辑
|
||
- 了解 Human-in-the-Loop（人机协作）模式
|
||
- 掌握事件流的处理和过滤
|
||
|
||
---
|
||
|
||
## 7.1 事件系统概述
|
||
|
||
### 7.1.1 什么是事件（Event）？
|
||
|
||
在 ADK 中，Agent 的每一次操作都会产生**事件（Event）**。事件是 Agent 执行过程中的基本单元，记录了从接收用户输入到返回最终响应的完整过程。
|
||
|
||
```
|
||
用户输入 → [事件1] → [事件2] → ... → [事件N] → 最终响应
|
||
│ │ │
|
||
│ │ └── 最终响应事件
|
||
│ └── 工具调用事件
|
||
└── 模型调用事件
|
||
```
|
||
|
||
### 7.1.2 事件类型
|
||
|
||
| 事件类型 | 说明 | 触发时机 |
|
||
|----------|------|----------|
|
||
| **Model Event** | LLM 模型调用事件 | Agent 调用 LLM 时 |
|
||
| **Tool Event** | 工具调用事件 | Agent 调用工具时 |
|
||
| **Transfer Event** | Agent 转移事件 | LLM 决定委派给其他 Agent 时 |
|
||
| **Final Response** | 最终响应事件 | Agent 生成最终回复时 |
|
||
| **Error Event** | 错误事件 | 执行过程中发生错误时 |
|
||
|
||
---
|
||
|
||
## 7.2 回调函数详解
|
||
|
||
ADK 提供了四种回调函数，允许你在 Agent 执行的关键节点插入自定义逻辑：
|
||
|
||
```
|
||
用户输入
|
||
│
|
||
▼
|
||
┌─────────────────────────────────────┐
|
||
│ before_model_callback │ ← 模型调用前
|
||
│ （可以修改输入、记录日志等） │
|
||
├─────────────────────────────────────┤
|
||
│ LLM 模型调用 │
|
||
├─────────────────────────────────────┤
|
||
│ after_model_callback │ ← 模型调用后
|
||
│ （可以修改输出、记录响应等） │
|
||
├─────────────────────────────────────┤
|
||
│ before_tool_callback │ ← 工具调用前
|
||
│ （可以验证参数、阻止调用等） │
|
||
├─────────────────────────────────────┤
|
||
│ 工具执行 │
|
||
├─────────────────────────────────────┤
|
||
│ after_tool_callback │ ← 工具调用后
|
||
│ （可以处理结果、记录日志等） │
|
||
└─────────────────────────────────────┘
|
||
```
|
||
|
||
### 7.2.1 before_model_callback（模型调用前）
|
||
|
||
```python
|
||
"""
|
||
before_model_callback 示例
|
||
在 LLM 模型调用之前执行
|
||
可以修改输入、记录日志、添加上下文等
|
||
"""
|
||
|
||
from google.adk.agents import Agent # 导入 Agent 类
|
||
from google.adk.agents.invocation_context import InvocationContext # 导入调用上下文
|
||
|
||
|
||
# ========================================
|
||
# 定义 before_model_callback
|
||
# ========================================
|
||
|
||
async def my_before_model_callback(
|
||
callback_context, # 回调上下文
|
||
invocation_context: InvocationContext, # 调用上下文
|
||
):
|
||
"""
|
||
模型调用前的回调函数
|
||
|
||
Args:
|
||
callback_context: 回调上下文，包含工具调用信息
|
||
invocation_context: 调用上下文，包含会话和状态信息
|
||
"""
|
||
# 记录日志：模型即将被调用
|
||
print(f"[日志] Agent '{invocation_context.agent.name}' 即将调用模型") # 打印日志
|
||
|
||
# 访问会话状态
|
||
call_count = invocation_context.session.state.get( # 读取调用计数
|
||
"model_call_count", # 状态键名
|
||
0, # 默认值
|
||
)
|
||
|
||
# 更新调用计数
|
||
invocation_context.session.state["model_call_count"] = call_count + 1 # 递增计数
|
||
|
||
print(f"[日志] 这是第 {call_count + 1} 次模型调用") # 打印调用次数
|
||
|
||
# 可以在这里添加额外的上下文信息
|
||
# 例如：将用户偏好注入到状态中
|
||
invocation_context.session.state["temp:call_timestamp"] = "2025-04-05T10:00:00Z" # 记录时间戳
|
||
|
||
|
||
# ========================================
|
||
# 创建使用回调的 Agent
|
||
# ========================================
|
||
|
||
agent = Agent(
|
||
name="callback_agent", # Agent 名称
|
||
model="gemini-2.0-flash", # 模型
|
||
instruction="你是一个助手。", # 指令
|
||
before_model_callback=my_before_model_callback, # 🔑 注册模型前回调
|
||
)
|
||
```
|
||
|
||
### 7.2.2 after_model_callback（模型调用后）
|
||
|
||
```python
|
||
"""
|
||
after_model_callback 示例
|
||
在 LLM 模型调用之后执行
|
||
可以修改输出、记录响应、触发后续操作等
|
||
"""
|
||
|
||
from google.adk.agents import Agent # 导入 Agent 类
|
||
from google.adk.agents.invocation_context import InvocationContext # 导入调用上下文
|
||
|
||
|
||
async def my_after_model_callback(
|
||
callback_context, # 回调上下文
|
||
invocation_context: InvocationContext, # 调用上下文
|
||
):
|
||
"""
|
||
模型调用后的回调函数
|
||
|
||
Args:
|
||
callback_context: 回调上下文，包含模型响应信息
|
||
invocation_context: 调用上下文
|
||
"""
|
||
# 获取模型响应
|
||
response = callback_context.response # 模型响应对象
|
||
|
||
if response: # 如果有响应
|
||
# 记录响应信息
|
||
print(f"[日志] 模型返回了响应") # 打印日志
|
||
|
||
# 检查响应中是否有工具调用
|
||
if response.function_calls: # 如果有函数调用
|
||
for fc in response.function_calls: # 遍历函数调用
|
||
print(f"[日志] 模型决定调用工具: {fc.name}") # 打印工具名
|
||
else: # 如果没有函数调用
|
||
print(f"[日志] 模型直接返回了文本响应") # 打印日志
|
||
|
||
# 可以在这里进行响应后处理
|
||
# 例如：记录 token 使用量、响应延迟等
|
||
invocation_context.session.state["temp:last_model_call"] = "completed" # 标记完成
|
||
|
||
|
||
agent = Agent(
|
||
name="callback_agent", # Agent 名称
|
||
model="gemini-2.0-flash", # 模型
|
||
instruction="你是一个助手。", # 指令
|
||
after_model_callback=my_after_model_callback, # 🔑 注册模型后回调
|
||
)
|
||
```
|
||
|
||
### 7.2.3 before_tool_callback（工具调用前）
|
||
|
||
```python
|
||
"""
|
||
before_tool_callback 示例
|
||
在工具调用之前执行
|
||
可以验证参数、阻止调用、修改参数等
|
||
"""
|
||
|
||
from google.adk.agents import Agent # 导入 Agent 类
|
||
from google.adk.agents.invocation_context import InvocationContext # 导入调用上下文
|
||
|
||
|
||
async def my_before_tool_callback(
|
||
callback_context, # 回调上下文
|
||
invocation_context: InvocationContext, # 调用上下文
|
||
):
|
||
"""
|
||
工具调用前的回调函数
|
||
|
||
Args:
|
||
callback_context: 回调上下文，包含工具调用信息
|
||
invocation_context: 调用上下文
|
||
"""
|
||
# 获取工具调用信息
|
||
tool_call = callback_context.function_call # 函数调用对象
|
||
tool_name = tool_call.name # 工具名称
|
||
tool_args = tool_call.args # 工具参数
|
||
|
||
print(f"[日志] 即将调用工具: {tool_name}") # 打印工具名
|
||
print(f"[日志] 工具参数: {tool_args}") # 打印参数
|
||
|
||
# ========================================
|
||
# 示例：敏感操作确认（Human-in-the-Loop）
|
||
# ========================================
|
||
sensitive_tools = ["delete_data", "send_email", "make_payment"] # 敏感操作列表
|
||
|
||
if tool_name in sensitive_tools: # 如果是敏感操作
|
||
print(f"[警告] 敏感操作: {tool_name}，需要人工确认！") # 打印警告
|
||
# 在实际应用中，这里可以：
|
||
# 1. 发送确认请求给用户
|
||
# 2. 阻止工具执行
|
||
# 3. 记录审计日志
|
||
|
||
# ========================================
|
||
# 示例：参数验证
|
||
# ========================================
|
||
if tool_name == "search": # 如果是搜索工具
|
||
query = tool_args.get("query", "") # 获取查询参数
|
||
if len(query) < 2: # 如果查询太短
|
||
print("[警告] 搜索查询过短") # 打印警告
|
||
|
||
|
||
agent = Agent(
|
||
name="callback_agent", # Agent 名称
|
||
model="gemini-2.0-flash", # 模型
|
||
instruction="你是一个助手。", # 指令
|
||
before_tool_callback=my_before_tool_callback, # 🔑 注册工具前回调
|
||
)
|
||
```
|
||
|
||
### 7.2.4 after_tool_callback（工具调用后）
|
||
|
||
```python
|
||
"""
|
||
after_tool_callback 示例
|
||
在工具调用之后执行
|
||
可以处理结果、记录日志、触发后续操作等
|
||
"""
|
||
|
||
from google.adk.agents import Agent # 导入 Agent 类
|
||
from google.adk.agents.invocation_context import InvocationContext # 导入调用上下文
|
||
|
||
|
||
async def my_after_tool_callback(
|
||
callback_context, # 回调上下文
|
||
invocation_context: InvocationContext, # 调用上下文
|
||
):
|
||
"""
|
||
工具调用后的回调函数
|
||
|
||
Args:
|
||
callback_context: 回调上下文，包含工具调用结果
|
||
invocation_context: 调用上下文
|
||
"""
|
||
# 获取工具调用信息
|
||
tool_call = callback_context.function_call # 函数调用对象
|
||
tool_result = callback_context.tool_result # 工具执行结果
|
||
|
||
print(f"[日志] 工具 {tool_call.name} 执行完成") # 打印完成日志
|
||
print(f"[日志] 执行结果: {tool_result}") # 打印结果
|
||
|
||
# ========================================
|
||
# 示例：结果后处理
|
||
# ========================================
|
||
if tool_result: # 如果有结果
|
||
# 将工具结果保存到状态中
|
||
invocation_context.session.state[ # 保存到状态
|
||
f"temp:last_{tool_call.name}_result" # 动态键名
|
||
] = str(tool_result) # 保存结果字符串
|
||
|
||
# ========================================
|
||
# 示例：错误处理
|
||
# ========================================
|
||
if tool_result and tool_result.get("status") == "error": # 如果工具返回错误
|
||
print(f"[错误] 工具执行失败: {tool_result.get('error_message')}") # 打印错误
|
||
# 可以在这里进行错误恢复或通知
|
||
|
||
|
||
agent = Agent(
|
||
name="callback_agent", # Agent 名称
|
||
model="gemini-2.0-flash", # 模型
|
||
instruction="你是一个助手。", # 指令
|
||
after_tool_callback=my_after_tool_callback, # 🔑 注册工具后回调
|
||
)
|
||
```
|
||
|
||
### 7.2.5 组合使用所有回调
|
||
|
||
```python
|
||
"""
|
||
组合使用所有回调函数
|
||
创建一个完整的带日志和监控的 Agent
|
||
"""
|
||
|
||
from google.adk.agents import Agent # 导入 Agent 类
|
||
from google.adk.agents.invocation_context import InvocationContext # 导入调用上下文
|
||
import time # 导入时间模块
|
||
|
||
|
||
# ========================================
|
||
# 日志回调：记录所有操作
|
||
# ========================================
|
||
|
||
async def log_before_model(cb_ctx, inv_ctx):
|
||
"""模型调用前：记录开始时间"""
|
||
inv_ctx.session.state["temp:model_start_time"] = time.time() # 记录开始时间
|
||
print(f"🔍 [模型] 开始调用 LLM...") # 打印日志
|
||
|
||
|
||
async def log_after_model(cb_ctx, inv_ctx):
|
||
"""模型调用后：计算耗时"""
|
||
start = inv_ctx.session.state.get("temp:model_start_time", time.time()) # 获取开始时间
|
||
elapsed = time.time() - start # 计算耗时
|
||
print(f"✅ [模型] LLM 调用完成，耗时: {elapsed:.2f}s") # 打印耗时
|
||
|
||
|
||
async def log_before_tool(cb_ctx, inv_ctx):
|
||
"""工具调用前：记录工具信息"""
|
||
fc = cb_ctx.function_call # 获取函数调用
|
||
print(f"🔧 [工具] 调用工具: {fc.name}({fc.args})") # 打印工具信息
|
||
|
||
|
||
async def log_after_tool(cb_ctx, inv_ctx):
|
||
"""工具调用后：记录结果"""
|
||
fc = cb_ctx.function_call # 获取函数调用
|
||
result = cb_ctx.tool_result # 获取工具结果
|
||
status = "成功" if result and result.get("status") == "success" else "失败" # 判断状态
|
||
print(f"📊 [工具] {fc.name} 执行{status}") # 打印结果
|
||
|
||
|
||
# ========================================
|
||
# 创建带完整回调的 Agent
|
||
# ========================================
|
||
|
||
def get_weather(city: str) -> dict:
|
||
"""获取天气"""
|
||
return {"status": "success", "weather": "晴天"} # 模拟天气数据
|
||
|
||
|
||
monitored_agent = Agent(
|
||
name="monitored_agent", # Agent 名称
|
||
model="gemini-2.0-flash", # 模型
|
||
instruction="你是一个天气助手。", # 指令
|
||
tools=[get_weather], # 工具
|
||
before_model_callback=log_before_model, # 模型前回调
|
||
after_model_callback=log_after_model, # 模型后回调
|
||
before_tool_callback=log_before_tool, # 工具前回调
|
||
after_tool_callback=log_after_tool, # 工具后回调
|
||
)
|
||
```
|
||
|
||
---
|
||
|
||
## 7.3 Human-in-the-Loop（人机协作）
|
||
|
||
### 7.3.1 实现人工确认
|
||
|
||
```python
|
||
"""
|
||
Human-in-the-Loop 模式
|
||
在关键操作前暂停，等待人工确认
|
||
"""
|
||
|
||
from google.adk.agents import Agent # 导入 Agent 类
|
||
from google.adk.agents.invocation_context import InvocationContext # 导入调用上下文
|
||
|
||
|
||
# ========================================
|
||
# 定义需要人工确认的工具
|
||
# ========================================
|
||
|
||
def delete_file(filename: str) -> dict:
|
||
"""
|
||
删除文件（危险操作）
|
||
|
||
Args:
|
||
filename (str): 要删除的文件名
|
||
|
||
Returns:
|
||
dict: 操作结果
|
||
"""
|
||
# 在实际应用中，这里会执行真正的删除操作
|
||
print(f"🗑️ 已删除文件: {filename}") # 打印删除信息
|
||
|
||
return { # 返回结果
|
||
"status": "success", # 状态
|
||
"message": f"文件 '{filename}' 已删除", # 消息
|
||
}
|
||
|
||
|
||
def send_email(to: str, subject: str, body: str) -> dict:
|
||
"""
|
||
发送邮件（敏感操作）
|
||
|
||
Args:
|
||
to (str): 收件人邮箱
|
||
subject (str): 邮件主题
|
||
body (str): 邮件正文
|
||
|
||
Returns:
|
||
dict: 操作结果
|
||
"""
|
||
print(f"📧 已发送邮件到: {to}") # 打印发送信息
|
||
|
||
return { # 返回结果
|
||
"status": "success",
|
||
"message": f"邮件已发送到 {to}",
|
||
}
|
||
|
||
|
||
# ========================================
|
||
# 工具调用前回调：人工确认
|
||
# ========================================
|
||
|
||
async def human_confirmation_callback(
|
||
callback_context, # 回调上下文
|
||
invocation_context: InvocationContext, # 调用上下文
|
||
):
|
||
"""
|
||
人工确认回调
|
||
在执行敏感操作前暂停，等待人工确认
|
||
"""
|
||
tool_name = callback_context.function_call.name # 获取工具名
|
||
tool_args = callback_context.function_call.args # 获取工具参数
|
||
|
||
# 定义需要人工确认的操作
|
||
dangerous_operations = { # 危险操作映射
|
||
"delete_file": "删除文件", # 删除文件
|
||
"send_email": "发送邮件", # 发送邮件
|
||
}
|
||
|
||
if tool_name in dangerous_operations: # 如果是危险操作
|
||
operation = dangerous_operations[tool_name] # 获取操作描述
|
||
print(f"\n⚠️ 需要人工确认！") # 打印警告
|
||
print(f"操作: {operation}") # 打印操作
|
||
print(f"参数: {tool_args}") # 打印参数
|
||
print(f"请确认是否继续...") # 提示确认
|
||
|
||
# 在实际应用中，这里应该：
|
||
# 1. 通过 API 发送确认请求给前端
|
||
# 2. 等待用户确认
|
||
# 3. 根据确认结果决定是否继续
|
||
|
||
# 模拟人工确认（实际应用中应替换为真正的确认机制）
|
||
# 如果用户拒绝，可以抛出异常来阻止工具执行
|
||
# raise ValueError("用户拒绝了此操作")
|
||
|
||
|
||
# ========================================
|
||
# 创建带人工确认的 Agent
|
||
# ========================================
|
||
|
||
safe_agent = Agent(
|
||
name="safe_agent", # Agent 名称
|
||
model="gemini-2.0-flash", # 模型
|
||
instruction=( # 指令
|
||
"你是一个文件管理助手。\n"
|
||
"可以查看和删除文件，也可以发送邮件。\n"
|
||
"执行敏感操作前需要人工确认。"
|
||
),
|
||
tools=[ # 工具列表
|
||
delete_file, # 删除文件
|
||
send_email, # 发送邮件
|
||
],
|
||
before_tool_callback=human_confirmation_callback, # 🔑 人工确认回调
|
||
)
|
||
```
|
||
|
||
---
|
||
|
||
## 7.4 事件流处理
|
||
|
||
### 7.4.1 处理不同类型的事件
|
||
|
||
```python
|
||
"""
|
||
事件流处理
|
||
从 Runner 的事件流中提取和处理不同类型的事件
|
||
"""
|
||
|
||
from google.adk.agents import Agent # 导入 Agent 类
|
||
from google.adk.runners import Runner # 导入运行器
|
||
from google.adk.sessions import InMemorySessionService # 导入会话服务
|
||
from google.genai import types # 导入类型定义
|
||
import asyncio # 导入异步库
|
||
|
||
|
||
agent = Agent(
|
||
name="event_agent", # Agent 名称
|
||
model="gemini-2.0-flash", # 模型
|
||
instruction="你是一个助手。", # 指令
|
||
)
|
||
|
||
|
||
async def process_events():
|
||
"""处理事件流的完整示例"""
|
||
|
||
# 初始化
|
||
session_service = InMemorySessionService() # 会话服务
|
||
await session_service.create_session( # 创建会话
|
||
app_name="event_app", # 应用名称
|
||
user_id="user_001", # 用户 ID
|
||
session_id="event_001", # 会话 ID
|
||
)
|
||
|
||
runner = Runner(
|
||
agent=agent, # Agent
|
||
app_name="event_app", # 应用名称
|
||
session_service=session_service, # 会话服务
|
||
)
|
||
|
||
# 构造消息
|
||
content = types.Content(
|
||
role='user', # 角色
|
||
parts=[types.Part(text='你好')], # 内容
|
||
)
|
||
|
||
# 运行并处理事件
|
||
events = runner.run_async(
|
||
user_id="user_001", # 用户 ID
|
||
session_id="event_001", # 会话 ID
|
||
new_message=content, # 消息
|
||
)
|
||
|
||
async for event in events: # 遍历事件流
|
||
# 检查事件类型并分别处理
|
||
event_type = type(event).__name__ # 获取事件类型名
|
||
print(f"[事件] 类型: {event_type}") # 打印事件类型
|
||
|
||
# 处理最终响应
|
||
if event.is_final_response(): # 如果是最终响应
|
||
text = event.content.parts[0].text # 提取文本
|
||
print(f"[最终响应] {text}") # 打印响应
|
||
|
||
# 处理工具调用
|
||
if hasattr(event, 'function_call') and event.function_call: # 如果有工具调用
|
||
fc = event.function_call # 获取函数调用
|
||
print(f"[工具调用] {fc.name}({fc.args})") # 打印调用信息
|
||
|
||
# 处理工具响应
|
||
if hasattr(event, 'function_response') and event.function_response: # 如果有工具响应
|
||
fr = event.function_response # 获取函数响应
|
||
print(f"[工具响应] {fr.name}: {fr.response}") # 打印响应
|
||
|
||
|
||
asyncio.run(process_events()) # 执行事件处理
|
||
```
|
||
|
||
---
|
||
|
||
## 📌 本章小结
|
||
|
||
- ADK 通过事件系统记录 Agent 的完整执行过程
|
||
- 四种回调函数：before_model、after_model、before_tool、after_tool
|
||
- 回调函数可用于日志记录、参数验证、人工确认等
|
||
- Human-in-the-Loop 模式通过 before_tool_callback 实现
|
||
- Runner 返回的事件流可以逐个处理不同类型的事件
|
||
|
||
**下一章**：[第08章 - 智能体评估](./08-evaluation.md) → 学习如何系统地评估 Agent 的性能。
|