package conversion import "nex/backend/internal/conversion/canonical" // StreamDecoder 流式解码器接口 type StreamDecoder interface { ProcessChunk(rawChunk []byte) []canonical.CanonicalStreamEvent Flush() []canonical.CanonicalStreamEvent } // StreamEncoder 流式编码器接口 type StreamEncoder interface { EncodeEvent(event canonical.CanonicalStreamEvent) [][]byte Flush() [][]byte } // StreamConverter 流式转换器接口 type StreamConverter interface { ProcessChunk(rawChunk []byte) [][]byte Flush() [][]byte } // PassthroughStreamConverter 同协议透传流式转换器 type PassthroughStreamConverter struct{} // NewPassthroughStreamConverter 创建透传流式转换器 func NewPassthroughStreamConverter() *PassthroughStreamConverter { return &PassthroughStreamConverter{} } // ProcessChunk 直接传递原始字节 func (c *PassthroughStreamConverter) ProcessChunk(rawChunk []byte) [][]byte { return [][]byte{rawChunk} } // Flush 无缓冲数据 func (c *PassthroughStreamConverter) Flush() [][]byte { return nil } // SmartPassthroughStreamConverter 同协议 Smart Passthrough 流式转换器 // 逐 chunk 改写 model 字段 type SmartPassthroughStreamConverter struct { adapter ProtocolAdapter modelOverride string interfaceType InterfaceType } // NewSmartPassthroughStreamConverter 创建 Smart Passthrough 流式转换器 func NewSmartPassthroughStreamConverter(adapter ProtocolAdapter, modelOverride string, interfaceType InterfaceType) *SmartPassthroughStreamConverter { return &SmartPassthroughStreamConverter{ adapter: adapter, modelOverride: modelOverride, interfaceType: interfaceType, } } // ProcessChunk 改写 chunk 中的 model 字段 func (c *SmartPassthroughStreamConverter) ProcessChunk(rawChunk []byte) [][]byte { if len(rawChunk) == 0 { return nil } rewrittenChunk, err := c.adapter.RewriteResponseModelName(rawChunk, c.modelOverride, c.interfaceType) if err != nil { // 改写失败,返回原始 chunk return [][]byte{rawChunk} } return [][]byte{rewrittenChunk} } // Flush 无缓冲数据 func (c *SmartPassthroughStreamConverter) Flush() [][]byte { return nil } // CanonicalStreamConverter 跨协议规范流式转换器 type CanonicalStreamConverter struct { decoder StreamDecoder encoder StreamEncoder chain *MiddlewareChain ctx ConversionContext clientProtocol string providerProtocol string modelOverride string } // NewCanonicalStreamConverter 创建规范流式转换器 func NewCanonicalStreamConverter(decoder StreamDecoder, encoder StreamEncoder) *CanonicalStreamConverter { return &CanonicalStreamConverter{ decoder: decoder, encoder: encoder, } } // NewCanonicalStreamConverterWithMiddleware 创建带中间件的规范流式转换器 func NewCanonicalStreamConverterWithMiddleware(decoder StreamDecoder, encoder StreamEncoder, chain *MiddlewareChain, ctx ConversionContext, clientProtocol, providerProtocol, modelOverride string) *CanonicalStreamConverter { return &CanonicalStreamConverter{ decoder: decoder, encoder: encoder, chain: chain, ctx: ctx, clientProtocol: clientProtocol, providerProtocol: providerProtocol, modelOverride: modelOverride, } } // ProcessChunk 解码 → 中间件 → modelOverride → 编码管道 func (c *CanonicalStreamConverter) ProcessChunk(rawChunk []byte) [][]byte { events := c.decoder.ProcessChunk(rawChunk) var result [][]byte for i := range events { if c.chain != nil { processed, err := c.chain.ApplyStreamEvent(&events[i], c.clientProtocol, c.providerProtocol, &c.ctx) if err != nil { continue } events[i] = *processed } c.applyModelOverride(&events[i]) chunks := c.encoder.EncodeEvent(events[i]) result = append(result, chunks...) } return result } // Flush 刷新解码器和编码器缓冲区 func (c *CanonicalStreamConverter) Flush() [][]byte { events := c.decoder.Flush() var result [][]byte for i := range events { if c.chain != nil { processed, err := c.chain.ApplyStreamEvent(&events[i], c.clientProtocol, c.providerProtocol, &c.ctx) if err != nil { continue } events[i] = *processed } c.applyModelOverride(&events[i]) chunks := c.encoder.EncodeEvent(events[i]) result = append(result, chunks...) } encoderChunks := c.encoder.Flush() result = append(result, encoderChunks...) return result } // applyModelOverride 在跨协议场景下覆写流式事件中的 Model 字段 func (c *CanonicalStreamConverter) applyModelOverride(event *canonical.CanonicalStreamEvent) { if c.modelOverride != "" && event.Message != nil { event.Message.Model = c.modelOverride } }