构建复杂的AI应用，特别是处理多模态输入并需要实时响应的应用，经常感觉像是在拼装复杂拼图：需要将不同的数据处理步骤、异步API调用和自定义逻辑拼接在一起。随着复杂性的增长，这可能导致脆弱、难以维护的代码。2025年7月，Google DeepMind发布了GenAI Processors，这是一个专为解决这些技术挑战而设计的开源Python库。GenAI Processors的核心创新在于其ProcessorParts数据结构。每个ProcessorPart可以被视为标准化的数据部分（例如，音频块、文本转录、图像帧），它们携带相关的元数据在管道中流动。这种设计有几个关键技术优势：classProcessorPart:content rocessorContent#实际数据载荷metadata ict[str,Any]#元数据字典mime_type:str#MIME类型标识timestamp:float#时间戳sequence_id:str#序列标识符 - 异步流处理能力库提供了用于分割、连接和合并ProcessorParts异步流的实用工具。这意味着数据可以在不阻塞主线程的情况下连续处理：
asyncdefprocess_stream(input_stream:AsyncIterator[ProcessorPart])->AsyncIterator[ProcessorPart]:asyncforpartininput_stream:#处理每个部分processed_part=awaittransform_part(part)yieldprocessed_part - 双向流控制与传统的单向数据流不同，GenAI Processors支持双向流控制，允许下游处理器向上游发送反馈信息
classBidirectionalProcessor:asyncdefprocess(self,input_stream,feedback_stream):#同时处理输入和反馈asyncforinput_part,feedback_partinzip(input_stream,feedback_stream):result=awaitself.handle_with_feedback(input_part,feedback_part)yieldresult 每个Processor都实现了标准接口，这提供了强大的组合能力：classProcessor(ABC) abstractmethodasyncdefprocess(self,input_stream:AsyncIterator[ProcessorPart])->AsyncIterator[ProcessorPart]:passdef__call__(self,input_stream):returnself.process(input_stream) #处理链组合audio_processor=AudioTranscriber()text_processor=TextAnalyzer()response_generator=ResponseGenerator()#链式处理asyncdefprocess_audio_input(audio_stream):transcribed=audio_processor(audio_stream)analyzed=text_processor(transcribed)responses=response_generator(analyzed)returnresponses GenAI Processors库需要Python 3.10+。这个版本要求确保了对现代异步特性的完全支持：pip install genai-processors核心模块core/目录包含一组基本处理器，可以在你自己的应用程序中使用。它包括大多数实时应用程序所需的通用构建块。- AudioProcessor: 处理音频数据的专用处理器
- ModelProcessor: 与AI模型交互的处理器
- StreamSplitter: 将单一流分割为多个并行流
- FilterProcessor: 基于条件过滤数据
- TransformProcessor: 数据格式转换
该库提供了与Google Gemini API的现成连接器，包括同步的基于文本的调用和用于流式应用的Gemini Live API。1.同步文本处理fromgenai_processors.modelsimportGeminiTextProcessortext_processor=GeminiTextProcessor(model_name="gemini-pro",api_key="your-api-key",temperature=0.7,max_tokens=1000)asyncdefprocess_text_query(query:str):input_part=ProcessorPart(content=TextContent(query),metadata={"user_id":"123","session_id":"abc"})asyncforresponse_partintext_processor(async_iter([input_part])):returnresponse_part.content.text2.LiveAPI流式处理fromgenai_processors.modelsimportGeminiLiveProcessorlive_processor=GeminiLiveProcessor(model_name="gemini-live",api_key="your-api-key",streaming=True,real_time_factor=1.0)asyncdefhandle_live_audio(audio_stream):asyncforaudio_chunkinaudio_stream:input_part=ProcessorPart(content=AudioContent(audio_chunk),metadata={"format":"wav","sample_rate":16000})asyncforresponseinlive_processor(async_iter([input_part])):ifresponse.content.type=="audio":yieldresponse.content.audio_dataelifresponse.content.type=="text":print(f"Transcription:{response.content.text}")GenAI Processors的异步设计带来了几个关键的性能优势：1. 非阻塞I/O处理传统的同步处理在等待API响应时会阻塞整个线程。GenAI Processors通过异步设计避免了这个问题：classAsyncModelProcessor:asyncdefprocess_batch(self,inputs ist[ProcessorPart]):#并发处理多个输入tasks=[self.process_single(input_part)forinput_partininputs]results=awaitasyncio.gather(*tasks)returnresultsasyncdefprocess_single(self,input_partrocessorPart):#异步API调用asyncwithaiohttp.ClientSession()assession:response=awaitsession.post(self.api_endpoint,json=input_part.to_dict())returnProcessorPart.from_response(awaitresponse.json()) 2.只需几行代码即可使用 Gemini Live API 轻松构建能够实时处理音频和视频流的“Live Agent”。在以下示例中，使用+运算符组合输入源和处理步骤，从而创建清晰的数据流fromgenai_processors.coreimportaudio_io,live_model,video#Inputprocessor:combinescamerastreamsandaudiostreamsinput_processor=video.VideoIn()+audio_io.PyAudioIn(...)#Outputprocessor:playstheaudioparts.Handlesinterruptionsandpauses#audiooutputwhentheuserisspeaking.play_output=audio_io.PyAudioOut(...)#GeminiLiveAPIprocessorlive_processor=live_model.LiveProcessor(...)#Composetheagent:mic+camera->GeminiLiveAPI->playaudiolive_processor=live_model.LiveProcessor(...)live_agent=input_processor+live_processor+play_outputasyncforpartinlive_agent(streams.endless_stream()):#Processtheoutputparts(e.g.,printtranscription,modeloutput,metadata)print(part)
classBatchProcessor:def__init__(self,batch_size:int=32,max_concurrency:int=10):self.batch_size=batch_sizeself.semaphore=asyncio.Semaphore(max_concurrency)asyncdefprocess_batch(self,input_stream):batch=[]asyncforitemininput_stream:batch.append(item)iflen(batch)>=self.batch_size:asyncwithself.semaphore:results=awaitself.process_batch_items(batch)forresultinresults:yieldresultbatch=[]ifbatch:asyncwithself.semaphore:results=awaitself.process_batch_items(batch)forresultinresults:yieldresult GenAI Processors实现了智能的背压控制机制和内存管理和资源清理，当然它也支持自定义Processor。创建自定义处理器的典型步骤包括创建Processor或PartProcessor，classCustomAudioProcessor(Processor):def__init__(self,model_path:str,configict[str,Any]):self.model=load_model(model_path)self.config=configasyncdefprocess(self,input_stream:AsyncIterator[ProcessorPart])->AsyncIterator[ProcessorPart]:asyncforpartininput_stream:#验证输入类型ifnotisinstance(part.content,AudioContent):raiseValueError(f"ExpectedAudioContent,got{type(part.content)}")audio_data=awaitself.preprocess_audio(part.content.audio_data)result=awaitself.model.predict(audio_data)#创建输出ProcessorPartoutput_part=ProcessorPart(content=TextContent(result.transcription),metadata={**part.metadata,'confidence':result.confidence,'processing_time':result.processing_time})yieldoutput_partasyncdefpreprocess_audio(self,audio_data:bytes)->np.ndarray:#音频预处理逻辑audio_array=np.frombuffer(audio_data,dtype=np.int16)#标准化audio_array=audio_array.astype(np.float32)/32768.0#重采样到目标频率ifself.config.get('target_sample_rate'):audio_array=resample(audio_array,self.config['target_sample_rate'])returnaudio_arrayPartProcessor的高级用法对于需要更细粒度控制的场景，可以使用PartProcessor：classAdvancedPartProcessor(PartProcessor):asyncdefprocess_part(self,partrocessorPart)->AsyncIterator[ProcessorPart]:#检查是否需要分割大型数据ifpart.content.size>self.max_chunk_size:#分割为较小的块chunks=awaitself.split_content(part.content)fori,chunkinenumerate(chunks):chunk_part=ProcessorPart(content=chunk,metadata={**part.metadata,'chunk_index':i,'total_chunks':len(chunks)})processed_chunk=awaitself.process_chunk(chunk_part)yieldprocessed_chunkelse:#直接处理小数据result=awaitself.process_single_part(part)yieldresult 与现有技术的深度对比，相比Apache Kafka Streams而言，它是AI原生设计，专门为AI工作负载设计，内置了对多模态数据和AI模型的支持，Kafka Streams需要额外的适配层来处理AI特定的数据类型GenAI Processors通过其创新的ProcessorParts流式架构和统一的Processor接口，为AI应用开发提供了一个强大而灵活的基础设施。这些"模型处理器"抽象了批处理、上下文管理和流式I/O的复杂性，使得交互式系统的快速原型开发成为可能。ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;letter-spacing: 0.544px;line-height: 1.6em;margin: 0px;text-indent: 0em;outline: 0px;visibility: visible;text-wrap: wrap;caret-color: rgba(0, 0, 0, 0.9);-webkit-tap-highlight-color: rgba(0, 0, 0, 0);text-size-adjust: 100%;"> |
