InjectOpenAIStream wraps a data channel to inject tool_call chunks into a real OpenAI streaming response.
(dataChan <-chan []byte, rule *config.ToolCallInjectionRule, modelName string)
| 36 | // InjectOpenAIStream wraps a data channel to inject tool_call chunks into a |
| 37 | // real OpenAI streaming response. |
| 38 | func InjectOpenAIStream(dataChan <-chan []byte, rule *config.ToolCallInjectionRule, modelName string) <-chan []byte { |
| 39 | out := make(chan []byte, 16) |
| 40 | go func() { |
| 41 | defer close(out) |
| 42 | |
| 43 | var streamID string |
| 44 | created := time.Now().Unix() |
| 45 | injected := false |
| 46 | |
| 47 | for chunk := range dataChan { |
| 48 | // Extract stream metadata from early chunks (chunks are raw JSON). |
| 49 | if streamID == "" { |
| 50 | if id := gjson.GetBytes(chunk, "id").String(); id != "" { |
| 51 | streamID = id |
| 52 | } |
| 53 | if c := gjson.GetBytes(chunk, "created").Int(); c > 0 { |
| 54 | created = c |
| 55 | } |
| 56 | } |
| 57 | |
| 58 | // Detect terminal chunk: finish_reason is non-null. |
| 59 | if !injected { |
| 60 | fr := gjson.GetBytes(chunk, "choices.0.finish_reason") |
| 61 | if fr.Exists() && fr.Type != gjson.Null { |
| 62 | argsJSON, _ := json.Marshal(rule.Arguments) |
| 63 | callID := GenerateOpenAIToolCallID(rule.TaskID) |
| 64 | |
| 65 | // Emit tool_call start chunk (raw JSON). |
| 66 | startDelta := map[string]any{ |
| 67 | "role": "assistant", |
| 68 | "content": nil, |
| 69 | "tool_calls": []map[string]any{{ |
| 70 | "index": 0, |
| 71 | "id": callID, |
| 72 | "type": "function", |
| 73 | "function": map[string]any{ |
| 74 | "name": rule.ToolName, |
| 75 | "arguments": "", |
| 76 | }, |
| 77 | }}, |
| 78 | } |
| 79 | out <- buildOpenAIChunkJSON(streamID, modelName, created, startDelta) |
| 80 | |
| 81 | // Emit arguments chunk. |
| 82 | argsDelta := map[string]any{ |
| 83 | "tool_calls": []map[string]any{{ |
| 84 | "index": 0, |
| 85 | "function": map[string]any{ |
| 86 | "arguments": string(argsJSON), |
| 87 | }, |
| 88 | }}, |
| 89 | } |
| 90 | out <- buildOpenAIChunkJSON(streamID, modelName, created, argsDelta) |
| 91 | |
| 92 | // Change finish_reason to "tool_calls". |
| 93 | chunk, _ = sjson.SetBytes(chunk, "choices.0.finish_reason", "tool_calls") |
| 94 | injected = true |
| 95 | } |
no test coverage detected