package openai import ( "bufio" "bytes" "context" "encoding/json" "fmt" "io" "iter" "net/http" "strings" "github.com/VictorVargas/rony-llm-agent/pkg/llm" ) // Config holds the settings needed to create an OpenAI client. type Config struct { APIKey string Model string BaseURL string // defaults to https://api.openai.com/v1 } // Client implements llm.LLMClient for OpenAI. type Client struct { apiKey string model string baseURL string http *http.Client } // New returns a new OpenAI client. func New(cfg Config) (*Client, error) { if cfg.APIKey == "" { return nil, fmt.Errorf("openai: API key is required") } baseURL := cfg.BaseURL if baseURL == "" { baseURL = "https://api.openai.com/v1" } return &Client{ apiKey: cfg.APIKey, model: cfg.Model, baseURL: baseURL, http: http.DefaultClient, }, nil } func (c *Client) Generate(ctx context.Context, req llm.CompletionRequest) (llm.CompletionResponse, error) { endpoint := c.baseURL + "/chat/completions" payload, err := c.buildRequest(req, false) if err != nil { return llm.CompletionResponse{}, fmt.Errorf("building request: %w", err) } httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, payload) if err != nil { return llm.CompletionResponse{}, fmt.Errorf("creating request: %w", err) } httpReq.Header.Set("Authorization", "Bearer "+c.apiKey) httpReq.Header.Set("Content-Type", "application/json") resp, err := c.http.Do(httpReq) if err != nil { return llm.CompletionResponse{}, fmt.Errorf("request failed: %w", err) } defer resp.Body.Close() if resp.StatusCode < 200 || resp.StatusCode >= 300 { body, _ := io.ReadAll(resp.Body) return llm.CompletionResponse{}, fmt.Errorf("API error %d: %s", resp.StatusCode, string(body)) } var apiResp openaiChatResponse if err := json.NewDecoder(resp.Body).Decode(&apiResp); err != nil { return llm.CompletionResponse{}, fmt.Errorf("decoding response: %w", err) } return c.toResponse(apiResp) } func (c *Client) Stream(ctx context.Context, req llm.CompletionRequest) iter.Seq2[llm.StreamChunk, error] { return func(yield func(llm.StreamChunk, error) bool) { endpoint := c.baseURL + "/chat/completions" payload, err := c.buildRequest(req, true) if err != nil { yield(llm.StreamChunk{}, fmt.Errorf("building request: %w", err)) return } httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, payload) if err != nil { yield(llm.StreamChunk{}, fmt.Errorf("creating request: %w", err)) return } httpReq.Header.Set("Authorization", "Bearer "+c.apiKey) httpReq.Header.Set("Content-Type", "application/json") httpReq.Header.Set("Accept", "text/event-stream") resp, err := c.http.Do(httpReq) if err != nil { yield(llm.StreamChunk{}, fmt.Errorf("request failed: %w", err)) return } defer resp.Body.Close() if resp.StatusCode < 200 || resp.StatusCode >= 300 { body, _ := io.ReadAll(resp.Body) yield(llm.StreamChunk{}, fmt.Errorf("API error %d: %s", resp.StatusCode, string(body))) return } // toolCallAccum buffers one tool call's fragments as they stream in: // the SSE format sends the id/name in the first delta for a given // tool-call index and the (potentially large) arguments JSON in // pieces across many subsequent deltas, so it can't be handed to a // tool handler until it's fully assembled. Same accumulation the // llamacpp client does — without it, tool calls made over a stream // were silently dropped and the agent loop never executed them. type toolCallAccum struct { id string name string args strings.Builder } toolCallFrags := map[int]*toolCallAccum{} var toolCallOrder []int flushToolCalls := func() []llm.ToolCall { if len(toolCallOrder) == 0 { return nil } calls := make([]llm.ToolCall, 0, len(toolCallOrder)) for _, idx := range toolCallOrder { frag := toolCallFrags[idx] calls = append(calls, llm.ToolCall{ ID: frag.id, Name: frag.name, Arguments: json.RawMessage(frag.args.String()), }) } toolCallFrags = map[int]*toolCallAccum{} toolCallOrder = nil return calls } scanner := bufio.NewScanner(resp.Body) // A single SSE line can exceed bufio.Scanner's 64KB default cap // (e.g. a large tool-call arguments delta), which would kill the // stream with "token too long" — same headroom the anthropic // client already reserves. scanner.Buffer(make([]byte, 0, 64*1024), 4*1024*1024) for scanner.Scan() { line := scanner.Text() if !strings.HasPrefix(line, "data: ") { continue } data := strings.TrimPrefix(line, "data: ") if data == "[DONE]" { return } var event openaiStreamEvent if err := json.Unmarshal([]byte(data), &event); err != nil { yield(llm.StreamChunk{}, fmt.Errorf("decoding event: %w", err)) return } var usage llm.TokenUsage if event.Usage != nil { usage = llm.TokenUsage{ InputTokens: event.Usage.PromptTokens, OutputTokens: event.Usage.CompletionTokens, TotalTokens: event.Usage.TotalTokens, } } if len(event.Choices) == 0 { // The usage-only event (per stream_options.include_usage) // carries no choices, so it needs its own chunk. if event.Usage != nil { if !yield(llm.StreamChunk{Usage: usage}, nil) { return } } continue } for _, choice := range event.Choices { hasFragment := len(choice.Delta.ToolCalls) > 0 for _, tc := range choice.Delta.ToolCalls { frag, ok := toolCallFrags[tc.Index] if !ok { frag = &toolCallAccum{} toolCallFrags[tc.Index] = frag toolCallOrder = append(toolCallOrder, tc.Index) } if tc.ID != "" { frag.id = tc.ID } if tc.Function.Name != "" { frag.name = tc.Function.Name } frag.args.WriteString(tc.Function.Arguments) } chunk := llm.StreamChunk{ Delta: choice.Delta.Content, Usage: usage, } if choice.FinishReason != "" { chunk.FinishReason = choice.FinishReason chunk.ToolCalls = flushToolCalls() } // A fragment-only event (a piece of a tool call's streamed // arguments, with nothing else in this delta) has nothing // yet for the agent loop to act on: it was buffered above, // so skip yielding an empty chunk for it. if hasFragment && chunk.Delta == "" && chunk.FinishReason == "" { continue } if !yield(chunk, nil) { return } } } if err := scanner.Err(); err != nil { yield(llm.StreamChunk{}, fmt.Errorf("stream error: %w", err)) } } } func (c *Client) Name() string { return "openai" } func (c *Client) Capabilities() llm.ProviderCapabilities { return llm.ProviderCapabilities{ SupportsTools: true, SupportsVision: true, SupportsVideo: false, SupportsJSON: true, MaxContextWindow: 128000, } } // buildRequest converts an llm.CompletionRequest to the OpenAI API format. func (c *Client) buildRequest(req llm.CompletionRequest, stream bool) (io.Reader, error) { // Convert messages to OpenAI format messages := make([]openaiMessage, len(req.Messages)) for i, m := range req.Messages { content, err := buildContentValue(m) if err != nil { return nil, err } messages[i] = openaiMessage{ Role: string(m.Role), Content: content, ToolCallID: m.ToolCallID, Name: m.Name, } if len(m.ToolCalls) > 0 { calls := make([]openaiToolCall, len(m.ToolCalls)) for j, tc := range m.ToolCalls { calls[j] = openaiToolCall{ ID: tc.ID, Type: "function", Function: openaiFunction{ Name: tc.Name, Arguments: string(tc.Arguments), }, } } messages[i].ToolCalls = calls } } tools := make([]openaiTool, len(req.Tools)) for i, t := range req.Tools { var tool openaiTool if err := json.Unmarshal(t, &tool); err != nil { return nil, fmt.Errorf("parsing tool %d: %w", i, err) } tools[i] = tool } // The per-request model wins when set; otherwise fall back to the // client's configured one (Config.Model used to be discarded entirely, // so every request went out with an empty model — a hard API error on // OpenAI, and the agent loop never sets req.Model). model := req.Model if model == "" { model = c.model } openaiReq := openaiChatRequest{ Model: model, Messages: messages, Stream: stream, } if stream { // Ask for a final SSE event carrying token usage (OpenAI-style // streaming omits it otherwise), so callers can track real token // counts per turn instead of always seeing zero. openaiReq.StreamOptions = &openaiStreamOptions{IncludeUsage: true} } if len(tools) > 0 { openaiReq.Tools = tools } if req.ToolChoice != nil { openaiReq.ToolChoice = req.ToolChoice } if req.Temperature != nil { tmp := *req.Temperature openaiReq.Temperature = &tmp } if req.MaxTokens != nil { max := *req.MaxTokens openaiReq.MaxTokens = &max } if len(req.Stop) > 0 { openaiReq.Stop = req.Stop } data, err := json.Marshal(openaiReq) if err != nil { return nil, fmt.Errorf("marshaling request: %w", err) } return bytes.NewReader(data), nil } // toResponse converts an OpenAI API response to our CompletionResponse. func (c *Client) toResponse(resp openaiChatResponse) (llm.CompletionResponse, error) { if len(resp.Choices) == 0 { return llm.CompletionResponse{}, fmt.Errorf("openai: response contained no choices") } choice := resp.Choices[0] result := llm.CompletionResponse{ ID: resp.ID, Model: resp.Model, Content: choice.Message.Content, StopReason: choice.FinishReason, } for _, tc := range choice.Message.ToolCalls { result.ToolCalls = append(result.ToolCalls, llm.ToolCall{ ID: tc.ID, Name: tc.Function.Name, Arguments: json.RawMessage(tc.Function.Arguments), }) } result.Usage = llm.TokenUsage{ InputTokens: resp.Usage.PromptTokens, OutputTokens: resp.Usage.CompletionTokens, TotalTokens: resp.Usage.TotalTokens, } return result, nil } // OpenAI API types type openaiChatRequest struct { Model string `json:"model"` Messages []openaiMessage `json:"messages"` Tools []openaiTool `json:"tools,omitempty"` ToolChoice interface{} `json:"tool_choice,omitempty"` Temperature *float32 `json:"temperature,omitempty"` MaxTokens *int `json:"max_tokens,omitempty"` Stop []string `json:"stop,omitempty"` Stream bool `json:"stream"` StreamOptions *openaiStreamOptions `json:"stream_options,omitempty"` } type openaiStreamOptions struct { IncludeUsage bool `json:"include_usage"` } type openaiMessage struct { Role string `json:"role"` // Content is either a plain string (the common case) or a // []openaiContentPart when the source llm.Message carried Parts - see // buildContentValue. Content interface{} `json:"content"` ToolCallID string `json:"tool_call_id,omitempty"` Name string `json:"name,omitempty"` ToolCalls []openaiToolCall `json:"tool_calls,omitempty"` } // openaiContentPart is one block of a multipart "content" array, following // the same shape OpenAI's vision-capable chat completions endpoint expects. type openaiContentPart struct { Type string `json:"type"` Text string `json:"text,omitempty"` ImageURL *openaiMediaURL `json:"image_url,omitempty"` } type openaiMediaURL struct { URL string `json:"url"` } // buildContentValue converts an llm.Message's Parts into the OpenAI // multipart content shape, or falls back to the plain Content string when // there are no Parts - existing callers building a plain-text Message are // completely unaffected. A video part is rejected outright: OpenAI's chat // completions API has no video content type, so sending one would just // produce a confusing API error instead of this clear one. func buildContentValue(m llm.Message) (interface{}, error) { if len(m.Parts) == 0 { return m.Content, nil } parts := make([]openaiContentPart, 0, len(m.Parts)) for _, p := range m.Parts { switch p.Type { case "text": parts = append(parts, openaiContentPart{Type: "text", Text: p.Text}) case "image": parts = append(parts, openaiContentPart{Type: "image_url", ImageURL: &openaiMediaURL{URL: p.MediaURL}}) case "video": return nil, fmt.Errorf("openai: video attachments are not supported by the chat completions API") default: return nil, fmt.Errorf("openai: unknown content part type %q", p.Type) } } return parts, nil } type openaiTool struct { Type string `json:"type"` Function json.RawMessage `json:"function"` } type openaiChatResponse struct { ID string `json:"id"` Model string `json:"model"` Choices []openaiChoice `json:"choices"` Usage openaiUsage `json:"usage"` } type openaiChoice struct { Index int `json:"index"` Message openaiMessageResult `json:"message"` FinishReason string `json:"finish_reason"` } type openaiMessageResult struct { Role string `json:"role"` Content string `json:"content"` ToolCalls []openaiToolCall `json:"tool_calls"` } type openaiToolCall struct { ID string `json:"id"` Type string `json:"type"` Function openaiFunction `json:"function"` } type openaiFunction struct { Name string `json:"name"` Arguments string `json:"arguments"` } type openaiUsage struct { PromptTokens int `json:"prompt_tokens"` CompletionTokens int `json:"completion_tokens"` TotalTokens int `json:"total_tokens"` } // Stream event types type openaiStreamEvent struct { ID string `json:"id"` Choices []openaiStreamChoice `json:"choices"` Usage *openaiUsage `json:"usage"` } type openaiStreamChoice struct { Index int `json:"index"` Delta openaiStreamDelta `json:"delta"` FinishReason string `json:"finish_reason"` } type openaiStreamDelta struct { Content string `json:"content"` Role string `json:"role"` ToolCalls []openaiStreamToolCall `json:"tool_calls"` } type openaiStreamToolCall struct { Index int `json:"index"` ID string `json:"id"` Type string `json:"type"` Function openaiStreamFunction `json:"function"` } type openaiStreamFunction struct { Name string `json:"name"` Arguments string `json:"arguments"` }