feat: stream AI response tokens via SSE #248
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -77,3 +77,58 @@ func (p *OllamaProvider) Generate(ctx context.Context, text string) (string, err | |
|
|
||
| return response, nil | ||
| } | ||
|
|
||
| type ollamaStream struct { | ||
| resp *http.Response | ||
| decoder *json.Decoder | ||
| } | ||
|
|
||
| func (s *ollamaStream) Recv() (string, error) { | ||
| var chunk map[string]interface{} | ||
| if err := s.decoder.Decode(&chunk); err != nil { | ||
| return "", err // will return io.EOF when stream ends | ||
| } | ||
| if response, ok := chunk["response"].(string); ok { | ||
| return response, nil | ||
| } | ||
| return "", fmt.Errorf("invalid response from Ollama: missing response field") | ||
| } | ||
|
Comment on lines
+86
to
+95
Contributor
There was a problem hiding this comment. 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
echo "=== relevant files ==="
git ls-files | rg 'gateway/internal/ai/ollama.go|go.mod|go.sum|README|openapi|ollama|ai'
echo
echo "=== ollama.go excerpt ==="
cat -n gateway/internal/ai/ollama.go | sed -n '1,220p'
echo
echo "=== dependency versions mentioning ollama ==="
rg -n '"ollama|github.com/ollama' go.mod go.sum gateway -g '!**/node_modules/**'
echo
echo "=== repo mentions of /api/generate or error field ==="
rg -n 'api/generate|"\berror\b"|missing response field|ollama' gateway README.md . -g '!**/node_modules/**'Repository: AnkanMisra/MicroAI-Paygate Length of output: 6137 🌐 Web query:
💡 Result: When using the Ollama /api/generate endpoint with streaming enabled (the default), the API uses newline-delimited JSON (NDJSON) [1][2]. Each chunk in the stream is an independent JSON object sent on a new line [2][3]. If an error occurs mid-stream, Ollama sends an error object as a JSON chunk within the NDJSON stream [4][5]. This error chunk typically contains an "error" field describing the issue [5][6][7]. Because the HTTP connection is already open and partial data may have been sent, the API does not change the HTTP status code when this occurs [5]. To handle this, client implementations must parse each line of the stream independently [2][3]. If a chunk contains an "error" field, the client should treat it as an error and stop processing the stream [5][7]. Example of an error chunk in a stream: {"error":"an error was encountered while running the model"} [5] If you prefer to receive a single, standard JSON response instead of a stream—which simplifies error handling by allowing the API to return a traditional HTTP error status—you can disable streaming by setting "stream": false in your request body [1][8]. Citations:
Surface Ollama stream errors 🤖 Prompt for AI Agents |
||
|
|
||
| func (s *ollamaStream) Close() error { | ||
| return s.resp.Body.Close() | ||
| } | ||
|
|
||
| func (p *OllamaProvider) GenerateStream(ctx context.Context, text string) (Stream, error) { | ||
| prompt := fmt.Sprintf("Summarize this text in 2 sentences: %s", text) | ||
|
|
||
| reqBody, _ := json.Marshal(map[string]interface{}{ | ||
| "model": p.model, | ||
| "prompt": prompt, | ||
| "stream": true, | ||
| }) | ||
|
|
||
| req, err := http.NewRequestWithContext(ctx, "POST", p.url+"/api/generate", bytes.NewBuffer(reqBody)) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to create Ollama request: %w", err) | ||
| } | ||
| req.Header.Set("Content-Type", "application/json") | ||
|
|
||
| resp, err := http.DefaultClient.Do(req) | ||
| if err != nil { | ||
| if errors.Is(err, context.DeadlineExceeded) || ctx.Err() == context.DeadlineExceeded { | ||
| return nil, context.DeadlineExceeded | ||
| } | ||
| return nil, fmt.Errorf("failed to connect to Ollama: %w", err) | ||
| } | ||
|
|
||
| if resp.StatusCode != http.StatusOK { | ||
| body, _ := io.ReadAll(resp.Body) | ||
| resp.Body.Close() | ||
| return nil, fmt.Errorf("ollama returned status %d: %s", resp.StatusCode, string(body)) | ||
| } | ||
|
|
||
| return &ollamaStream{ | ||
| resp: resp, | ||
| decoder: json.NewDecoder(resp.Body), | ||
| }, nil | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,6 +1,7 @@ | ||
| package ai | ||
|
|
||
| import ( | ||
| "bufio" | ||
| "bytes" | ||
| "context" | ||
| "encoding/json" | ||
|
|
@@ -100,3 +101,93 @@ func (p *OpenRouterProvider) Generate(ctx context.Context, text string) (string, | |
|
|
||
| return content, nil | ||
| } | ||
|
|
||
| type openRouterStream struct { | ||
| resp *http.Response | ||
| reader *bufio.Reader | ||
| sawDone bool | ||
| } | ||
|
|
||
| func (s *openRouterStream) Recv() (string, error) { | ||
| for { | ||
| line, err := s.reader.ReadBytes('\n') | ||
| if err != nil { | ||
| if errors.Is(err, io.EOF) && !s.sawDone { | ||
| return "", fmt.Errorf("stream ended without [DONE] sentinel") | ||
| } | ||
| return "", err | ||
|
Comment on lines
+114
to
+118
Contributor
There was a problem hiding this comment.
A bare Useful? React with 👍 / 👎. |
||
| } | ||
| line = bytes.TrimSpace(line) | ||
| if len(line) == 0 { | ||
| continue | ||
| } | ||
| if bytes.HasPrefix(line, []byte("data: ")) { | ||
| data := bytes.TrimPrefix(line, []byte("data: ")) | ||
| if string(data) == "[DONE]" { | ||
| s.sawDone = true | ||
| return "", io.EOF | ||
| } | ||
| var chunk map[string]interface{} | ||
| if err := json.Unmarshal(data, &chunk); err != nil { | ||
| continue // skip invalid json | ||
| } | ||
| choices, ok := chunk["choices"].([]interface{}) | ||
| if !ok || len(choices) == 0 { | ||
| continue | ||
| } | ||
| choice, ok := choices[0].(map[string]interface{}) | ||
| if !ok { | ||
| continue | ||
| } | ||
| delta, ok := choice["delta"].(map[string]interface{}) | ||
| if !ok { | ||
| continue | ||
| } | ||
| if content, ok := delta["content"].(string); ok { | ||
| return content, nil | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
Comment on lines
+111
to
+151
Contributor
There was a problem hiding this comment. 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
printf '\n== File listing ==\n'
git ls-files 'gateway/internal/ai/openrouter.go' 'gateway/internal/ai/*' | sed -n '1,120p'
printf '\n== Outline ==\n'
ast-grep outline gateway/internal/ai/openrouter.go --view expanded || true
printf '\n== Relevant lines from gateway/internal/ai/openrouter.go ==\n'
cat -n gateway/internal/ai/openrouter.go | sed -n '1,240p'Repository: AnkanMisra/MicroAI-Paygate Length of output: 7140 🌐 Web query:
💡 Result: OpenRouter handles streaming errors differently depending on whether tokens have already been sent to the client [1][2]. If an error occurs before any tokens are streamed, the API returns a standard JSON error response with the appropriate HTTP status code [1][2]. If an error occurs mid-stream (after tokens have been sent), the HTTP status code remains 200 OK because headers were already transmitted [1][2]. In this case, the error is delivered as a Server-Sent Event (SSE) containing a unified JSON payload [1][2]. The structure of this mid-stream error SSE payload is as follows [2]: data: {"id":"cmpl-abc123","object":"chat.completion.chunk","created":1234567890,"model":"model-name","provider":"provider-name","error":{"code":"error_code","message":"error message"},"choices":[{"index":0,"delta":{"content":""},"finish_reason":"error"}]} Key characteristics of this payload include: - The error object is included at the top level of the JSON alongside standard fields like id, object, and model [1][2]. - A choices array is provided with finish_reason set to "error" to formally terminate the stream [1][2]. - The stream is terminated by the server immediately after this event is sent [3][2]. OpenRouter normalizes finish_reason values across different models, and "error" is one of the standardized values used to indicate an abnormal stream termination [4]. Citations:
🏁 Script executed: #!/bin/bash
set -euo pipefail
printf '\n== Search for OpenRouter stream handling in repo ==\n'
rg -n "finish_reason|\\[DONE\\]|data: \\{\"error\"|openrouter" gateway -g '*.go' -g '*.md' -g '*.yaml' | sed -n '1,200p'Repository: AnkanMisra/MicroAI-Paygate Length of output: 2763 🏁 Script executed: #!/bin/bash
set -euo pipefail
python3 - <<'PY'
import json, textwrap, sys, re
from pathlib import Path
p = Path('gateway/internal/ai/openrouter.go')
text = p.read_text()
for i, line in enumerate(text.splitlines(), 1):
if 100 <= i <= 170:
print(f"{i:4}: {line}")
PYRepository: AnkanMisra/MicroAI-Paygate Length of output: 2302 Surface OpenRouter stream errors in 🤖 Prompt for AI Agents |
||
|
|
||
| func (s *openRouterStream) Close() error { | ||
| return s.resp.Body.Close() | ||
| } | ||
|
|
||
| func (p *OpenRouterProvider) GenerateStream(ctx context.Context, text string) (Stream, error) { | ||
| prompt := fmt.Sprintf("Summarize this text in 2 sentences: %s", text) | ||
|
|
||
| reqBody, _ := json.Marshal(map[string]interface{}{ | ||
| "model": p.model, | ||
| "messages": []map[string]string{ | ||
| {"role": "user", "content": prompt}, | ||
| }, | ||
| "stream": true, | ||
| }) | ||
|
|
||
| req, err := http.NewRequestWithContext(ctx, "POST", p.url, bytes.NewBuffer(reqBody)) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to create OpenRouter request: %w", err) | ||
| } | ||
| req.Header.Set("Authorization", "Bearer "+p.apiKey) | ||
| req.Header.Set("Content-Type", "application/json") | ||
|
|
||
| resp, err := http.DefaultClient.Do(req) | ||
| if err != nil { | ||
| if errors.Is(err, context.DeadlineExceeded) || ctx.Err() == context.DeadlineExceeded { | ||
| return nil, context.DeadlineExceeded | ||
| } | ||
| return nil, err | ||
| } | ||
|
|
||
| if resp.StatusCode < 200 || resp.StatusCode >= 300 { | ||
| body, _ := io.ReadAll(resp.Body) | ||
| resp.Body.Close() | ||
| return nil, fmt.Errorf("openrouter returned status %d: %s", resp.StatusCode, string(body)) | ||
| } | ||
|
|
||
| return &openRouterStream{ | ||
| resp: resp, | ||
| reader: bufio.NewReader(resp.Body), | ||
| }, nil | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This new cache-miss branch no longer uses the
bodyBytes := writer.body.Bytes()assignment just above it;cd gateway && go test ./...fails with./cache.go:207:3: declared and not used: bodyBytes, so the gateway package cannot build. Remove the assignment or use it when parsing/caching the SSE result.Useful? React with 👍 / 👎.