From 353ad56dddb08406cf74bb12df66b78a95df588f Mon Sep 17 00:00:00 2001 From: Max Date: Sat, 15 Nov 2025 11:12:30 +0800 Subject: [PATCH] Implement message preprocessing and enhance OpenAI provider functionality - Added message preprocessing in the base provider to filter unsupported content types (vision and audio) based on model capabilities. - Introduced new methods in the OpenAI provider for handling streaming and non-streaming requests with retry logic and tool call validation. - Enhanced request body building to support various options and improved error handling for API interactions. - Implemented validation for tool call arguments against JSON schemas, ensuring compliance with expected formats. --- agent/llm/providers/base/base.go | 124 ++- agent/llm/providers/openai/openai.go | 664 ++++++++++++++- agent/llm/providers/openai/openai_test.go | 953 ++++++++++++++++++++++ agent/llm/providers/openai/types.go | 83 ++ 4 files changed, 1784 insertions(+), 40 deletions(-) create mode 100644 agent/llm/providers/openai/openai_test.go create mode 100644 agent/llm/providers/openai/types.go diff --git a/agent/llm/providers/base/base.go b/agent/llm/providers/base/base.go index ad9250b0..dee1f889 100644 --- a/agent/llm/providers/base/base.go +++ b/agent/llm/providers/base/base.go @@ -1,6 +1,8 @@ package base import ( + "fmt" + "github.com/yaoapp/gou/connector" "github.com/yaoapp/yao/agent/context" ) @@ -22,13 +24,49 @@ func NewProvider(conn connector.Connector, capabilities *context.ModelCapabiliti // PreprocessMessages preprocess messages before sending to LLM // Handles vision messages, audio messages, tool messages, etc. +// Filters out unsupported content types based on model capabilities func (p *Provider) PreprocessMessages(messages []context.Message) ([]context.Message, error) { - // TODO: Implement message preprocessing - // - Remove vision content if not supported - // - Remove audio content if not supported - // - Convert tool messages if needed - // - Validate message format - return messages, nil + processed := make([]context.Message, 0, len(messages)) + + for _, msg := range messages { + processedMsg := msg + + // Handle multimodal content (array of ContentPart) + if contentParts, ok := msg.Content.([]context.ContentPart); ok { + filteredParts := make([]context.ContentPart, 0, len(contentParts)) + + for _, part := range contentParts { + // Filter vision content if not supported + if part.Type == context.ContentImageURL { + if !p.SupportsVision() { + // Skip image content if vision not supported + continue + } + } + + // Filter audio content if not supported + if part.Type == context.ContentInputAudio { + if !p.SupportsAudio() { + // Skip audio content if audio not supported + continue + } + } + + filteredParts = append(filteredParts, part) + } + + // If all parts were filtered out, convert to text message + if len(filteredParts) == 0 { + processedMsg.Content = "[Content not supported by this model]" + } else { + processedMsg.Content = filteredParts + } + } + + processed = append(processed, processedMsg) + } + + return processed, nil } // SupportsVision check if this provider supports vision @@ -46,20 +84,66 @@ func (p *Provider) SupportsTools() bool { return p.Capabilities != nil && p.Capabilities.ToolCalls != nil && *p.Capabilities.ToolCalls } -// BuildRequestBody build the request body for the LLM API -func (p *Provider) BuildRequestBody(messages []context.Message, options *context.CompletionOptions) (map[string]interface{}, error) { - // TODO: Implement request body building - // - Convert messages to API format - // - Apply options (temperature, max_tokens, etc.) - // - Add model-specific parameters - return nil, nil +// SupportsStreaming check if this provider supports streaming +func (p *Provider) SupportsStreaming() bool { + return p.Capabilities != nil && p.Capabilities.Streaming != nil && *p.Capabilities.Streaming } -// ParseResponse parse the response from LLM API -func (p *Provider) ParseResponse(data []byte, isStreaming bool) (*context.CompletionResponse, error) { - // TODO: Implement response parsing - // - Parse JSON response - // - Extract content, tool calls, reasoning, etc. - // - Handle streaming chunks - return nil, nil +// SupportsJSON check if this provider supports JSON mode +func (p *Provider) SupportsJSON() bool { + return p.Capabilities != nil && p.Capabilities.JSON != nil && *p.Capabilities.JSON +} + +// SupportsReasoning check if this provider supports reasoning mode +func (p *Provider) SupportsReasoning() bool { + return p.Capabilities != nil && p.Capabilities.Reasoning != nil && *p.Capabilities.Reasoning +} + +// GetConnectorSetting gets a setting value from the connector +func (p *Provider) GetConnectorSetting(key string) (interface{}, error) { + if p.Connector == nil { + return nil, fmt.Errorf("connector is nil") + } + + settings := p.Connector.Setting() + if settings == nil { + return nil, fmt.Errorf("connector settings are nil") + } + + value, exists := settings[key] + if !exists { + return nil, fmt.Errorf("setting '%s' not found", key) + } + + return value, nil +} + +// GetConnectorStringSetting gets a string setting value from the connector +func (p *Provider) GetConnectorStringSetting(key string) (string, error) { + value, err := p.GetConnectorSetting(key) + if err != nil { + return "", err + } + + strValue, ok := value.(string) + if !ok { + return "", fmt.Errorf("setting '%s' is not a string", key) + } + + return strValue, nil +} + +// GetModel gets the model name from connector settings +func (p *Provider) GetModel() (string, error) { + return p.GetConnectorStringSetting("model") +} + +// GetAPIKey gets the API key from connector settings +func (p *Provider) GetAPIKey() (string, error) { + return p.GetConnectorStringSetting("key") +} + +// GetHost gets the host URL from connector settings +func (p *Provider) GetHost() (string, error) { + return p.GetConnectorStringSetting("host") } diff --git a/agent/llm/providers/openai/openai.go b/agent/llm/providers/openai/openai.go index ed4d4a4e..8c265086 100644 --- a/agent/llm/providers/openai/openai.go +++ b/agent/llm/providers/openai/openai.go @@ -1,9 +1,18 @@ package openai import ( + gocontext "context" + "fmt" + "strings" + "time" + + jsoniter "github.com/json-iterator/go" "github.com/yaoapp/gou/connector" + "github.com/yaoapp/gou/http" + "github.com/yaoapp/kun/log" "github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/llm/providers/base" + "github.com/yaoapp/yao/utils/jsonschema" ) // Provider OpenAI-compatible provider @@ -21,30 +30,645 @@ func New(conn connector.Connector, capabilities *context.ModelCapabilities) *Pro // Stream stream completion from OpenAI API func (p *Provider) Stream(ctx *context.Context, messages []context.Message, options *context.CompletionOptions, handler context.StreamFunc) (*context.CompletionResponse, error) { - // TODO: Implement OpenAI streaming - // - Preprocess messages (vision, audio, tools) - // - Remove vision content if not supported - // - Remove audio content if not supported - // - Convert to text where needed - // - Build request body - // - Make streaming HTTP request - // - Parse SSE chunks - // - Call handler for each chunk - // - Aggregate final response - return nil, nil + maxRetries := 3 + maxValidationRetries := 3 + var lastErr error + + // Make a copy of messages to avoid modifying the original + currentMessages := make([]context.Message, len(messages)) + copy(currentMessages, messages) + + // Outer loop: handle network/API errors with exponential backoff + for attempt := 0; attempt < maxRetries; attempt++ { + if attempt > 0 { + // Exponential backoff: 1s, 2s, 4s + backoff := time.Duration(1< 0 { + choice := chunk.Choices[0] + delta := choice.Delta + + // Update accumulator metadata + if accumulator.id == "" { + accumulator.id = chunk.ID + accumulator.model = chunk.Model + accumulator.created = chunk.Created + } + + // Handle role + if delta.Role != "" { + accumulator.role = delta.Role + } + + // Handle content + if delta.Content != "" { + accumulator.content += delta.Content + if handler != nil { + handler(context.ChunkText, []byte(delta.Content)) + } + } + + // Handle refusal + if delta.Refusal != "" { + accumulator.refusal += delta.Refusal + if handler != nil { + handler(context.ChunkRefusal, []byte(delta.Refusal)) + } + } + + // Handle tool calls + if len(delta.ToolCalls) > 0 { + for _, tc := range delta.ToolCalls { + if _, exists := accumulator.toolCalls[tc.Index]; !exists { + accumulator.toolCalls[tc.Index] = &accumulatedToolCall{} + } + accTC := accumulator.toolCalls[tc.Index] + + if tc.ID != "" { + accTC.id = tc.ID + } + if tc.Type != "" { + accTC.typ = tc.Type + } + if tc.Function.Name != "" { + accTC.functionName = tc.Function.Name + } + if tc.Function.Arguments != "" { + accTC.functionArgs += tc.Function.Arguments + } + } + + // Notify handler of tool call progress + if handler != nil { + toolCallData, _ := jsoniter.Marshal(delta.ToolCalls) + handler(context.ChunkToolCall, toolCallData) + } + } + + // Handle finish reason + if choice.FinishReason != nil && *choice.FinishReason != "" { + accumulator.finishReason = *choice.FinishReason + } + + // Handle usage (in choices, for older API versions) + if chunk.Usage != nil { + accumulator.usage = &context.UsageInfo{ + PromptTokens: chunk.Usage.PromptTokens, + CompletionTokens: chunk.Usage.CompletionTokens, + TotalTokens: chunk.Usage.TotalTokens, + } + } + } + + // Check for usage at the top level (newer API versions with stream_options) + if chunk.Usage != nil && accumulator.usage == nil { + accumulator.usage = &context.UsageInfo{ + PromptTokens: chunk.Usage.PromptTokens, + CompletionTokens: chunk.Usage.CompletionTokens, + TotalTokens: chunk.Usage.TotalTokens, + } + } + + return http.HandlerReturnOk + } + + // Make streaming request + goCtx := ctx.Context + if goCtx == nil { + goCtx = gocontext.Background() + } + + err = req.Stream(goCtx, "POST", requestBody, streamHandler) + if err != nil { + // Notify handler of error if provided + if handler != nil { + errData := []byte(err.Error()) + handler(context.ChunkError, errData) + } + return nil, fmt.Errorf("streaming request failed: %w", err) + } + + // Check if we received any data + if accumulator.id == "" { + log.Warn("OpenAI stream completed but no data was received (accumulator.id is empty)") + err := fmt.Errorf("no data received from OpenAI API") + // Notify handler of error if provided + if handler != nil { + errData := []byte(err.Error()) + handler(context.ChunkError, errData) + } + return nil, err + } + + // Build final response + response := &context.CompletionResponse{ + ID: accumulator.id, + Object: "chat.completion", + Created: accumulator.created, + Model: accumulator.model, + Role: accumulator.role, + Content: accumulator.content, + Refusal: accumulator.refusal, + FinishReason: accumulator.finishReason, + Usage: accumulator.usage, + } + + // Convert accumulated tool calls to ToolCall slice + if len(accumulator.toolCalls) > 0 { + toolCalls := make([]context.ToolCall, 0, len(accumulator.toolCalls)) + for i := 0; i < len(accumulator.toolCalls); i++ { + if tc, exists := accumulator.toolCalls[i]; exists { + toolCalls = append(toolCalls, context.ToolCall{ + ID: tc.id, + Type: context.ToolCallType(tc.typ), + Function: context.Function{ + Name: tc.functionName, + Arguments: tc.functionArgs, + }, + }) + } + } + response.ToolCalls = toolCalls + + // Validate tool call results if schema is provided + if err := p.validateToolCallResults(options, toolCalls); err != nil { + // Tool call validation failed, need to retry with error feedback + return nil, fmt.Errorf("tool call validation failed: %w", err) + } + } + + return response, nil } // Post post completion request to OpenAI API func (p *Provider) Post(ctx *context.Context, messages []context.Message, options *context.CompletionOptions) (*context.CompletionResponse, error) { - // TODO: Implement OpenAI non-streaming completion - // - Preprocess messages - // - Build request body - // - Make HTTP POST request - // - Parse response - return nil, nil + maxRetries := 3 + maxValidationRetries := 3 + var lastErr error + + // Make a copy of messages to avoid modifying the original + currentMessages := make([]context.Message, len(messages)) + copy(currentMessages, messages) + + // Outer loop: handle network/API errors with exponential backoff + for attempt := 0; attempt < maxRetries; attempt++ { + if attempt > 0 { + // Exponential backoff + backoff := time.Duration(1< 0 { + if err := p.validateToolCallResults(options, response.ToolCalls); err != nil { + return nil, fmt.Errorf("tool call validation failed: %w", err) + } + } + + return response, nil +} + +// buildRequestBody builds the request body for OpenAI API +func (p *Provider) buildRequestBody(messages []context.Message, options *context.CompletionOptions, streaming bool) (map[string]interface{}, error) { + if options == nil { + return nil, fmt.Errorf("options are required") + } + + // Get model from connector settings + setting := p.Connector.Setting() + model, ok := setting["model"].(string) + if !ok || model == "" { + return nil, fmt.Errorf("model is not set in connector") + } + + // Convert messages to API format + apiMessages := make([]map[string]interface{}, 0, len(messages)) + for _, msg := range messages { + apiMsg := map[string]interface{}{ + "role": string(msg.Role), + } + + if msg.Content != nil { + apiMsg["content"] = msg.Content + } + + if msg.Name != nil { + apiMsg["name"] = *msg.Name + } + + if msg.ToolCallID != nil { + apiMsg["tool_call_id"] = *msg.ToolCallID + } + + if len(msg.ToolCalls) > 0 { + apiMsg["tool_calls"] = msg.ToolCalls + } + + if msg.Refusal != nil { + apiMsg["refusal"] = *msg.Refusal + } + + apiMessages = append(apiMessages, apiMsg) + } + + // Build request body + body := map[string]interface{}{ + "model": model, + "messages": apiMessages, + "stream": streaming, + } + + // Add optional parameters + if options.Temperature != nil { + body["temperature"] = *options.Temperature + } + + if options.MaxCompletionTokens != nil { + body["max_completion_tokens"] = *options.MaxCompletionTokens + } else if options.MaxTokens != nil { + body["max_tokens"] = *options.MaxTokens + } + + if options.TopP != nil { + body["top_p"] = *options.TopP + } + + if options.N != nil { + body["n"] = *options.N + } + + if options.Stop != nil { + body["stop"] = options.Stop + } + + if options.PresencePenalty != nil { + body["presence_penalty"] = *options.PresencePenalty + } + + if options.FrequencyPenalty != nil { + body["frequency_penalty"] = *options.FrequencyPenalty + } + + if len(options.LogitBias) > 0 { + body["logit_bias"] = options.LogitBias + } + + if options.User != "" { + body["user"] = options.User + } + + if options.ResponseFormat != nil { + body["response_format"] = options.ResponseFormat + } + + if options.Seed != nil { + body["seed"] = *options.Seed + } + + if len(options.Tools) > 0 { + body["tools"] = options.Tools + } + + if options.ToolChoice != nil { + body["tool_choice"] = options.ToolChoice + } + + // For streaming, include usage info by default + if streaming { + if options.StreamOptions != nil { + body["stream_options"] = options.StreamOptions + } else { + // Default: include usage info in streaming response + body["stream_options"] = map[string]interface{}{ + "include_usage": true, + } + } + } + + if options.Audio != nil { + body["audio"] = options.Audio + } + + return body, nil +} + +// validateToolCallResults validates tool call arguments against JSON schema +func (p *Provider) validateToolCallResults(options *context.CompletionOptions, toolCalls []context.ToolCall) error { + if options == nil || options.Tools == nil || len(options.Tools) == 0 { + return nil + } + + // Build tool schema map for quick lookup + toolSchemas := make(map[string]interface{}) + for _, tool := range options.Tools { + if function, ok := tool["function"].(map[string]interface{}); ok { + if name, ok := function["name"].(string); ok { + if parameters, ok := function["parameters"]; ok { + toolSchemas[name] = parameters + } + } + } + } + + // Validate each tool call + for _, tc := range toolCalls { + schema, hasSchema := toolSchemas[tc.Function.Name] + if !hasSchema { + continue // No schema to validate against + } + + // Parse arguments JSON + var args interface{} + if err := jsoniter.UnmarshalFromString(tc.Function.Arguments, &args); err != nil { + return fmt.Errorf("tool call %s has invalid JSON arguments: %w", tc.Function.Name, err) + } + + // Validate against schema + if err := jsonschema.ValidateData(schema, args); err != nil { + return fmt.Errorf("tool call %s arguments validation failed: %w", tc.Function.Name, err) + } + } + + return nil +} + +// isToolCallValidationError checks if an error is a tool call validation error +func isToolCallValidationError(err error) bool { + if err == nil { + return false + } + errStr := err.Error() + return strings.Contains(errStr, "tool call validation failed") || + strings.Contains(errStr, "arguments validation failed") +} + +// isRetryableError checks if an error is retryable +func isRetryableError(err error) bool { + if err == nil { + return false + } + + errStr := err.Error() + + // Retryable: network errors, timeouts, rate limits, server errors + retryablePatterns := []string{ + "timeout", + "connection refused", + "connection reset", + "EOF", + "HTTP 429", // Rate limit + "HTTP 500", // Internal server error + "HTTP 502", // Bad gateway + "HTTP 503", // Service unavailable + "HTTP 504", // Gateway timeout + } + + for _, pattern := range retryablePatterns { + if strings.Contains(strings.ToLower(errStr), strings.ToLower(pattern)) { + return true + } + } + + return false } diff --git a/agent/llm/providers/openai/openai_test.go b/agent/llm/providers/openai/openai_test.go new file mode 100644 index 00000000..55b1c290 --- /dev/null +++ b/agent/llm/providers/openai/openai_test.go @@ -0,0 +1,953 @@ +package openai_test + +import ( + stdContext "context" + "encoding/json" + "strings" + "testing" + + "github.com/yaoapp/gou/connector" + "github.com/yaoapp/gou/plan" + "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/llm" + "github.com/yaoapp/yao/config" + "github.com/yaoapp/yao/openapi/oauth/types" + "github.com/yaoapp/yao/test" +) + +// TestOpenAIStreamBasic tests basic streaming completion with short output +func TestOpenAIStreamBasic(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + // Create connector from real configuration + conn, err := connector.Select("openai.gpt-4o") + if err != nil { + t.Fatalf("Failed to select connector: %v", err) + } + + // Create LLM instance with capabilities + trueVal := true + options := &context.CompletionOptions{ + Capabilities: &context.ModelCapabilities{ + Streaming: &trueVal, + ToolCalls: &trueVal, + }, + } + + llmInstance, err := llm.New(conn, options) + if err != nil { + t.Fatalf("Failed to create LLM instance: %v", err) + } + + // Prepare messages with concise prompt + messages := []context.Message{ + { + Role: context.RoleUser, + Content: "Say 'Hello' in one word.", + }, + } + + // Set short max tokens to ensure quick response + maxTokens := 5 + options.MaxTokens = &maxTokens + + // Create context + ctx := newTestContext("test-stream-basic", "openai.gpt-4o") + + // Track streaming chunks + var chunks []string + handler := func(chunkType context.StreamChunkType, data []byte) int { + chunks = append(chunks, string(data)) + t.Logf("Stream chunk [%s]: %s", chunkType, string(data)) + return 0 // Continue + } + + // Call Stream + response, err := llmInstance.Stream(ctx, messages, options, handler) + if err != nil { + t.Fatalf("Stream failed: %v", err) + } + + // Validate response + if response == nil { + t.Fatal("Response is nil") + } + + if response.ID == "" { + t.Error("Response ID is empty") + } + if response.Model == "" { + t.Error("Response Model is empty") + } + if response.Content == "" { + t.Error("Response content is empty") + } + if response.FinishReason == "" { + t.Error("FinishReason is empty") + } + if response.Usage == nil { + t.Error("Response Usage is nil") + } else { + if response.Usage.TotalTokens == 0 { + t.Error("Response Usage.TotalTokens is 0") + } + t.Logf("Usage: prompt=%d, completion=%d, total=%d", + response.Usage.PromptTokens, response.Usage.CompletionTokens, response.Usage.TotalTokens) + } + if len(chunks) == 0 { + t.Error("No streaming chunks received") + } + + t.Logf("Final response: %+v", response) + t.Logf("Total chunks received: %d", len(chunks)) +} + +// TestOpenAIPostBasic tests basic non-streaming completion +func TestOpenAIPostBasic(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + // Create connector + conn, err := connector.Select("openai.gpt-4o") + if err != nil { + t.Fatalf("Failed to select connector: %v", err) + } + + // Create LLM instance + trueVal := true + options := &context.CompletionOptions{ + Capabilities: &context.ModelCapabilities{ + ToolCalls: &trueVal, + }, + } + + llmInstance, err := llm.New(conn, options) + if err != nil { + t.Fatalf("Failed to create LLM instance: %v", err) + } + + // Prepare messages with concise prompt + messages := []context.Message{ + { + Role: context.RoleUser, + Content: "Reply with only the word 'OK'.", + }, + } + + // Set short max tokens + maxTokens := 5 + options.MaxTokens = &maxTokens + + // Create context + ctx := newTestContext("test-stream-basic", "openai.gpt-4o") + + // Call Post + response, err := llmInstance.Post(ctx, messages, options) + if err != nil { + t.Fatalf("Post failed: %v", err) + } + + // Validate response + if response == nil { + t.Fatal("Response is nil") + } + + if response.ID == "" { + t.Error("Response ID is empty") + } + if response.Model == "" { + t.Error("Response Model is empty") + } + if response.Content == "" { + t.Error("Response content is empty") + } + if response.FinishReason == "" { + t.Error("FinishReason is empty") + } + if response.Usage == nil { + t.Error("Response Usage is nil") + } else { + if response.Usage.TotalTokens == 0 { + t.Error("Response Usage.TotalTokens is 0") + } + t.Logf("Usage: prompt=%d, completion=%d, total=%d", + response.Usage.PromptTokens, response.Usage.CompletionTokens, response.Usage.TotalTokens) + } + + t.Logf("Response: %+v", response) +} + +// TestOpenAIStreamWithToolCalls tests streaming with tool calls and JSON schema validation +func TestOpenAIStreamWithToolCalls(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + // Create connector + conn, err := connector.Select("openai.gpt-4o") + if err != nil { + t.Fatalf("Failed to select connector: %v", err) + } + + // Create LLM instance with tool call capabilities + trueVal := true + options := &context.CompletionOptions{ + Capabilities: &context.ModelCapabilities{ + Streaming: &trueVal, + ToolCalls: &trueVal, + }, + } + + // Define a simple weather tool with JSON schema + weatherTool := map[string]interface{}{ + "type": "function", + "function": map[string]interface{}{ + "name": "get_weather", + "description": "Get the current weather for a location", + "parameters": map[string]interface{}{ + "type": "object", + "properties": map[string]interface{}{ + "location": map[string]interface{}{ + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + "unit": map[string]interface{}{ + "type": "string", + "enum": []string{"celsius", "fahrenheit"}, + }, + }, + "required": []string{"location"}, + }, + }, + } + + options.Tools = []map[string]interface{}{weatherTool} + options.ToolChoice = "auto" + + llmInstance, err := llm.New(conn, options) + if err != nil { + t.Fatalf("Failed to create LLM instance: %v", err) + } + + // Prepare messages that should trigger tool call + messages := []context.Message{ + { + Role: context.RoleUser, + Content: "What's the weather in Tokyo? Use celsius.", + }, + } + + // Create context + ctx := newTestContext("test-stream-basic", "openai.gpt-4o") + + // Track streaming chunks + var toolCallChunks int + handler := func(chunkType context.StreamChunkType, data []byte) int { + if chunkType == context.ChunkToolCall { + toolCallChunks++ + } + t.Logf("Stream chunk [%s]: %s", chunkType, string(data)) + return 0 // Continue + } + + // Call Stream + response, err := llmInstance.Stream(ctx, messages, options, handler) + if err != nil { + t.Fatalf("Stream with tool calls failed: %v", err) + } + + // Validate response + if response == nil { + t.Fatal("Response is nil") + } + + // Should have tool calls + if len(response.ToolCalls) == 0 { + t.Error("Expected tool calls but got none") + } else { + t.Logf("Received %d tool call(s)", len(response.ToolCalls)) + for i, tc := range response.ToolCalls { + t.Logf("Tool call %d: %s(%s)", i, tc.Function.Name, tc.Function.Arguments) + + // Validate tool call has required fields + if tc.ID == "" { + t.Errorf("Tool call %d missing ID", i) + } + if tc.Function.Name == "" { + t.Errorf("Tool call %d missing function name", i) + } + if tc.Function.Arguments == "" { + t.Errorf("Tool call %d missing arguments", i) + } + } + } + + if response.FinishReason != context.FinishReasonToolCalls { + t.Logf("Warning: Expected finish_reason='tool_calls', got '%s'", response.FinishReason) + } + + if toolCallChunks == 0 { + t.Error("No tool call chunks received during streaming") + } + + t.Logf("Final response: %+v", response) +} + +// TestOpenAIPostWithToolCalls tests non-streaming with tool calls +func TestOpenAIPostWithToolCalls(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + // Create connector + conn, err := connector.Select("openai.gpt-4o") + if err != nil { + t.Fatalf("Failed to select connector: %v", err) + } + + // Create LLM instance with tool call capabilities + trueVal := true + options := &context.CompletionOptions{ + Capabilities: &context.ModelCapabilities{ + ToolCalls: &trueVal, + }, + } + + // Define a calculation tool + calcTool := map[string]interface{}{ + "type": "function", + "function": map[string]interface{}{ + "name": "calculate", + "description": "Perform a mathematical calculation", + "parameters": map[string]interface{}{ + "type": "object", + "properties": map[string]interface{}{ + "expression": map[string]interface{}{ + "type": "string", + "description": "The mathematical expression to evaluate", + }, + }, + "required": []string{"expression"}, + }, + }, + } + + options.Tools = []map[string]interface{}{calcTool} + options.ToolChoice = "auto" + + llmInstance, err := llm.New(conn, options) + if err != nil { + t.Fatalf("Failed to create LLM instance: %v", err) + } + + // Prepare messages + messages := []context.Message{ + { + Role: context.RoleUser, + Content: "Calculate 15 * 8", + }, + } + + // Create context + ctx := newTestContext("test-stream-basic", "openai.gpt-4o") + + // Call Post + response, err := llmInstance.Post(ctx, messages, options) + if err != nil { + t.Fatalf("Post with tool calls failed: %v", err) + } + + // Validate response + if response == nil { + t.Fatal("Response is nil") + } + + // Validate response metadata + if response.ID == "" { + t.Error("Response ID is empty") + } + if response.Model == "" { + t.Error("Response Model is empty") + } + if response.FinishReason != "tool_calls" { + t.Errorf("FinishReason is %s, expected tool_calls", response.FinishReason) + } + if response.Usage == nil { + t.Error("Response Usage is nil") + } else { + if response.Usage.TotalTokens == 0 { + t.Error("Response Usage.TotalTokens is 0") + } + t.Logf("Usage: prompt=%d, completion=%d, total=%d", + response.Usage.PromptTokens, response.Usage.CompletionTokens, response.Usage.TotalTokens) + } + + // Should have tool calls + if len(response.ToolCalls) == 0 { + t.Error("Expected tool calls but got none") + } else { + tc := response.ToolCalls[0] + + // Validate tool call structure + if tc.ID == "" { + t.Error("Tool call ID is empty") + } + if tc.Type != context.ToolTypeFunction { + t.Errorf("Tool call Type is %s, expected %s", tc.Type, context.ToolTypeFunction) + } + if tc.Function.Name != "calculate" { + t.Errorf("Tool call function name is %s, expected calculate", tc.Function.Name) + } + if tc.Function.Arguments == "" { + t.Error("Tool call arguments are empty") + } + + t.Logf("Received %d tool call(s)", len(response.ToolCalls)) + for i, tc := range response.ToolCalls { + t.Logf("Tool call %d: %s(%s)", i, tc.Function.Name, tc.Function.Arguments) + } + } + + t.Logf("Response: %+v", response) +} + +// TestOpenAIStreamWithInvalidToolCall tests that invalid tool calls trigger validation error +func TestOpenAIStreamWithInvalidToolCall(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + // Create connector + conn, err := connector.Select("openai.gpt-4o") + if err != nil { + t.Fatalf("Failed to select connector: %v", err) + } + + // Create LLM instance + trueVal := true + options := &context.CompletionOptions{ + Capabilities: &context.ModelCapabilities{ + Streaming: &trueVal, + ToolCalls: &trueVal, + }, + } + + // Define a strict tool that requires specific format + strictTool := map[string]interface{}{ + "type": "function", + "function": map[string]interface{}{ + "name": "send_email", + "description": "Send an email", + "parameters": map[string]interface{}{ + "type": "object", + "properties": map[string]interface{}{ + "to": map[string]interface{}{ + "type": "string", + "pattern": "^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$", + }, + "subject": map[string]interface{}{ + "type": "string", + "minLength": 1, + }, + "body": map[string]interface{}{ + "type": "string", + "minLength": 1, + }, + }, + "required": []string{"to", "subject", "body"}, + }, + }, + } + + options.Tools = []map[string]interface{}{strictTool} + options.ToolChoice = map[string]interface{}{ + "type": "function", + "function": map[string]interface{}{ + "name": "send_email", + }, + } + + llmInstance, err := llm.New(conn, options) + if err != nil { + t.Fatalf("Failed to create LLM instance: %v", err) + } + + // Prepare messages with incomplete information (should cause validation error) + messages := []context.Message{ + { + Role: context.RoleUser, + Content: "Send email to invalid-email without subject", + }, + } + + // Create context + ctx := newTestContext("test-stream-basic", "openai.gpt-4o") + + handler := func(chunkType context.StreamChunkType, data []byte) int { + return 0 // Continue + } + + // Call Stream - should succeed but may trigger validation if tool call is malformed + response, err := llmInstance.Stream(ctx, messages, options, handler) + + // The API might return a valid tool call despite the bad prompt, + // so we just log the result + if err != nil { + t.Logf("Stream failed as expected with validation error: %v", err) + } else { + t.Logf("Stream succeeded, response: %+v", response) + if len(response.ToolCalls) > 0 { + t.Logf("Tool calls: %v", response.ToolCalls) + } + } +} + +// TestOpenAIStreamRetry tests the retry mechanism with invalid API key +func TestOpenAIStreamRetry(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + // Create connector with invalid API key to trigger 401 error (non-retryable) + connDSL := `{ + "type": "openai", + "options": { + "model": "gpt-4o", + "key": "sk-invalid-key-should-fail-auth", + "host": "https://api.openai.com" + } + }` + + conn, err := connector.New("openai", "test-retry", []byte(connDSL)) + if err != nil { + t.Fatalf("Failed to create test connector: %v", err) + } + + // Create LLM instance + trueVal := true + options := &context.CompletionOptions{ + Capabilities: &context.ModelCapabilities{ + Streaming: &trueVal, + ToolCalls: &trueVal, // Need this to select OpenAI provider + }, + } + + llmInstance, err := llm.New(conn, options) + if err != nil { + t.Fatalf("Failed to create LLM instance: %v", err) + } + + messages := []context.Message{ + { + Role: context.RoleUser, + Content: "Test", + }, + } + + ctx := newTestContext("test-retry", "test-retry") + + // This should fail quickly without retry (401 is non-retryable) + _, err = llmInstance.Stream(ctx, messages, options, nil) + if err == nil { + t.Fatal("Expected error due to invalid API key, but got success") + } + + // Verify it's an error related to invalid API key + // Could be: 401, unauthorized, authentication error, or no data (empty response) + errMsg := err.Error() + hasExpectedError := strings.Contains(strings.ToLower(errMsg), "401") || + strings.Contains(strings.ToLower(errMsg), "unauthorized") || + strings.Contains(strings.ToLower(errMsg), "authentication") || + strings.Contains(strings.ToLower(errMsg), "incorrect api key") || + strings.Contains(strings.ToLower(errMsg), "no data received") + + if !hasExpectedError { + t.Errorf("Expected authentication or empty response error, got: %v", err) + } + + // Should mention non-retryable (these errors should not trigger retry) + if !strings.Contains(strings.ToLower(errMsg), "non-retryable") { + t.Errorf("Error should indicate non-retryable: %v", err) + } + + t.Logf("Failed as expected with error: %v", err) +} + +// TestOpenAIStreamChunkTypes tests that stream handler receives correct chunk types +func TestOpenAIStreamChunkTypes(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + conn, err := connector.Select("openai.gpt-4o") + if err != nil { + t.Fatalf("Failed to select connector: %v", err) + } + + trueVal := true + options := &context.CompletionOptions{ + Capabilities: &context.ModelCapabilities{ + Streaming: &trueVal, + ToolCalls: &trueVal, + }, + } + + llmInstance, err := llm.New(conn, options) + if err != nil { + t.Fatalf("Failed to create LLM instance: %v", err) + } + + messages := []context.Message{ + { + Role: context.RoleUser, + Content: "Say 'test' in one word.", + }, + } + + ctx := newTestContext("test-chunk-types", "openai.gpt-4o") + + // Track chunk types + chunkTypes := make(map[context.StreamChunkType]int) + handler := func(chunkType context.StreamChunkType, data []byte) int { + chunkTypes[chunkType]++ + t.Logf("Received chunk type: %s, data length: %d", chunkType, len(data)) + return 1 // Continue + } + + response, err := llmInstance.Stream(ctx, messages, options, handler) + if err != nil { + t.Fatalf("Stream failed: %v", err) + } + + if response == nil { + t.Fatal("Response is nil") + } + + // Validate chunk types received + if chunkTypes[context.ChunkText] == 0 { + t.Error("Expected to receive ChunkText, but got 0") + } + + t.Logf("Chunk types received: %+v", chunkTypes) +} + +// TestOpenAIStreamErrorCallback tests that errors are sent to stream handler +func TestOpenAIStreamErrorCallback(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + // Create connector with invalid API key to trigger error + connDSL := `{ + "type": "openai", + "options": { + "model": "gpt-4o", + "key": "sk-invalid-for-error-test", + "host": "https://api.openai.com" + } + }` + + conn, err := connector.New("openai", "test-error-callback", []byte(connDSL)) + if err != nil { + t.Fatalf("Failed to create test connector: %v", err) + } + + trueVal := true + options := &context.CompletionOptions{ + Capabilities: &context.ModelCapabilities{ + Streaming: &trueVal, + ToolCalls: &trueVal, + }, + } + + llmInstance, err := llm.New(conn, options) + if err != nil { + t.Fatalf("Failed to create LLM instance: %v", err) + } + + messages := []context.Message{ + { + Role: context.RoleUser, + Content: "Test", + }, + } + + ctx := newTestContext("test-error-callback", "test-error-callback") + + // Track if error chunk was received + receivedError := false + var errorMessage string + handler := func(chunkType context.StreamChunkType, data []byte) int { + if chunkType == context.ChunkError { + receivedError = true + errorMessage = string(data) + t.Logf("Received error chunk: %s", errorMessage) + } + return 1 // Continue + } + + // This should fail and send error to handler + _, err = llmInstance.Stream(ctx, messages, options, handler) + if err == nil { + t.Fatal("Expected error due to invalid API key") + } + + // Verify error was sent to handler + if !receivedError { + t.Error("Expected to receive ChunkError in handler, but didn't") + } + + if errorMessage == "" { + t.Error("Error message in chunk is empty") + } + + t.Logf("Error callback test passed. Error: %v", err) +} + +// TestOpenAIToolCallValidationRetry tests automatic tool call validation retry with LLM feedback +func TestOpenAIToolCallValidationRetry(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + conn, err := connector.Select("openai.gpt-4o") + if err != nil { + t.Fatalf("Failed to select connector: %v", err) + } + + trueVal := true + options := &context.CompletionOptions{ + Capabilities: &context.ModelCapabilities{ + Streaming: &trueVal, + ToolCalls: &trueVal, + }, + Tools: []map[string]interface{}{ + { + "type": "function", + "function": map[string]interface{}{ + "name": "test_strict_validation", + "description": "A function with very strict validation rules", + "parameters": map[string]interface{}{ + "type": "object", + "properties": map[string]interface{}{ + "status": map[string]interface{}{ + "type": "string", + "description": "Must be exactly 'active' or 'inactive'", + "enum": []string{"active", "inactive"}, + }, + "priority": map[string]interface{}{ + "type": "integer", + "description": "Must be between 1 and 5", + "minimum": 1, + "maximum": 5, + }, + }, + "required": []string{"status", "priority"}, + }, + }, + }, + }, + } + + llmInstance, err := llm.New(conn, options) + if err != nil { + t.Fatalf("Failed to create LLM instance: %v", err) + } + + ctx := newTestContext("test-tool-validation-retry", "openai.gpt-4o") + + // Try to make LLM call with intentionally unclear requirements + // This may or may not trigger validation, depending on LLM behavior + messages := []context.Message{ + { + Role: context.RoleUser, + Content: "Call test_strict_validation function with status='pending' and priority=10", + }, + } + + // The Provider will automatically: + // 1. Call LLM + // 2. If validation fails, add error feedback to conversation + // 3. Retry up to 3 times with feedback + // 4. Return success or validation error after max retries + response, err := llmInstance.Stream(ctx, messages, options, nil) + + if err != nil { + // Check if it's a validation error after retries + if strings.Contains(err.Error(), "tool call validation failed after") && + strings.Contains(err.Error(), "retries") { + t.Logf("✓ Automatic validation retry exhausted: %v", err) + } else if strings.Contains(err.Error(), "validation") { + t.Logf("✓ Validation failed: %v", err) + } else { + t.Logf("Request failed (non-validation): %v", err) + } + } else if response != nil { + if len(response.ToolCalls) > 0 { + t.Logf("✓ Tool call succeeded (possibly after auto-retry): %+v", response.ToolCalls[0]) + + // Verify the tool call arguments are valid + tc := response.ToolCalls[0] + var args map[string]interface{} + if err := json.Unmarshal([]byte(tc.Function.Arguments), &args); err == nil { + if status, ok := args["status"].(string); ok { + if status != "active" && status != "inactive" { + t.Errorf("Status should be 'active' or 'inactive', got: %s", status) + } + } + if priority, ok := args["priority"].(float64); ok { + if priority < 1 || priority > 5 { + t.Errorf("Priority should be between 1-5, got: %v", priority) + } + } + } + } else { + t.Log("✓ Response returned but no tool calls") + } + } + + t.Log("Automatic tool call validation retry test completed") +} + +// TestOpenAIProxySupport tests that HTTP proxy configuration is respected +func TestOpenAIProxySupport(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + // This test verifies proxy support exists in the connector configuration + // Actual proxy testing requires a real proxy server setup + + conn, err := connector.Select("openai.gpt-4o") + if err != nil { + t.Fatalf("Failed to select connector: %v", err) + } + + settings := conn.Setting() + t.Logf("Connector settings: %+v", settings) + + // Verify host field exists in settings (host is the API endpoint) + if host, hasHost := settings["host"]; hasHost { + t.Logf("API host configured: %v", host) + } else { + t.Log("Host field not in settings (will use default)") + } + + // The actual HTTP proxy functionality is implemented via environment variables + // (HTTP_PROXY, HTTPS_PROXY, NO_PROXY) and handled by http.GetTransport + t.Log("HTTP proxy support is implemented via http.GetTransport using environment variables") +} + +// TestOpenAIStreamWithTemperature tests different temperature settings +func TestOpenAIStreamWithTemperature(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + + conn, err := connector.Select("openai.gpt-4o") + if err != nil { + t.Fatalf("Failed to select connector: %v", err) + } + + trueVal := true + temperature := 0.7 // Moderate temperature + + options := &context.CompletionOptions{ + Capabilities: &context.ModelCapabilities{ + Streaming: &trueVal, + ToolCalls: &trueVal, // Need this to select OpenAI provider + }, + Temperature: &temperature, + } + + llmInstance, err := llm.New(conn, options) + if err != nil { + t.Fatalf("Failed to create LLM instance: %v", err) + } + + messages := []context.Message{ + { + Role: context.RoleUser, + Content: "Say 'yes' in one word.", + }, + } + + ctx := newTestContext("test-temperature", "openai.gpt-4o") + + // Use callback to collect chunks + chunkCount := 0 + var callback context.StreamFunc = func(chunkType context.StreamChunkType, data []byte) int { + chunkCount++ + return 1 // Continue + } + + response, err := llmInstance.Stream(ctx, messages, options, callback) + if err != nil { + t.Fatalf("Stream failed: %v", err) + } + + if response == nil { + t.Fatal("Response is nil") + } + + // Validate response data + if response.ID == "" { + t.Error("Response ID is empty") + } + if response.Model == "" { + t.Error("Response Model is empty") + } + if response.Content == "" { + t.Error("Response Content is empty") + } + if response.FinishReason == "" { + t.Error("Response FinishReason is empty") + } + if response.Usage == nil { + t.Error("Response Usage is nil") + } else { + if response.Usage.TotalTokens == 0 { + t.Error("Response Usage.TotalTokens is 0") + } + t.Logf("Usage: prompt=%d, completion=%d, total=%d", + response.Usage.PromptTokens, response.Usage.CompletionTokens, response.Usage.TotalTokens) + } + if chunkCount == 0 { + t.Error("No chunks received") + } + + t.Logf("Response with temperature=0.7: %+v", response) + t.Logf("Total chunks received: %d", chunkCount) +} + +// ============================================================================ +// Helper Functions +// ============================================================================ + +// newTestContext creates a real Context for testing OpenAI provider +func newTestContext(chatID, connectorID string) *context.Context { + return &context.Context{ + Context: stdContext.Background(), + Space: plan.NewMemorySharedSpace(), + ChatID: chatID, + AssistantID: "test-assistant", + Connector: connectorID, + Locale: "en-us", + Theme: "light", + Client: context.Client{ + Type: "web", + UserAgent: "OpenAIProviderTest/1.0", + IP: "127.0.0.1", + }, + Referer: context.RefererAPI, + Accept: context.AcceptStandard, + Route: "/api/test", + Metadata: make(map[string]interface{}), + Authorized: &types.AuthorizedInfo{ + Subject: "test-user", + ClientID: "test-client", + UserID: "test-user-123", + TeamID: "test-team-456", + TenantID: "test-tenant-789", + SessionID: "test-session-id", + Constraints: types.DataConstraints{ + TeamOnly: true, + Extra: map[string]interface{}{ + "test": "openai-provider", + }, + }, + }, + } +} diff --git a/agent/llm/providers/openai/types.go b/agent/llm/providers/openai/types.go new file mode 100644 index 00000000..c6a2db00 --- /dev/null +++ b/agent/llm/providers/openai/types.go @@ -0,0 +1,83 @@ +package openai + +import "github.com/yaoapp/yao/agent/context" + +// StreamChunk represents a chunk from OpenAI's streaming response +type StreamChunk struct { + ID string `json:"id"` + Object string `json:"object"` + Created int64 `json:"created"` + Model string `json:"model"` + Choices []Delta `json:"choices"` + Usage *struct { + PromptTokens int `json:"prompt_tokens"` + CompletionTokens int `json:"completion_tokens"` + TotalTokens int `json:"total_tokens"` + } `json:"usage,omitempty"` +} + +// Delta represents the delta in a streaming chunk +type Delta struct { + Index int `json:"index"` + Delta DeltaContent `json:"delta"` + FinishReason *string `json:"finish_reason"` +} + +// DeltaContent represents the content in a delta +type DeltaContent struct { + Role string `json:"role,omitempty"` + Content string `json:"content,omitempty"` + ToolCalls []ToolCallDelta `json:"tool_calls,omitempty"` + Refusal string `json:"refusal,omitempty"` +} + +// ToolCallDelta represents a tool call delta in streaming +type ToolCallDelta struct { + Index int `json:"index"` + ID string `json:"id,omitempty"` + Type string `json:"type,omitempty"` + Function FunctionCallDelta `json:"function,omitempty"` +} + +// FunctionCallDelta represents a function call delta +type FunctionCallDelta struct { + Name string `json:"name,omitempty"` + Arguments string `json:"arguments,omitempty"` +} + +// CompletionResponseFull represents the full non-streaming response +type CompletionResponseFull struct { + ID string `json:"id"` + Object string `json:"object"` + Created int64 `json:"created"` + Model string `json:"model"` + Choices []struct { + Index int `json:"index"` + Message context.Message `json:"message"` + FinishReason string `json:"finish_reason"` + } `json:"choices"` + Usage *context.UsageInfo `json:"usage,omitempty"` + SystemFingerprint string `json:"system_fingerprint,omitempty"` +} + +// streamAccumulator accumulates streaming response data +type streamAccumulator struct { + id string + model string + created int64 + role string + content string + refusal string + toolCalls map[int]*accumulatedToolCall + finishReason string + usage *context.UsageInfo +} + +// accumulatedToolCall accumulates a single tool call +type accumulatedToolCall struct { + id string + typ string + functionName string + functionArgs string +} +