GenAI Processors：轻量级Python库支持高效并行处理多模态内容

GenAI Processors 是一个由 Google DeepMind 开发的开源 Python 库，专注于高效并行处理多模态内容。它基于 Python 的 asyncio 框架，提供模块化、可复用的处理器接口，简化复杂 AI 应用的开发。用户可以通过这个库处理文本、音频、视频等数据流，并与 Gemini API 无缝集成。它支持实时流处理和转基于回合的交互，适合构建需要快速响应的 AI 应用。代码托管在 GitHub 上，社区可贡献处理器模块，扩展功能。项目采用 Apache 2.0 许可证，适合开发者快速构建生产环境可用的 AI 应用。

功能列表

• 异步并行处理：基于 Python asyncio，支持高效处理 I/O 和计算密集型任务。
• 模块化处理器设计：提供 Processor 和 PartProcessor 单元，可组合或并行处理复杂数据流。
• Gemini API 集成：内置 GenaiModel 和 LiveProcessor，支持回合式和实时流式交互。
• 多模态流处理：支持文本、音频、视频等数据流的拆分、合并和处理。
• 实时交互支持：通过 LiveProcessor 处理实时音视频流，适合构建实时 AI 代理。
• 社区贡献扩展：支持用户在 contrib/ 目录下添加自定义处理器，增强功能。
• 工具集成：内置 Google 搜索等工具，增强 AI 代理的上下文处理能力。

使用帮助

安装流程

GenAI Processors 需要 Python 3.10 或更高版本。以下是详细安装步骤：

1. 设置环境：
   • 确保系统已安装 Python 3.10+。
   • 安装 Git，以便克隆代码仓库。

   sudo apt update && sudo apt install python3.10 git
   

2. 克隆仓库：
   • 从 GitHub 克隆 GenAI Processors 仓库。

   git clone https://github.com/google-gemini/genai-processors
   cd genai-processors
   

3. 安装依赖：
   • 使用 pip 安装所需依赖，包括 pyaudio、google-genai 和 termcolor。

   pip install --upgrade pyaudio genai-processors google-genai termcolor
   

4. 配置 API 密钥：
   • 获取 Google AI Studio 的 API 密钥。
   • 设置环境变量 GOOGLE_API_KEY 和 GOOGLE_PROJECT_ID。

   export GOOGLE_API_KEY="你的API密钥"
   export GOOGLE_PROJECT_ID="你的项目ID"
   

使用方式

GenAI Processors 的核心是 Processor 模块，用于处理输入输出流。以下是主要功能的详细操作流程：

1. 创建简单文本处理器

• 功能：处理文本输入并输出结果。
• 操作流程：
  • 导入模块并创建输入流。
  • 使用 stream_content 将文本转为 ProcessorPart 流。
  • 应用处理器并迭代输出。

  from genai_processors import content_api, streams
  input_parts = ["Hello", content_api.ProcessorPart("World")]
  input_stream = streams.stream_content(input_parts)
  async for part in simple_text_processor(input_stream):
  print(part.text)
  

• 效果：将输入文本逐部分处理并打印，适合简单文本任务。

2. 构建实时音视频代理

• 功能：通过 LiveProcessor 处理实时音视频流。
• 操作流程：
  • 初始化音频输入设备（如 PyAudio）。
  • 配置视频输入（如摄像头或屏幕流）。
  • 使用 LiveProcessor 调用 Gemini Live API。
  • 组合输入、处理和输出模块。

  from genai_processors.core import audio_io, live_model, video
  import pyaudio
  pya = pyaudio.PyAudio()
  input_processor = video.VideoIn() + audio_io.PyAudioIn(pya, use_pcm_mimetype=True)
  live_processor = live_model.LiveProcessor(api_key="你的API密钥", model_name="gemini-2.5-flash-preview-native-audio-dialog")
  play_output = audio_io.PyAudioOut(pya)
  live_agent = input_processor + live_processor + play_output
  async for part in live_agent(text.terminal_input()):
  print(part)
  

• 效果：实现麦克风和摄像头输入，通过 Gemini API 处理后输出音频，适合实时对话代理。

3. 研究主题生成

• 功能：根据用户输入生成研究主题。
• 操作流程：
  • 使用 topic_generator.py 示例，配置 GenaiModel。
  • 设置模型参数，如主题数量和输出格式。
  • 输入研究查询，获取 JSON 格式的主题列表。

  from genai_processors.examples.research.processors import topic_generator
  processor = topic_generator.TopicGenerator(api_key="你的API密钥")
  async for part in processor(["研究AI在医疗领域的应用"]):
  print(part.text)
  

• 效果：生成指定数量的研究主题及其与输入的关系，适合学术研究场景。

4. 自定义处理器

• 功能：创建自定义处理器处理特定任务。
• 操作流程：
  • 参考 create_your_own_processor.ipynb 笔记本。
  • 定义 Processor 类，继承 processor.Processor。
  • 实现 call 方法处理输入流。
  • 将自定义处理器加入管道。
• 效果：用户可根据需求扩展功能，如处理特定文件格式或集成其他 API。

运行示例

• 实时 CLI 示例：
  • 运行 realtime_simple_cli.py 创建音频对话代理。

  python3 examples/realtime_simple_cli.py
  

  • 输入语音，系统将语音转为文本，处理后输出语音回复。
• 旅行计划 CLI：
  • 运行 trip_request_cli.py 生成旅行计划。

  python3 examples/trip_request_cli.py
  

  • 输入目的地和日期，获取详细计划。

注意事项

• 确保 API 密钥有效，避免请求失败。
• 调试时可设置 --debug=True 查看日志。
• 实时处理需稳定网络和硬件支持。

应用场景

1. 实时对话代理
   • 描述：开发语音或视频驱动的 AI 助手，处理用户实时输入，适合客服或虚拟助手。
2. 学术研究支持
   • 描述：生成研究主题或分析文献，适合学生和研究人员快速整理思路。
3. 多模态内容处理
   • 描述：处理音视频流，生成字幕或实时解说，适合直播或视频分析。
4. 自动化工作流
   • 描述：构建自动化处理管道，处理批量数据，适合企业数据处理。

QA

1. 需要哪些前置条件？
   • 需要 Python 3.10+，安装 pyaudio 和 google-genai 库，设置 Google API 密钥。
2. 如何调试处理流程？
   • 运行脚本时添加 --debug=True，查看日志输出，检查输入输出流。
3. 支持哪些数据类型？
   • 支持文本、音频、视频和自定义数据流，可通过 ProcessorPart 处理。
4. 如何贡献代码？
   • 参考 CONTRIBUTING.md，在 contrib/ 目录提交自定义处理器。