Replace go-openai with native client
Drop the github.com/sashabaranov/go-openai dependency in favor of an internal HTTP client. The LLM package is now split into: - llm.go: type definitions (Message, ChatRequest, ChatResponse, etc.) - client.go: client construction and configuration - client_completions.go: Completions and CompletionsStream methods - client_models.go: ListModels - client_tools.go: tool call execution CompletionsStream uses typed SSE events (delta, done, message) so the frontend can render progressive streaming output. Move tool.go and tool_test.go from internal/tool/ into internal/llm/ to co-locate the registry with the client that consumes it. Update job.go and service.go to use the new llm.Message types and Completions/CompletionsStream methods. Remove the Voice field from chat request/response (voice is now tracked via message metadata). Frontend: handle delta/done/message SSE events, render a streaming placeholder that receives progressive text deltas, and finalize with full message binding (reasoning, tool calls, actions).
This commit is contained in:
dwrz
482
internal/llm/client_completions.go
Normal file
482
internal/llm/client_completions.go
Normal file
@@ -0,0 +1,482 @@
|
||||
package llm
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"slices"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// ErrMaxIterations is returned when the agent loop exceeds the configured
|
||||
// maximum number of iterations.
|
||||
var ErrMaxIterations = fmt.Errorf("max iterations exceeded")
|
||||
|
||||
// Completions sends messages to the LLM using the specified model.
|
||||
// If systemMessage is non-empty, it overrides the configured system message.
|
||||
// maxIterations caps the number of agent loop iterations.
|
||||
// A value of 0 means unlimited (no cap).
|
||||
// Returns all messages generated during the completions request, including
|
||||
// tool calls and tool results.
|
||||
func (c *Client) Completions(
|
||||
ctx context.Context,
|
||||
model string,
|
||||
systemMessage string,
|
||||
maxIterations int,
|
||||
messages []Message,
|
||||
) ([]Message, error) {
|
||||
ctx, cancel := context.WithTimeout(ctx, c.timeout)
|
||||
defer cancel()
|
||||
|
||||
// Use request system message if provided, config default otherwise.
|
||||
smsg := c.systemMessage
|
||||
if systemMessage != "" {
|
||||
smsg = systemMessage
|
||||
}
|
||||
// Prepend system message, if configured and not already present.
|
||||
if smsg != "" &&
|
||||
(len(messages) == 0 || messages[0].Role != RoleSystem) {
|
||||
messages = append(
|
||||
[]Message{{
|
||||
Role: RoleSystem,
|
||||
ContentParts: []ContentPart{{
|
||||
Type: ContentTypeText,
|
||||
Text: smsg,
|
||||
}},
|
||||
}},
|
||||
messages...,
|
||||
)
|
||||
}
|
||||
|
||||
// Remember the starting length so we can return only the messages
|
||||
// generated during this call (the suffix of the conversation history).
|
||||
start := len(messages)
|
||||
// Loop for tool calls. maxIterations == 0 means unlimited.
|
||||
for i := 0; maxIterations == 0 || i < maxIterations; i++ {
|
||||
cr, err := c.completions(ctx, model, messages)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("completions: %w", err)
|
||||
}
|
||||
// Use first choice for tool call loop.
|
||||
message := &cr.Choices[0].Message
|
||||
messages = append(messages, *message)
|
||||
if len(message.ToolCalls) == 0 {
|
||||
return messages[start:], nil
|
||||
}
|
||||
toolResults, err := c.callTools(ctx, *message)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("completions: %w", err)
|
||||
}
|
||||
messages = append(messages, toolResults...)
|
||||
}
|
||||
|
||||
return messages[start:], ErrMaxIterations
|
||||
}
|
||||
|
||||
// completions sends a single non-streaming request and returns the
|
||||
// full API response.
|
||||
func (c *Client) completions(
|
||||
ctx context.Context, model string, messages []Message,
|
||||
) (*ChatResponse, error) {
|
||||
body, err := json.Marshal(ChatRequest{
|
||||
Model: model,
|
||||
Messages: messages,
|
||||
Tools: c.tools,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf(
|
||||
"completions: marshal request body: %w", err,
|
||||
)
|
||||
}
|
||||
req, err := http.NewRequestWithContext(
|
||||
ctx,
|
||||
http.MethodPost,
|
||||
c.baseURL+"/chat/completions",
|
||||
bytes.NewReader(body),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("completions: create request: %w", err)
|
||||
}
|
||||
if c.key != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+c.key)
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
res, err := c.httpc.Do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("completions: %w", err)
|
||||
}
|
||||
defer res.Body.Close()
|
||||
|
||||
if res.StatusCode != http.StatusOK {
|
||||
body, err := io.ReadAll(res.Body)
|
||||
if err != nil {
|
||||
c.log.ErrorContext(
|
||||
ctx,
|
||||
"completions: failed to read error response body",
|
||||
slog.Any("error", err),
|
||||
)
|
||||
return nil, fmt.Errorf(
|
||||
"completions: status code %d", res.StatusCode,
|
||||
)
|
||||
}
|
||||
var apiError struct {
|
||||
Error *struct {
|
||||
Type string `json:"type"`
|
||||
Code string `json:"code"`
|
||||
Message string `json:"message"`
|
||||
} `json:"error"`
|
||||
}
|
||||
if err := json.Unmarshal(
|
||||
body, &apiError,
|
||||
); err == nil && apiError.Error != nil {
|
||||
c.log.ErrorContext(
|
||||
ctx,
|
||||
"completions: request error",
|
||||
slog.Int("status", res.StatusCode),
|
||||
slog.String("code", apiError.Error.Code),
|
||||
slog.String(
|
||||
"message", apiError.Error.Message,
|
||||
),
|
||||
slog.String("type", apiError.Error.Type),
|
||||
)
|
||||
return nil, fmt.Errorf(
|
||||
"completions: %s: %s",
|
||||
apiError.Error.Code,
|
||||
apiError.Error.Message,
|
||||
)
|
||||
}
|
||||
c.log.ErrorContext(
|
||||
ctx,
|
||||
"completions: API error",
|
||||
slog.Int("status", res.StatusCode),
|
||||
slog.String("body", string(body)),
|
||||
)
|
||||
return nil, fmt.Errorf(
|
||||
"completions: %d: %s",
|
||||
res.StatusCode, string(body),
|
||||
)
|
||||
}
|
||||
|
||||
var cr = &ChatResponse{}
|
||||
if err := json.NewDecoder(res.Body).Decode(cr); err != nil {
|
||||
return nil, fmt.Errorf("completions: decode response: %w", err)
|
||||
}
|
||||
if len(cr.Choices) == 0 {
|
||||
return nil, fmt.Errorf(
|
||||
"completions: no response choices returned",
|
||||
)
|
||||
}
|
||||
return cr, nil
|
||||
}
|
||||
|
||||
// StreamEvent wraps a message or delta produced during streaming.
|
||||
type StreamEvent struct {
|
||||
// Type identifies the event kind: delta, done, or message.
|
||||
Type string
|
||||
// Delta is a text fragment from the assistant response.
|
||||
Delta string
|
||||
// Message is the complete chat completion message.
|
||||
Message Message
|
||||
}
|
||||
|
||||
const (
|
||||
streamEventDelta = "delta"
|
||||
streamEventDone = "done"
|
||||
streamEventMessage = "message"
|
||||
)
|
||||
|
||||
// CompletionsStream sends messages to the LLM using the specified model and
|
||||
// streams results. Each complete message (assistant reply, tool call,
|
||||
// tool result) is sent to the events channel as it becomes available.
|
||||
// The channel is closed before returning.
|
||||
// If systemMessage is non-empty, it overrides the configured system message.
|
||||
// maxIterations caps the number of agent loop iterations; a value of 0
|
||||
// means unlimited (no cap).
|
||||
// Returns ErrMaxIterations if the iteration limit is exceeded.
|
||||
func (c *Client) CompletionsStream(
|
||||
ctx context.Context,
|
||||
model string,
|
||||
systemMessage string,
|
||||
maxIterations int,
|
||||
messages []Message,
|
||||
events chan<- StreamEvent,
|
||||
) error {
|
||||
defer close(events)
|
||||
ctx, cancel := context.WithTimeout(ctx, c.timeout)
|
||||
defer cancel()
|
||||
|
||||
// Use request system message if provided, config default otherwise.
|
||||
smsg := c.systemMessage
|
||||
if systemMessage != "" {
|
||||
smsg = systemMessage
|
||||
}
|
||||
|
||||
// Prepend system message, if configured and not already present.
|
||||
if smsg != "" &&
|
||||
(len(messages) == 0 || messages[0].Role != RoleSystem) {
|
||||
messages = append(
|
||||
[]Message{{
|
||||
Role: RoleSystem,
|
||||
ContentParts: []ContentPart{
|
||||
{Type: ContentTypeText, Text: smsg},
|
||||
},
|
||||
}},
|
||||
messages...,
|
||||
)
|
||||
}
|
||||
|
||||
// Loop for tool calls. maxIterations == 0 means unlimited.
|
||||
for i := 0; maxIterations == 0 || i < maxIterations; i++ {
|
||||
choices, err := c.completionsStream(ctx, messages, model, events)
|
||||
if err != nil {
|
||||
return fmt.Errorf("completions stream: %w", err)
|
||||
}
|
||||
// Use first choice for tool call loop.
|
||||
message := choices[0]
|
||||
if len(message.ToolCalls) == 0 {
|
||||
return nil
|
||||
}
|
||||
toolResults, err := c.callTools(ctx, message)
|
||||
if err != nil {
|
||||
return fmt.Errorf("completions stream: %w", err)
|
||||
}
|
||||
for _, tr := range toolResults {
|
||||
events <- StreamEvent{
|
||||
Type: streamEventMessage,
|
||||
Message: tr,
|
||||
}
|
||||
}
|
||||
messages = append(messages, choices...)
|
||||
messages = append(messages, toolResults...)
|
||||
}
|
||||
|
||||
return ErrMaxIterations
|
||||
}
|
||||
|
||||
// streamAccum accumulates deltas for a single choice in a streaming response.
|
||||
type streamAccum struct {
|
||||
content strings.Builder
|
||||
reasoning strings.Builder
|
||||
refusal strings.Builder
|
||||
role string
|
||||
toolCalls []ToolCall
|
||||
}
|
||||
|
||||
// completionsStream sends a single streaming request, accumulates the
|
||||
// response, and returns all assistant messages from the response choices.
|
||||
// Sends each message to the events channel.
|
||||
func (c *Client) completionsStream(ctx context.Context, messages []Message, model string, events chan<- StreamEvent) ([]Message, error) {
|
||||
body, err := json.Marshal(ChatRequest{
|
||||
Model: model,
|
||||
Messages: messages,
|
||||
Stream: true,
|
||||
Tools: c.tools,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf(
|
||||
"completions stream: marshal request body: %w", err,
|
||||
)
|
||||
}
|
||||
req, err := http.NewRequestWithContext(
|
||||
ctx, http.MethodPost,
|
||||
c.baseURL+"/chat/completions",
|
||||
bytes.NewReader(body),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf(
|
||||
"completions stream: create request: %w", err,
|
||||
)
|
||||
}
|
||||
if c.key != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+c.key)
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Accept", "text/event-stream")
|
||||
|
||||
res, err := c.httpc.Do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("completions stream: %w", err)
|
||||
}
|
||||
|
||||
if res.StatusCode != http.StatusOK {
|
||||
body, err := io.ReadAll(res.Body)
|
||||
res.Body.Close()
|
||||
if err != nil {
|
||||
c.log.ErrorContext(
|
||||
ctx,
|
||||
"completions stream: failed to read error response body",
|
||||
slog.Any("error", err),
|
||||
)
|
||||
return nil, fmt.Errorf(
|
||||
"completions stream: status code %d",
|
||||
res.StatusCode,
|
||||
)
|
||||
}
|
||||
var apiError struct {
|
||||
Error *struct {
|
||||
Type string `json:"type"`
|
||||
Code string `json:"code"`
|
||||
Message string `json:"message"`
|
||||
} `json:"error"`
|
||||
}
|
||||
if err := json.Unmarshal(
|
||||
body, &apiError,
|
||||
); err == nil && apiError.Error != nil {
|
||||
c.log.ErrorContext(
|
||||
ctx,
|
||||
"completions stream: request error",
|
||||
slog.Int("status", res.StatusCode),
|
||||
slog.String("code", apiError.Error.Code),
|
||||
slog.String(
|
||||
"message", apiError.Error.Message,
|
||||
),
|
||||
slog.String("type", apiError.Error.Type),
|
||||
)
|
||||
return nil, fmt.Errorf(
|
||||
"completions stream: %s: %s",
|
||||
apiError.Error.Code,
|
||||
apiError.Error.Message,
|
||||
)
|
||||
}
|
||||
c.log.ErrorContext(
|
||||
ctx,
|
||||
"completions stream: API error",
|
||||
slog.Int("status", res.StatusCode),
|
||||
slog.String("body", string(body)),
|
||||
)
|
||||
return nil, fmt.Errorf(
|
||||
"completions stream: %d: %s",
|
||||
res.StatusCode, string(body),
|
||||
)
|
||||
}
|
||||
|
||||
// Accumulate the streamed response per choice index.
|
||||
accums := make(map[int]*streamAccum)
|
||||
|
||||
scanner := bufio.NewScanner(res.Body)
|
||||
for scanner.Scan() {
|
||||
line := scanner.Text()
|
||||
// Skip empty lines and comments.
|
||||
if line == "" || line == ":" {
|
||||
continue
|
||||
}
|
||||
// Parse data lines.
|
||||
if !strings.HasPrefix(line, "data: ") {
|
||||
continue
|
||||
}
|
||||
data := strings.TrimPrefix(line, "data: ")
|
||||
// [DONE] signals end of stream.
|
||||
if data == "[DONE]" {
|
||||
break
|
||||
}
|
||||
var chunk StreamChunk
|
||||
if err := json.Unmarshal([]byte(data), &chunk); err != nil {
|
||||
continue
|
||||
}
|
||||
for _, ch := range chunk.Choices {
|
||||
acc, ok := accums[ch.Index]
|
||||
if !ok {
|
||||
acc = &streamAccum{}
|
||||
accums[ch.Index] = acc
|
||||
}
|
||||
delta := ch.Delta
|
||||
if delta.Role != "" {
|
||||
acc.role = delta.Role
|
||||
}
|
||||
if delta.Content != "" {
|
||||
acc.content.WriteString(delta.Content)
|
||||
events <- StreamEvent{
|
||||
Type: streamEventDelta,
|
||||
Delta: delta.Content,
|
||||
}
|
||||
}
|
||||
if delta.Refusal != "" {
|
||||
acc.refusal.WriteString(delta.Refusal)
|
||||
}
|
||||
if delta.ReasoningContent != "" {
|
||||
acc.reasoning.WriteString(
|
||||
delta.ReasoningContent,
|
||||
)
|
||||
}
|
||||
// Accumulate tool call deltas by index.
|
||||
for _, tc := range delta.ToolCalls {
|
||||
idx := 0
|
||||
if tc.Index != nil {
|
||||
idx = *tc.Index
|
||||
}
|
||||
// Grow the slice as needed.
|
||||
for len(acc.toolCalls) <= idx {
|
||||
acc.toolCalls = append(
|
||||
acc.toolCalls,
|
||||
ToolCall{Type: "function"},
|
||||
)
|
||||
}
|
||||
if tc.ID != "" {
|
||||
acc.toolCalls[idx].ID = tc.ID
|
||||
}
|
||||
if tc.Function.Name != "" {
|
||||
acc.toolCalls[idx].Function.Name +=
|
||||
tc.Function.Name
|
||||
}
|
||||
if tc.Function.Arguments != "" {
|
||||
acc.toolCalls[idx].Function.Arguments +=
|
||||
tc.Function.Arguments
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
res.Body.Close()
|
||||
|
||||
if err := scanner.Err(); err != nil {
|
||||
return nil, fmt.Errorf(
|
||||
"completions stream: read error: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
// Build messages in index order.
|
||||
indices := make([]int, 0, len(accums))
|
||||
for idx := range accums {
|
||||
indices = append(indices, idx)
|
||||
}
|
||||
slices.Sort(indices)
|
||||
|
||||
msgs := make([]Message, 0, len(accums))
|
||||
for _, idx := range indices {
|
||||
acc := accums[idx]
|
||||
|
||||
// Content and refusal are mutually exclusive in the API.
|
||||
var contentParts []ContentPart
|
||||
if refusalStr := acc.refusal.String(); refusalStr != "" {
|
||||
contentParts = []ContentPart{{
|
||||
Type: ContentTypeRefusal,
|
||||
Refusal: refusalStr,
|
||||
}}
|
||||
} else {
|
||||
contentParts = []ContentPart{{
|
||||
Type: ContentTypeText,
|
||||
Text: acc.content.String(),
|
||||
}}
|
||||
}
|
||||
|
||||
message := Message{
|
||||
Role: acc.role,
|
||||
ContentParts: contentParts,
|
||||
Refusal: acc.refusal.String(),
|
||||
ReasoningContent: acc.reasoning.String(),
|
||||
ToolCalls: acc.toolCalls,
|
||||
}
|
||||
events <- StreamEvent{
|
||||
Type: streamEventDone,
|
||||
Message: message,
|
||||
}
|
||||
msgs = append(msgs, message)
|
||||
}
|
||||
|
||||
return msgs, nil
|
||||
}
|
||||
Reference in New Issue
Block a user