Google ADK Python 项目中生成器对象序列化问题的分析与解决

在 Google ADK Python 项目中，开发者在使用 LongRunningFunctionTool 工具时经常会遇到一个典型错误："TypeError: cannot pickle 'generator' object"。这个问题源于 Python 生成器对象的序列化限制，本文将深入分析问题本质并提供完整的解决方案。

问题背景

当开发者尝试使用 LongRunningFunctionTool 包装一个生成器函数时，系统会抛出序列化错误。这是因为 ADK 框架内部需要对函数调用结果进行深拷贝和序列化操作，而 Python 的生成器对象本身不支持 pickle 序列化。

典型的错误场景出现在以下代码中：

def process_large_file(file_path: str) -> dict:
    yield {"status": "pending"}
    # ...更多yield语句...
    return {"status": "completed"}

long_running_tool = LongRunningFunctionTool(func=process_large_file)

技术原理分析

问题的核心在于 Python 的序列化机制和生成器的特性：

1. 生成器本质：Python 生成器是一种特殊的迭代器，它保存的是执行状态而非数据本身，这种状态无法被序列化。

2. ADK 框架需求：框架需要在会话中保存事件历史，这要求所有传递的对象必须支持深拷贝和序列化。

3. 会话管理：无论是 InMemorySessionService 还是 DatabaseSessionService，最终都需要将事件内容转换为可存储格式。

解决方案

根据官方建议和实际验证，正确的实现方式应该是使用异步生成器模式：

1. 异步生成器实现

async def async_process_file(file_path: str) -> AsyncIterator[dict]:
    yield {"status": "pending"}
    # 模拟处理过程
    for i in range(5):
        await asyncio.sleep(1)
        yield {"progress": f"{20*(i+1)}%"}
    return {"status": "completed"}

2. 工具包装方式

long_running_tool = LongRunningFunctionTool(func=async_process_file)

3. 运行方式调整

必须使用 run_live 方法而非 run 方法执行：

async for event in runner.run_live(...):
    handle_event(event)

实现要点

1. 异步化改造：将普通生成器函数改为异步生成器(async generator)，使用 async/await 语法。

2. 实时流处理：run_live 方法专为流式响应设计，能够正确处理异步生成器的逐步输出。

3. 状态管理：每个 yield 返回的字典应该包含足够的进度信息，便于前端展示。

最佳实践建议

1. 进度报告格式：建议采用统一的状态字段，如：

   {
       "status": "pending|completed|failed",
       "progress": "0-100%",
       "message": "当前状态描述"
   }
   

2. 错误处理：在生成器内部捕获异常并通过特定状态返回：

   try:
       # 处理逻辑
   except Exception as e:
       yield {"status": "failed", "error": str(e)}
       return
   

3. 资源清理：对于需要资源清理的操作，使用 try-finally 块确保资源释放。

总结

在 Google ADK Python 项目中正确处理长时运行任务需要理解框架的序列化需求和 Python 生成器的特性。通过改用异步生成器模式和正确的执行方法，开发者可以构建出稳定可靠的长时任务处理流程。这种模式特别适合文件处理、大数据计算等需要分阶段报告进度的场景。