GenAI 处理器库 📚

构建模块化、异步且可组合的生成式 AI 流水线。

GenAI 处理器是一个轻量级的 Python 库，能够实现高效、并行的内容处理。它通过三大核心支柱解决了 LLM API 的碎片化问题：

1. 统一内容模型：为跨模型、代理和工具的输入与输出提供单一、一致的表示。
2. 处理器：简单、可组合的 Python 类，使用原生 asyncio 转换内容流。
3. 流式处理：默认内置异步流式能力，无需额外的复杂配置。

该生态系统的核心是 Processor，它封装了一个工作单元。通过“双接口”模式，它处理了异步、多模态数据流的复杂性，同时向开发者暴露了一个简单的 API：

from typing import AsyncIterable
from genai_processors import content_api
from genai_processors import processor

class EchoProcessor(processor.Processor):
  # 生产者接口（面向处理器作者）：
  # 接受强大的 ProcessorStream 作为输入，并产出部分类型。
  async def call(
      self, content: content_api.ProcessorStream
  ) -> AsyncIterable[content_api.ProcessorPartTypes]:
      # 在内容流式传入时进行处理！
      async for part in content:
          yield part

应用 Processor 同样非常直观。消费者接口接受广泛且容错性强的输入类型，并返回一个功能强大的流，可以一次性等待所有结果，也可以逐块流式处理：

# 消费者接口（面向调用者）：
# 轻松提供输入。字符串会自动转换为 Part。
input_content = ["Hello ", content_api.ProcessorPart("World")]

# 1. 将所有输出轻松汇总到一个对象中：
result: content_api.ProcessorContent = await simple_text_processor(input_content).gather()

# 2. 或者对于纯文本代理，直接获取文本：
print(await simple_text_processor(input_content).text())

# 3. 对于流式场景，可以逐个处理到达的部分：
async for part in simple_text_processor(input_content):
  print(part.text, end="")

Processor 的概念为 Gemini 模型调用以及围绕其构建的日益复杂的行为提供了通用抽象，既能支持轮次交互，也能支持实时流式传输。

✨ 核心特性

• 模块化：将复杂任务分解为可重用的 Processor 和 PartProcessor 单元，这些单元可以轻松地串联（+）或并行化（//），以创建复杂的数据流和智能体行为。
• 与 GenAI API 集成：包含开箱即用的处理器，如用于轮次 API 调用的 GenaiModel 和用于实时流式交互的 LiveProcessor。
• 可扩展性：允许您通过继承基类或使用简单的函数装饰器来创建自定义处理器。
• 丰富的内容处理：
  • ProcessorPart：基于 genai.types.Part 的包装器，附加了 MIME 类型、角色和自定义属性等元数据。
  • 支持多种内容类型（文本、图像、音频、自定义 JSON）。
• 异步与并发：基于 Python 熟悉的 asyncio 框架，协调并发任务（包括网络 I/O 和与计算密集型子线程的通信）。
• 流管理：提供用于拆分、拼接和合并 ProcessorPart 异步流的实用工具。

📦 安装

GenAI 处理器库需要 Python 3.10 或更高版本。

安装命令如下：

pip install genai-processors

代码生成

生成式模型通常不了解最新的 API 和 SDK 更新，可能会建议过时或遗留的代码。

我们建议在生成使用 GenAI 处理器的代码时，使用我们的 代码生成说明，以引导您的模型使用更现代的 SDK 功能。请将说明复制并粘贴到您的开发环境中，以便为模型提供必要的上下文。

🚀 入门指南

我们建议从 文档微站点 开始，其中涵盖了核心概念、开发指南和架构设计。

您还可以查看以下 Colab 笔记本，以熟悉 GenAI 处理器（建议按顺序阅读）：

• 内容 API Colab - 解释了 ProcessorPart、ProcessorContent 的基础知识以及如何创建它们。
• 处理器入门 Colab - 介绍 GenAI 处理器的核心概念。
• 创建您自己的处理器 Colab - 演示了创建 Processor 或 PartProcessor 的典型步骤。
• 使用 Live API - 展示了使用 LiveProcessor 类基于 Gemini Live API 构建的几个实时处理器示例。

📖 示例

请浏览 examples/ 目录，获取实用演示：

• 实时 Live 示例 - 一个音频输入、音频输出的 Live 代理，以谷歌搜索作为工具。它是客户端侧的 Live 处理器实现（基于文本的 Gemini API 模型），展示了 GenAI 处理器的流式传输和编排能力。
• 研究代理示例 - 一个由处理器组成的研究代理，包含 3 个子处理器、链式处理、创建 ProcessorPart 等操作。
• 实时解说示例 - 描述了一个基于 Gemini Live API 构建的实时解说代理，由两个代理组成：一个用于事件检测，另一个用于管理对话。

🧩 内置处理器

core/ 目录包含一组基础处理器，您可以在自己的应用中使用。它提供了大多数实时应用所需的通用构建模块，并将随着时间的推移不断演进，加入更多核心组件。

社区贡献扩展的内置处理器集合位于 contrib/ 下——请参阅下方关于如何向 GenAI 处理器库添加代码的部分。

🤝 贡献

欢迎贡献！请参阅 CONTRIBUTING.md，了解如何为本项目做出贡献的指南。

📜 许可证

本项目采用 Apache License, Version 2.0 许可证。详情请参阅 LICENSE 文件。

Gemini 服务条款

如果您通过 Genai 处理器框架使用 Gemini，请务必查阅 服务条款。

GenAI Processors 快速上手指南

GenAI Processors 是一个轻量级 Python 库，旨在帮助开发者构建模块化、异步且可组合的生成式 AI 管道。它通过统一的内容模型和原生 asyncio 支持，简化了多模态数据流的处理与 LLM API 的集成。

环境准备

在开始之前，请确保您的开发环境满足以下要求：

• 操作系统：Linux, macOS 或 Windows
• Python 版本：3.10 或更高版本
• 前置依赖：建议预先安装好 pip 包管理工具
• API 密钥：如需使用 Gemini 模型功能，请准备好 Google AI API Key

安装步骤

使用 pip 直接安装官方发布版本：

pip install genai-processors

提示：如果您在国内网络环境下遇到下载速度慢的问题，可以使用国内镜像源加速安装：

pip install genai-processors -i https://pypi.tuna.tsinghua.edu.cn/simple

基本使用

GenAI Processors 的核心概念是 Processor（处理器），它支持流式输入输出。以下是最简单的使用示例，展示如何定义一个回声处理器并调用它。

1. 定义一个简单的 Processor

创建一个继承自 processor.Processor 的类，实现 call 方法来处理流式数据：

from typing import AsyncIterable
from genai_processors import content_api
from genai_processors import processor

class EchoProcessor(processor.Processor):
  # 生产者接口：接收 ProcessorStream 输入，并 yield 处理后的部分
  async def call(
      self, content: content_api.ProcessorStream
  ) -> AsyncIterable[content_api.ProcessorPartTypes]:
      # 模拟流式处理：接收到什么就输出什么
      async for part in content:
          yield part

2. 调用 Processor

消费者接口非常灵活，支持字符串、列表等多种输入格式，并提供多种获取结果的方式（一次性获取、仅获取文本或流式迭代）：

import asyncio
from genai_processors import content_api

# 初始化处理器实例
simple_text_processor = EchoProcessor()

# 准备输入内容：支持混合类型，字符串会自动转换为 Part
input_content = ["Hello ", content_api.ProcessorPart("World")]

async def main():
    # 方式 1: 一次性收集所有输出
    result: content_api.ProcessorContent = await simple_text_processor(input_content).gather()
    print(f"Gathered result: {result}")

    # 方式 2: 仅获取纯文本结果
    text_result = await simple_text_processor(input_content).text()
    print(f"Text result: {text_result}")

    # 方式 3: 流式迭代处理（适合实时场景）
    print("Streaming output: ", end="")
    async for part in simple_text_processor(input_content):
        print(part.text, end="")
    print() # 换行

if __name__ == "__main__":
    asyncio.run(main())

核心特性速览

• 组合与并行：使用 + 串联处理器，使用 // 并行执行处理器。
• 多模态支持：原生支持文本、图像、音频及自定义 JSON 数据的 ProcessorPart 封装。
• Gemini 集成：内置 GenaiModel 用于常规对话，LiveProcessor 用于实时音视频流交互。

更多高级用法（如创建自定义处理器、连接 Gemini Live API 等），建议参考官方 文档站点 或 GitHub 仓库中的 Colab 示例。