访问网站
扫码查看
复制链接

一个用于构建模块化AI流程的Python库

标签：AI , asyncio , Gemini , genai , generative-ai , Python , 多模态 , 实时 , 智能体 , 语言模型

GenAI Processors

AI专业工具 / AI基础服务9 次浏览

综合介绍

GenAI Processors 是一个由 Google DeepMind 开发的轻量级 Python 库。 它旨在简化生成式 AI 应用的开发过程，特别是那些需要处理多模态输入和实时响应的复杂应用。 这个库的核心思想是提供一个统一的“处理器”（Processor）抽象概念。 无论是处理数据、调用模型还是格式化输出，所有操作都可以被封装成一个独立的处理器模块。这些处理器可以像乐高积木一样，通过简单的操作符进行串联（+）或并联（//），从而构建出强大而灵活的 AI 处理流程。 该库基于 Python 的 asyncio 框架，支持异步和并发执行，能够高效处理实时数据流，例如音频和视频。 通过将复杂的任务分解为可复用的单元，开发者可以构建出更易于维护、测试和扩展的 AI 应用。

功能列表

  • 模块化设计: 将复杂的 AI 任务分解为独立的、可复用的“处理器”单元，可以像积木一样自由组合。
  • 异步与并发: 基于 asyncio 构建，能够高效处理网络请求和数据流，实现并发执行，优化实时应用的响应速度。
  • 简化组合: 使用简单的运算符（+ 用于串联，// 用于并联）即可将多个处理器组合成复杂的处理流水线。
  • 丰富的内容处理: 通过 ProcessorPart 对象统一处理文本、图像、音频和自定义 JSON 等多种类型的数据。
  • 高度可扩展: 开发者可以通过继承基类或使用函数装饰器轻松创建符合自己需求的自定义处理器。
  • 集成 Gemini API: 内置了与 Gemini 模型交互的处理器，包括用于常规调用的 GenaiModel 和用于实时流式交互的 LiveProcessor。
  • 流式处理: 将所有输入和输出都视作异步数据流，特别适合构建需要实时响应的AI智能体应用。

使用帮助

GenAI Processors 库的目的是让构建复杂的AI应用变得简单。下面将详细介绍如何安装和使用该库。

1. 安装

首先，你需要一个 Python 3.10 或更高版本的环境。可以使用 pip 命令轻松完成安装：

pip install genai-processors

2. 核心概念

在开始使用之前，理解两个核心概念至关重要：

  • Processor (处理器): 这是库中最核心的构建模块。 它可以是一个简单的函数，也可以是一个复杂的类，但都遵循一个统一的接口：接收一个数据流作为输入，并返回一个数据流作为输出。 每个处理器都封装了一个特定的工作单元，比如调用一次模型、转换数据格式或从API获取信息。
  • ProcessorPart (处理器部件): 这是处理器在数据流中处理的基本单位。 它不仅仅是原始数据（如文本或图像），还包含了元数据，例如数据的MIME类型、角色（用户或模型）以及其他自定义属性。 这种设计使得在处理流程中可以轻松传递和使用结构化数据。

3. 基本使用流程

使用 GenAI Processors 的基本流程非常直观。

步骤一: 创建数据流

你可以从一个简单的列表开始，库会自动将其转换为异步数据流。

from genai_processors import content_api
from genai_processors import streams
# 输入内容可以包含字符串或更复杂的 ProcessorPart 对象
input_parts = ["你好", content_api.ProcessorPart("世界")]
# 将输入内容转换为异步流
input_stream = streams.stream_content(input_parts)

步骤二: 定义并应用处理器

处理器可以是一个简单的异步函数。下面的例子定义了一个将文本转为大写的处理器。

import asyncio
from typing import AsyncIterable
from genai_processors import content_api
# 任何实现了 call 方法的类或拥有特定签名的异步函数都可以作为处理器
async def uppercase_processor(
content: AsyncIterable[content_api.ProcessorPart]
) -> AsyncIterable[content_api.ProcessorPart]:
async for part in content:
if part.text:
yield content_api.ProcessorPart(text=part.text.upper())
# 使用示例
async def main():
input_parts = ["Hello", "World"]
input_stream = streams.stream_content(input_parts)
# 将处理器应用到数据流上，并遍历输出
async for part in uppercase_processor(input_stream):
print(part.text)
# 运行异步主函数
# asyncio.run(main())
# 输出:
# HELLO
# WORLD

4. 组合处理器

这是 GenAI Processors 库最强大的功能之一。你可以使用简单的操作符来构建复杂的逻辑。

  • 串联 (+): 将多个处理器按顺序连接起来，前一个的输出会成为后一个的输入。
  • 并联 (//): 让多个处理器并行处理同一个输入流。
# 假设我们还有另一个处理器
async def exclamation_processor(
content: AsyncIterable[content_api.ProcessorPart]
) -> AsyncIterable[content_api.ProcessorPart]:
async for part in content:
if part.text:
yield content_api.ProcessorPart(text=part.text + "!")
# 使用 "+" 将两个处理器串联起来
chained_processor = uppercase_processor + exclamation_processor
# 运行串联后的处理器
# async for part in chained_processor(input_stream):
#     print(part.text)
# 输出:
# HELLO!
# WORLD!

5. 与 Gemini 模型交互

该库内置了与 Gemini API 交互的处理器，让模型调用变得异常简单。

import os
from genai_processors.processors import gemini
# 需要先设置你的 Google API 密钥
# os.environ['GOOGLE_API_KEY'] = 'YOUR_API_KEY'
# 初始化一个Gemini处理器
gemini_processor = gemini.GenaiModel(model_name="gemini-1.5-flash")
# 创建一个包含提示的输入流
prompt_stream = streams.stream_content(
[content_api.ProcessorPart("请讲一个关于程序员的笑话")]
)
# 直接调用处理器
# async for response_part in gemini_processor(prompt_stream):
#    print(response_part.text)

6. 官方示例

为了更好地理解如何构建真实世界的应用，强烈建议查看官方仓库中 examples/ 目录下的示例。

  • 实时语音智能体 (Real-Time Live Example): 演示了如何构建一个能接收音频输入并实时语音回应的智能体。
  • 研究智能体 (Research Agent Example): 展示了如何通过组合多个子处理器（例如搜索、分析、总结）来构建一个自动化的研究工具。

通过学习这些示例，你可以掌握如何利用 GenAI Processors 的模块化和并发能力来解决复杂的 AI 开发挑战。

应用场景

  1. 实时多模态智能体利用库的异步流式处理能力，可以开发能够同时处理音频、视频和文本输入的智能体，并进行实时互动，例如在线会议的实时翻译和内容摘要、或者为视频游戏提供实时解说的AI评论员。
  2. 自动化内容创作通过串联不同的处理器，可以构建一个自动化的内容创作流水线。例如，第一个处理器根据关键词生成大纲，第二个处理器调用Gemini模型扩写成文章，第三个处理器生成配图，最后将所有内容整合发布到博客或文档平台。
  3. 复杂的研究与分析工具构建一个研究智能体，它可以接收一个研究主题，然后并行执行多个任务：一个处理器负责在网络上搜索相关论文和资料，另一个处理器负责分析和提取关键信息，第三个处理器将结果进行总结和结构化，最终生成一份全面的研究报告。
  4. 模块化的企业级AI解决方案在企业环境中，可以将数据预处理、合规性检查、调用内部模型、生成报告等多个环节分别封装成独立的处理器。这使得AI工作流的开发和维护更加清晰、可靠，并且各个模块可以被不同团队复用。

QA

  1. 这个库对Python版本有什么要求？GenAI Processors 需要 Python 3.10 或更高的版本。
  2. GenAI Processors 和直接调用 Gemini API 有什么不同？直接调用API通常需要自行处理复杂的异步逻辑、数据流管理、错误重试以及多模态数据的整合。GenAI Processors 将这些复杂的底层操作进行了封装，提供了一个更高层次的抽象（Processor），让开发者可以专注于业务逻辑的编排，而不是底层实现细节，从而更高效地构建可维护的应用。
  3. 这个库只能处理文本吗？不是。该库通过 ProcessorPart 的设计，原生支持多种数据类型，包括文本（text）、图像（image）、音频（audio）甚至是自定义的JSON数据结构，非常适合开发多模态AI应用。
  4. 如何将我自己的功能封装成一个处理器？你可以通过两种主要方式创建自定义处理器：一是编写一个符合特定签名的异步函数（接收一个异步可迭代对象，返回一个异步可迭代对象）；二是通过继承库提供的 processor.Processor 基类并实现其 call 方法。这两种方式都非常灵活，能够满足绝大多数定制需求。
微信微博Email复制链接