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.
2025-11-15 11:12:30 +08:00 · 2025-11-15 11:12:30 +08:00 · 353ad56ddd
commit 353ad56ddd
parent cef75ff71b
4 changed files with 1784 additions and 40 deletions
--- 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")
 }
--- 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<<uint(attempt-1)) * time.Second
+			log.Warn("OpenAI stream request failed, retrying in %v (attempt %d/%d): %v", backoff, attempt+1, maxRetries, lastErr)
+			time.Sleep(backoff)
+		}
+
+		response, err := p.streamWithRetry(ctx, currentMessages, options, handler)
+		if err == nil {
+			return response, nil
+		}
+		lastErr = err
+
+		// Check if error is tool call validation failure
+		if isToolCallValidationError(err) {
+			// Handle tool call validation retry with feedback to LLM
+			validationRetryMessages := currentMessages
+			for validationAttempt := 0; validationAttempt < maxValidationRetries; validationAttempt++ {
+				log.Warn("Tool call validation failed (attempt %d/%d): %v", validationAttempt+1, maxValidationRetries, err)
+
+				// Add error feedback to conversation history
+				validationRetryMessages = append(validationRetryMessages, context.Message{
+					Role:    context.RoleSystem,
+					Content: fmt.Sprintf("Tool call validation error: %v. Please correct the tool call arguments to match the required schema.", err),
+				})
+
+				// Retry with feedback
+				response, err = p.streamWithRetry(ctx, validationRetryMessages, options, handler)
+				if err == nil {
+					return response, nil
+				}
+
+				// Check if still validation error
+				if !isToolCallValidationError(err) {
+					// Different error type, break out of validation retry loop
+					lastErr = err
+					break
+				}
+				lastErr = err
+			}
+
+			// If we exhausted validation retries, return the error
+			if isToolCallValidationError(lastErr) {
+				return nil, fmt.Errorf("tool call validation failed after %d retries: %w", maxValidationRetries, lastErr)
+			}
+		}
+
+		// Check if error is retryable (network errors, rate limits, etc.)
+		if !isRetryableError(err) {
+			return nil, fmt.Errorf("non-retryable error: %w", err)
+		}
+	}
+
+	return nil, fmt.Errorf("failed after %d retries: %w", maxRetries, lastErr)
+}
+
+// streamWithRetry performs a single streaming request attempt
+func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Message, options *context.CompletionOptions, handler context.StreamFunc) (*context.CompletionResponse, error) {
+	// Build request body
+	requestBody, err := p.buildRequestBody(messages, options, true)
+	if err != nil {
+		return nil, fmt.Errorf("failed to build request body: %w", err)
+	}
+
+	// Get connector settings
+	setting := p.Connector.Setting()
+	host, ok := setting["host"].(string)
+	if !ok || host == "" {
+		return nil, fmt.Errorf("no host found in connector settings")
+	}
+
+	key, ok := setting["key"].(string)
+	if !ok || key == "" {
+		return nil, fmt.Errorf("API key is not set")
+	}
+
+	// Build URL
+	endpoint := "/chat/completions"
+	if host == "https://api.openai.com" && !strings.HasPrefix(endpoint, "/v1") {
+		endpoint = "/v1" + endpoint
+	}
+	host = strings.TrimSuffix(host, "/")
+	url := host + endpoint
+
+	// Create HTTP request with proxy support
+	req := http.New(url).
+		SetHeader("Content-Type", "application/json").
+		SetHeader("Authorization", fmt.Sprintf("Bearer %s", key)).
+		SetHeader("Accept", "text/event-stream")
+
+	// Accumulate response data
+	accumulator := &streamAccumulator{
+		toolCalls: make(map[int]*accumulatedToolCall),
+	}
+
+	// Stream handler
+	streamHandler := func(data []byte) int {
+		if len(data) == 0 {
+			return http.HandlerReturnOk
+		}
+
+		// Parse SSE data
+		dataStr := string(data)
+		if !strings.HasPrefix(dataStr, "data: ") {
+			return http.HandlerReturnOk
+		}
+
+		dataStr = strings.TrimPrefix(dataStr, "data: ")
+		dataStr = strings.TrimSpace(dataStr)
+
+		// Check for [DONE] marker
+		if dataStr == "[DONE]" {
+			return http.HandlerReturnOk
+		}
+
+		// Parse JSON chunk
+		var chunk StreamChunk
+		if err := jsoniter.UnmarshalFromString(dataStr, &chunk); err != nil {
+			log.Warn("Failed to parse stream chunk: %v", err)
+			return http.HandlerReturnOk
+		}
+
+		// Process chunk
+		if len(chunk.Choices) > 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<<uint(attempt-1)) * time.Second
+			log.Warn("OpenAI post request failed, retrying in %v (attempt %d/%d): %v", backoff, attempt+1, maxRetries, lastErr)
+			time.Sleep(backoff)
+		}
+
+		response, err := p.postWithRetry(ctx, currentMessages, options)
+		if err == nil {
+			return response, nil
+		}
+		lastErr = err
+
+		// Check if error is tool call validation failure
+		if isToolCallValidationError(err) {
+			// Handle tool call validation retry with feedback to LLM
+			validationRetryMessages := currentMessages
+			for validationAttempt := 0; validationAttempt < maxValidationRetries; validationAttempt++ {
+				log.Warn("Tool call validation failed (attempt %d/%d): %v", validationAttempt+1, maxValidationRetries, err)
+
+				// Add error feedback to conversation history
+				validationRetryMessages = append(validationRetryMessages, context.Message{
+					Role:    context.RoleSystem,
+					Content: fmt.Sprintf("Tool call validation error: %v. Please correct the tool call arguments to match the required schema.", err),
+				})
+
+				// Retry with feedback
+				response, err = p.postWithRetry(ctx, validationRetryMessages, options)
+				if err == nil {
+					return response, nil
+				}
+
+				// Check if still validation error
+				if !isToolCallValidationError(err) {
+					// Different error type, break out of validation retry loop
+					lastErr = err
+					break
+				}
+				lastErr = err
+			}
+
+			// If we exhausted validation retries, return the error
+			if isToolCallValidationError(lastErr) {
+				return nil, fmt.Errorf("tool call validation failed after %d retries: %w", maxValidationRetries, lastErr)
+			}
+		}
+
+		// Check if error is retryable (network errors, rate limits, etc.)
+		if !isRetryableError(err) {
+			return nil, fmt.Errorf("non-retryable error: %w", err)
+		}
+	}
+
+	return nil, fmt.Errorf("failed after %d retries: %w", maxRetries, lastErr)
 }

-// SupportsAudio check if this provider supports audio
-func (p *Provider) SupportsAudio() bool {
-	return p.Capabilities != nil && p.Capabilities.Audio != nil && *p.Capabilities.Audio
+// postWithRetry performs a single POST request attempt
+func (p *Provider) postWithRetry(ctx *context.Context, messages []context.Message, options *context.CompletionOptions) (*context.CompletionResponse, error) {
+	// Build request body
+	requestBody, err := p.buildRequestBody(messages, options, false)
+	if err != nil {
+		return nil, fmt.Errorf("failed to build request body: %w", err)
+	}
+
+	// Get connector settings
+	setting := p.Connector.Setting()
+	host, ok := setting["host"].(string)
+	if !ok || host == "" {
+		return nil, fmt.Errorf("no host found in connector settings")
+	}
+
+	key, ok := setting["key"].(string)
+	if !ok || key == "" {
+		return nil, fmt.Errorf("API key is not set")
+	}
+
+	// Build URL
+	endpoint := "/chat/completions"
+	if host == "https://api.openai.com" && !strings.HasPrefix(endpoint, "/v1") {
+		endpoint = "/v1" + endpoint
+	}
+	host = strings.TrimSuffix(host, "/")
+	url := host + endpoint
+
+	// Create HTTP request with proxy support
+	req := http.New(url).
+		SetHeader("Content-Type", "application/json").
+		SetHeader("Authorization", fmt.Sprintf("Bearer %s", key))
+
+	// Make request
+	resp := req.Post(requestBody)
+	if resp.Code != 200 {
+		return nil, fmt.Errorf("HTTP %d: %s", resp.Code, resp.Message)
+	}
+
+	// Parse response
+	var fullResp CompletionResponseFull
+	respData, err := jsoniter.Marshal(resp.Data)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal response: %w", err)
+	}
+
+	if err := jsoniter.Unmarshal(respData, &fullResp); err != nil {
+		return nil, fmt.Errorf("failed to parse response: %w", err)
+	}
+
+	if len(fullResp.Choices) == 0 {
+		return nil, fmt.Errorf("no choices in response")
+	}
+
+	choice := fullResp.Choices[0]
+	response := &context.CompletionResponse{
+		ID:                fullResp.ID,
+		Object:            fullResp.Object,
+		Created:           fullResp.Created,
+		Model:             fullResp.Model,
+		Role:              string(choice.Message.Role),
+		Content:           choice.Message.Content,
+		ToolCalls:         choice.Message.ToolCalls,
+		FinishReason:      choice.FinishReason,
+		Usage:             fullResp.Usage,
+		SystemFingerprint: fullResp.SystemFingerprint,
+	}
+
+	if choice.Message.Refusal != nil {
+		response.Refusal = *choice.Message.Refusal
+	}
+
+	// Validate tool call results if present
+	if len(response.ToolCalls) > 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
 }
--- a/agent/llm/providers/openai/openai_test.go
+++ 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",
+				},
+			},
+		},
+	}
+}
--- a/agent/llm/providers/openai/types.go
+++ 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
+}
+