From 21c0ec50345b1bb1ed52be9c54b8b7f380d27908 Mon Sep 17 00:00:00 2001 From: Max Date: Sat, 22 Nov 2025 11:14:49 +0800 Subject: [PATCH] Enhance Assistant's Stream method and context handling for improved traceability and output management - Refactored the Stream method to retrieve connector capabilities early, allowing output adapters to utilize them effectively. - Introduced a new Info method in the Assistant to provide structured assistant information, enhancing context accessibility. - Updated StreamStartData to include additional fields such as ChatID and Assistant info for better event tracking. - Improved error handling in the Stream method to ensure robust management of connector retrieval failures. - Enhanced internationalization support for stream event messages, providing localized output for different clients. --- agent/assistant/agent.go | 46 +++++++++-- agent/context/context.go | 15 +++- agent/context/types.go | 12 +++ agent/context/types_llm.go | 9 ++- agent/i18n/builtin.go | 12 +++ agent/llm/handlers/stream.go | 11 ++- agent/output/BUILTIN_TYPES.md | 51 +++++++----- agent/output/README.md | 26 +++--- agent/output/adapters/cui/writer.go | 2 +- agent/output/adapters/openai/README.md | 81 +++++++++++++++++-- agent/output/adapters/openai/adapter.go | 24 +++++- agent/output/adapters/openai/converter.go | 99 ++++++++++++++++++++--- agent/output/adapters/openai/types.go | 13 +++ agent/output/adapters/openai/writer.go | 34 +++++++- agent/output/builtin.go | 2 +- agent/output/message/interfaces.go | 2 +- agent/output/message/types.go | 12 ++- agent/output/output.go | 2 +- openapi/chat/completions.go | 1 + 19 files changed, 381 insertions(+), 73 deletions(-) diff --git a/agent/assistant/agent.go b/agent/assistant/agent.go index b9250167..f85fc1f6 100644 --- a/agent/assistant/agent.go +++ b/agent/assistant/agent.go @@ -40,10 +40,27 @@ func (ast *Assistant) Stream(ctx *context.Context, inputMessages []context.Messa _ = traceID // traceID is available for trace logging + // Get connector and capabilities early (before sending stream_start) + // so that output adapters can use them when converting stream_start event + if ast.Prompts != nil || ast.MCP != nil { + _, capabilities, err := ast.GetConnector(ctx) + if err != nil { + streamHandler := ast.getStreamHandler(ctx, handler...) + ast.sendStreamEndOnError(ctx, streamHandler, streamStartTime, err) + return nil, err + } + + // Set capabilities in context for output adapters to use + if capabilities != nil { + ctx.Capabilities = capabilities + } + } + // Determine stream handler streamHandler := ast.getStreamHandler(ctx, handler...) // Send ChunkStreamStart only for root stack (agent-level stream start) + // Now ctx.Capabilities is set, so output adapters can use it ast.sendAgentStreamStart(ctx, streamHandler, streamStartTime) // Trace Add @@ -111,7 +128,7 @@ func (ast *Assistant) Stream(ctx *context.Context, inputMessages []context.Messa return nil, err } - // Get connector object and capabilities + // Get connector object (capabilities were already set above, before stream_start) conn, capabilities, err := ast.GetConnector(ctx) if err != nil { if agentNode != nil { @@ -338,6 +355,21 @@ func (ast *Assistant) BuildRequest(ctx *context.Context, messages []context.Mess return finalMessages, options, nil } +// Info get the assistant information +func (ast *Assistant) Info(locale ...string) *context.AssistantInfo { + lc := "en" + if len(locale) > 0 { + lc = locale[0] + } + return &context.AssistantInfo{ + ID: ast.ID, + Type: ast.Type, + Name: i18n.Tr(ast.ID, lc, ast.Name), + Avatar: ast.Avatar, + Description: i18n.Tr(ast.ID, lc, ast.Description), + } +} + // buildMessages builds the final message list with proper priority // Priority: createResponse.Messages > input messages // If createResponse is nil or has no messages, use input messages @@ -642,11 +674,13 @@ func (ast *Assistant) sendAgentStreamStart(ctx *context.Context, handler context return } - requestID := fmt.Sprintf("agent_req_%d", startTime.UnixNano()) - startData := &context.StreamStartData{ - RequestID: requestID, + // Build the start data + startData := context.StreamStartData{ + RequestID: ctx.RequestID(), Timestamp: startTime.UnixMilli(), - Model: ast.ID, // Use assistant ID as the "model" for agent-level stream + Assistant: ast.Info(ctx.Locale), + ChatID: ctx.ChatID, + TraceID: ctx.TraceID(), } if startJSON, err := jsoniter.Marshal(startData); err == nil { @@ -667,7 +701,7 @@ func (ast *Assistant) sendAgentStreamEnd(ctx *context.Context, handler context.S } endData := &context.StreamEndData{ - RequestID: fmt.Sprintf("agent_req_%d", startTime.UnixNano()), + RequestID: ctx.RequestID(), Timestamp: time.Now().UnixMilli(), DurationMs: time.Since(startTime).Milliseconds(), Status: status, diff --git a/agent/context/context.go b/agent/context/context.go index 65e9ef21..5dd98af5 100644 --- a/agent/context/context.go +++ b/agent/context/context.go @@ -358,5 +358,18 @@ func SendInterrupt(contextID string, signal *InterruptSignal) error { // generateContextID generates a unique context ID func generateContextID() string { - return fmt.Sprintf("ctx_%d", time.Now().UnixNano()) + return fmt.Sprintf("ctx-%d", time.Now().UnixNano()) +} + +// RequestID returns the request ID for the context +func (ctx *Context) RequestID() string { + return fmt.Sprintf("%s", ctx.ID) +} + +// TraceID returns the trace ID for the context +func (ctx *Context) TraceID() string { + if ctx.Stack != nil { + return ctx.Stack.TraceID + } + return "" } diff --git a/agent/context/types.go b/agent/context/types.go index ec9fe3ad..9ff39fd1 100644 --- a/agent/context/types.go +++ b/agent/context/types.go @@ -177,6 +177,15 @@ type InterruptController struct { contextID string `json:"-"` // Context ID to retrieve the parent context } +// AssistantInfo represents the assistant information structure +type AssistantInfo struct { + ID string `json:"assistant_id"` // Assistant ID + Type string `json:"type,omitempty"` // Assistant Type, default is assistant + Name string `json:"name,omitempty"` // Assistant Name + Avatar string `json:"avatar,omitempty"` // Assistant Avatar + Description string `json:"description,omitempty"` // Assistant Description +} + // Context the context type Context struct { @@ -190,6 +199,9 @@ type Context struct { Writer Writer `json:"-"` // Writer, it will be used to write response data to the client trace traceTypes.Manager `json:"-"` // Trace manager, lazy initialized on first access + // Model capabilities (set by assistant, used by output adapters) + Capabilities *ModelCapabilities `json:"-"` // Model capabilities for the current connector + // Interrupt control (all interrupt-related logic is encapsulated in InterruptController) Interrupt *InterruptController `json:"-"` // Interrupt controller for handling user interrupts during streaming diff --git a/agent/context/types_llm.go b/agent/context/types_llm.go index 211d4267..2b6dd853 100644 --- a/agent/context/types_llm.go +++ b/agent/context/types_llm.go @@ -221,10 +221,11 @@ type JSONSchema struct { // StreamStartData represents the data for stream_start event // Sent when a streaming request begins type StreamStartData struct { - RequestID string `json:"request_id"` // Unique identifier for this request - Timestamp int64 `json:"timestamp"` // Unix timestamp when stream started - Model string `json:"model,omitempty"` // Model being used (e.g., "gpt-4o") - Capabilities map[string]interface{} `json:"capabilities,omitempty"` // Model capabilities for this request + RequestID string `json:"request_id"` // Unique identifier for this request + Timestamp int64 `json:"timestamp"` // Unix timestamp when stream started + ChatID string `json:"chat_id"` // Chat ID being used (e.g., "chat-123") + TraceID string `json:"trace_id"` // Trace ID being used (e.g., "trace-123") + Assistant *AssistantInfo `json:"assistant,omitempty"` // Assistant information } // StreamEndData represents the data for stream_end event diff --git a/agent/i18n/builtin.go b/agent/i18n/builtin.go index a466e440..63d8e8c1 100644 --- a/agent/i18n/builtin.go +++ b/agent/i18n/builtin.go @@ -55,6 +55,10 @@ func init() { "output.cui.writer.send_error": "Failed to send data to client", "output.cui.writer.marshal_error": "Failed to marshal chunk", + // Output: Stream event messages + "output.stream_start": "Assistant is processing", + "output.view_trace": "View process", + // Common status messages "common.status.processing": "Processing", "common.status.completed": "Completed", @@ -111,6 +115,10 @@ func init() { "output.cui.writer.send_error": "发送数据到客户端失败", "output.cui.writer.marshal_error": "序列化数据块失败", + // Output: Stream event messages + "output.stream_start": "智能体正在处理", + "output.view_trace": "查看处理详情", + // Common status messages "common.status.processing": "处理中", "common.status.completed": "已完成", @@ -167,6 +175,10 @@ func init() { "output.cui.writer.send_error": "发送数据到客户端失败", "output.cui.writer.marshal_error": "序列化数据块失败", + // Output: Stream event messages + "output.stream_start": "智能体正在处理", + "output.view_trace": "查看处理详情", + // Common status messages "common.status.processing": "处理中", "common.status.completed": "已完成", diff --git a/agent/llm/handlers/stream.go b/agent/llm/handlers/stream.go index b41a8429..58867659 100644 --- a/agent/llm/handlers/stream.go +++ b/agent/llm/handlers/stream.go @@ -1,6 +1,8 @@ package handlers import ( + jsoniter "github.com/json-iterator/go" + "github.com/yaoapp/kun/log" "github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/i18n" "github.com/yaoapp/yao/agent/output" @@ -73,9 +75,14 @@ type streamState struct { func (s *streamState) handleStreamStart(data []byte) int { // Send event message to indicate stream has started // This is a lifecycle event, CUI clients can show it, OpenAI clients will ignore it - msg := output.NewEventMessage("stream_start", "Connecting...", nil) + var startData context.StreamStartData + err := jsoniter.Unmarshal(data, &startData) + if err != nil { + log.Error("Failed to unmarshal stream start data: %v", err) + } + msg := output.NewEventMessage("stream_start", "Stream started", startData) output.Send(s.ctx, msg) - return 0 // Continue + return 0 } // handleGroupStart handles group start event diff --git a/agent/output/BUILTIN_TYPES.md b/agent/output/BUILTIN_TYPES.md index 9f5eb2ae..d05131f0 100644 --- a/agent/output/BUILTIN_TYPES.md +++ b/agent/output/BUILTIN_TYPES.md @@ -343,18 +343,25 @@ msg := output.NewEventMessage("stream_start", "Starting stream...", map[string]i **Important Notes:** -- **Silent in OpenAI clients**: Event messages are NOT sent to standard chat clients -- **CUI clients only**: Only CUI clients process event messages -- **Lifecycle tracking**: Used for tracking agent/stream lifecycle, not chat content +- **Converted in OpenAI clients**: Event messages are typically NOT sent to OpenAI clients, **except** `stream_start`: + - `stream_start`: Converted to a clickable trace link in either `reasoning_content` (thinking models) or `content` (regular models) + - Other events: Silent (not sent to OpenAI clients) +- **CUI clients**: All event messages are processed and may show status indicators +- **Lifecycle tracking**: Used for tracking agent/stream lifecycle - **Non-blocking**: Events don't interrupt the main message flow **Example in Hook:** ```go -// Send stream start event -output.Send(ctx, output.NewEventMessage("stream_start", "Initializing...", map[string]interface{}{ - "timestamp": time.Now().Unix(), -})) +// Send stream start event (automatically generated by assistant) +// This is typically handled by the framework, not manually sent +startData := context.StreamStartData{ + RequestID: ctx.RequestID(), + Timestamp: time.Now().UnixMilli(), + TraceID: ctx.TraceID(), + ChatID: ctx.ChatID, +} +output.Send(ctx, output.NewEventMessage("stream_start", "Stream started", startData)) // Do processing processData() @@ -368,7 +375,11 @@ output.Send(ctx, output.NewEventMessage("stream_end", "Stream completed", map[st **Result:** - **CUI client**: Tracks lifecycle, may show status indicators -- **OpenAI client**: Events are silent (not sent to client) +- **OpenAI client (stream_start only)**: + - Reasoning models: Shows as 🔍 with trace link in `reasoning_content` field + - Regular models: Shows as 🚀 with trace link in `content` field + - Example: "🔍 智能体正在处理 - [查看处理详情](baseURL/trace/traceID/view)" +- **OpenAI client (other events)**: Silent (not sent) --- @@ -530,18 +541,18 @@ CUI adapter passes built-in types through without transformation: OpenAI adapter converts built-in types to OpenAI format: -| Type | OpenAI Format | Field | Note | -| ----------- | ------------------------- | ----------------------------- | ----------------------------------------- | -| `text` | `delta.content` | `props.content` | | -| `thinking` | `delta.reasoning_content` | `props.content` | Reasoning content (o1 models) | -| `loading` | `delta.reasoning_content` | `props.message` | Shows as thinking in OpenAI clients | -| `tool_call` | `delta.tool_calls` | `props.{id, name, arguments}` | | -| `error` | `error` | `props.{message, code}` | | -| `image` | `delta.content` | `props.{url, alt}` | Markdown: `![alt](url)` - displays inline | -| `audio` | `delta.content` | `props.url` | Markdown link (can't display inline) | -| `video` | `delta.content` | `props.url` | Markdown link (can't display inline) | -| `action` | (not sent) | - | Silent - system actions only | -| `event` | (not sent) | - | Silent - lifecycle events only | +| Type | OpenAI Format | Field | Note | +| ----------- | ------------------------- | ----------------------------- | -------------------------------------------------------------------- | +| `text` | `delta.content` | `props.content` | | +| `thinking` | `delta.reasoning_content` | `props.content` | Reasoning content (o1 models) | +| `loading` | `delta.reasoning_content` | `props.message` | Shows as thinking in OpenAI clients | +| `tool_call` | `delta.tool_calls` | `props.{id, name, arguments}` | | +| `error` | `error` | `props.{message, code}` | | +| `image` | `delta.content` | `props.{url, alt}` | Markdown: `![alt](url)` - displays inline | +| `audio` | `delta.content` | `props.url` | Markdown link (can't display inline) | +| `video` | `delta.content` | `props.url` | Markdown link (can't display inline) | +| `action` | (not sent) | - | Silent - system actions only | +| `event` | (conditional) | `props.{event, data}` | Most events silent; `stream_start` converted to trace link with i18n | --- diff --git a/agent/output/README.md b/agent/output/README.md index e10ced23..b3dcb631 100644 --- a/agent/output/README.md +++ b/agent/output/README.md @@ -7,7 +7,7 @@ The output module provides a unified API for sending messages to different clien ``` agent/output/ ├── message/ # Core types and interfaces (no dependencies) -│ ├── types.go # Message, MessageGroup, Props structures +│ ├── types.go # Message, Group, Props structures │ └── interfaces.go # Writer, Adapter, Factory interfaces ├── adapters/ # Client-specific adapters │ ├── cui/ # CUI adapter (native DSL) @@ -325,18 +325,18 @@ Adapters handle the transformation automatically based on `ctx.Accept`. 10 standardized message types with defined Props structures: -| Type | Purpose | CUI | OpenAI | -| ----------- | ------------------ | ------- | ------------------------- | -| `text` | Text content | Direct | `delta.content` | -| `thinking` | LLM reasoning | Direct | `delta.reasoning_content` | -| `loading` | Progress indicator | Direct | `delta.reasoning_content` | -| `tool_call` | Function calls | Direct | `delta.tool_calls` | -| `error` | Error messages | Direct | `error` | -| `image` | Images | Render | `![](url)` markdown | -| `audio` | Audio | Player | Link | -| `video` | Video | Player | Link | -| `action` | System commands | Execute | Silent | -| `event` | Lifecycle events | Track | Silent | +| Type | Purpose | CUI | OpenAI | +| ----------- | ------------------ | ------- | -------------------------------------------- | +| `text` | Text content | Direct | `delta.content` | +| `thinking` | LLM reasoning | Direct | `delta.reasoning_content` | +| `loading` | Progress indicator | Direct | `delta.reasoning_content` | +| `tool_call` | Function calls | Direct | `delta.tool_calls` | +| `error` | Error messages | Direct | `error` | +| `image` | Images | Render | `![](url)` markdown | +| `audio` | Audio | Player | Link | +| `video` | Video | Player | Link | +| `action` | System commands | Execute | Silent | +| `event` | Lifecycle events | Track | Conditional (stream_start converted to link) | ## Usage diff --git a/agent/output/adapters/cui/writer.go b/agent/output/adapters/cui/writer.go index 695a817f..a16e7723 100644 --- a/agent/output/adapters/cui/writer.go +++ b/agent/output/adapters/cui/writer.go @@ -50,7 +50,7 @@ func (w *Writer) Write(msg *message.Message) error { } // WriteGroup writes a message group to the output stream -func (w *Writer) WriteGroup(group *message.MessageGroup) error { +func (w *Writer) WriteGroup(group *message.Group) error { // For CUI, we send a group start message, all messages, then a group end message // The group structure itself is also sent for clients that want it diff --git a/agent/output/adapters/openai/README.md b/agent/output/adapters/openai/README.md index c90b0bfa..b8e6a777 100644 --- a/agent/output/adapters/openai/README.md +++ b/agent/output/adapters/openai/README.md @@ -8,14 +8,79 @@ OpenAI adapter converts universal DSL messages to OpenAI-compatible format. These types are defined in `output.types.go` and have standardized Props structures that all adapters must support: -| Message Type | Constant | Props Structure | OpenAI Format | Description | -| ------------ | --------------------- | --------------- | ------------------------- | ------------------------------------- | -| `text` | `output.TypeText` | `TextProps` | `delta.content` | Plain text or Markdown | -| `thinking` | `output.TypeThinking` | `ThinkingProps` | `delta.reasoning_content` | Reasoning process (o1 models) | -| `loading` | `output.TypeLoading` | `LoadingProps` | `delta.reasoning_content` | Loading indicator (shows as thinking) | -| `tool_call` | `output.TypeToolCall` | `ToolCallProps` | `delta.tool_calls` | Tool/function calls | -| `error` | `output.TypeError` | `ErrorProps` | `error` | Error messages | -| `action` | `output.TypeAction` | `ActionProps` | (not sent) | System actions (silent) | +| Message Type | Constant | Props Structure | OpenAI Format | Description | +| ------------ | --------------------- | --------------- | ------------------------- | ----------------------------------------- | +| `text` | `output.TypeText` | `TextProps` | `delta.content` | Plain text or Markdown | +| `thinking` | `output.TypeThinking` | `ThinkingProps` | `delta.reasoning_content` | Reasoning process (o1 models) | +| `loading` | `output.TypeLoading` | `LoadingProps` | `delta.reasoning_content` | Loading indicator (shows as thinking) | +| `tool_call` | `output.TypeToolCall` | `ToolCallProps` | `delta.tool_calls` | Tool/function calls | +| `error` | `output.TypeError` | `ErrorProps` | `error` | Error messages | +| `action` | `output.TypeAction` | `ActionProps` | (not sent) | System actions (silent) | +| `event` | `output.TypeEvent` | `EventProps` | (conditional) | Lifecycle events (stream_start converted) | + +### Event Type (Lifecycle Events) + +The `event` type has special handling in the OpenAI adapter: + +| Event Name | Conversion | Example Output | +| -------------- | ------------------------------------------- | --------------------------------------------------- | +| `stream_start` | Converted to trace link (with i18n support) | 🔍 智能体正在处理 - [查看处理详情](/trace/xxx/view) | +| Other events | Silent (not sent) | - | + +**Conversion Logic for `stream_start`:** + +1. **Extract trace data**: Gets `TraceID` from event data +2. **Check model capabilities**: Determines if model supports reasoning +3. **Format based on capabilities**: + - **Reasoning models** (o1, DeepSeek R1): Uses `reasoning_content` field with 🔍 icon + - **Regular models**: Uses `content` field with 🚀 icon +4. **Apply i18n**: Uses locale from context for localized text +5. **Generate trace link**: Creates clickable link to `/trace/{traceID}/view` for standalone viewing + +**Example Conversion:** + +```go +// Input (event message) +{ + "type": "event", + "props": { + "event": "stream_start", + "message": "Stream started", + "data": { + "trace_id": "20251122779905354593", + "request_id": "ctx-1763779905679380000", + "chat_id": "uP4CWZCMHy84nCw7" + } + } +} + +// Output (reasoning model - Chinese locale) +{ + "choices": [{ + "delta": { + "reasoning_content": "🔍 智能体正在处理 - [查看处理详情](http://localhost:8000/__yao_admin_root/trace/20251122779905354593/view)\n" + } + }] +} + +// Output (regular model - English locale) +{ + "choices": [{ + "delta": { + "content": "🚀 Assistant is processing - [View process](http://localhost:8000/__yao_admin_root/trace/20251122779905354593/view)\n" + } + }] +} +``` + +**Internationalization:** + +The adapter uses `i18n.T()` to provide localized text: + +| Key | English (en-us) | Chinese (zh-cn) | +| --------------------- | ----------------------- | --------------- | +| `output.stream_start` | Assistant is processing | 智能体正在处理 | +| `output.view_trace` | View process | 查看处理详情 | ### Custom Types diff --git a/agent/output/adapters/openai/adapter.go b/agent/output/adapters/openai/adapter.go index de6747d3..08e9f281 100644 --- a/agent/output/adapters/openai/adapter.go +++ b/agent/output/adapters/openai/adapter.go @@ -54,6 +54,20 @@ func WithModel(model string) Option { } } +// WithCapabilities sets the model capabilities +func WithCapabilities(capabilities *ModelCapabilities) Option { + return func(a *Adapter) { + a.config.Capabilities = capabilities + } +} + +// WithLocale sets the locale for internationalization +func WithLocale(locale string) Option { + return func(a *Adapter) { + a.config.Locale = locale + } +} + // WithConverter registers a custom converter for a message type func WithConverter(msgType string, converter ConverterFunc) Option { return func(a *Adapter) { @@ -63,8 +77,16 @@ func WithConverter(msgType string, converter ConverterFunc) Option { // Adapt converts a universal Message to OpenAI-compatible format func (a *Adapter) Adapt(msg *message.Message) ([]interface{}, error) { - // Skip event messages - they are CUI-only lifecycle events + // Handle event messages specially if msg.Type == message.TypeEvent { + // Check if this is a stream_start event + if event, ok := msg.Props["event"].(string); ok && event == message.EventStreamStart { + // Use the stream_start converter + if converter, exists := a.registry.GetConverter(message.EventStreamStart); exists { + return converter(msg, a.config) + } + } + // Other event messages are CUI-only, skip them return []interface{}{}, nil // Return empty array, nothing to send } diff --git a/agent/output/adapters/openai/converter.go b/agent/output/adapters/openai/converter.go index 16d4cd5d..9eb7faea 100644 --- a/agent/output/adapters/openai/converter.go +++ b/agent/output/adapters/openai/converter.go @@ -4,6 +4,8 @@ import ( "fmt" "time" + "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/i18n" "github.com/yaoapp/yao/agent/output/message" ) @@ -16,15 +18,16 @@ type ConverterRegistry struct { func NewConverterRegistry() *ConverterRegistry { return &ConverterRegistry{ converters: map[string]ConverterFunc{ - message.TypeText: convertText, - message.TypeThinking: convertThinking, - message.TypeLoading: convertLoading, - message.TypeToolCall: convertToolCall, - message.TypeError: convertError, - message.TypeImage: convertImage, - message.TypeAudio: convertToLink, - message.TypeVideo: convertToLink, - message.TypeAction: convertAction, + message.TypeText: convertText, + message.TypeThinking: convertThinking, + message.TypeLoading: convertLoading, + message.TypeToolCall: convertToolCall, + message.TypeError: convertError, + message.TypeImage: convertImage, + message.TypeAudio: convertToLink, + message.TypeVideo: convertToLink, + message.TypeAction: convertAction, + message.EventStreamStart: convertStreamStart, // Handle stream_start events }, } } @@ -137,6 +140,84 @@ func convertAction(msg *message.Message, config *AdapterConfig) ([]interface{}, return []interface{}{}, nil } +// convertStreamStart converts stream_start event to OpenAI format +// If model supports reasoning: converts to reasoning_content (thinking) +// Otherwise: converts to regular Markdown text with trace link +func convertStreamStart(msg *message.Message, config *AdapterConfig) ([]interface{}, error) { + // Extract stream_start data from props + data, ok := msg.Props["data"] + if !ok { + // No data, skip this message + return []interface{}{}, nil + } + + // Try to convert to StreamStartData + var startData context.StreamStartData + switch v := data.(type) { + case context.StreamStartData: + startData = v + case map[string]interface{}: + // If it's a map, try to extract traceID + if traceID, ok := v["trace_id"].(string); ok { + startData.TraceID = traceID + } + if requestID, ok := v["request_id"].(string); ok { + startData.RequestID = requestID + } + default: + // Unknown data type, skip + return []interface{}{}, nil + } + + // Check if we have a trace ID to link to + if startData.TraceID == "" { + // No trace ID, skip this message + return []interface{}{}, nil + } + + // Generate trace link + traceLink := generateTraceLink(startData.TraceID, config) + + // Check if model supports reasoning + supportsReasoning := false + if config.Capabilities != nil && config.Capabilities.Reasoning != nil { + supportsReasoning = *config.Capabilities.Reasoning + } + + // Get localized text using i18n + streamStartText := i18n.T(config.Locale, "output.stream_start") + viewTraceText := i18n.T(config.Locale, "output.view_trace") + + // Convert based on reasoning support + if supportsReasoning { + // Convert to thinking format (reasoning_content) + // Reasoning models display this as part of the thinking process + content := fmt.Sprintf("🔍 %s - [%s](%s)\n", streamStartText, viewTraceText, traceLink) + chunk := createOpenAIChunk(msg.ID, config.Model, map[string]interface{}{ + "reasoning_content": content, + }) + return []interface{}{chunk}, nil + } + + // Convert to regular Markdown text + content := fmt.Sprintf("🚀 %s - [%s](%s)\n", streamStartText, viewTraceText, traceLink) + chunk := createOpenAIChunk(msg.ID, config.Model, map[string]interface{}{ + "content": content, + }) + return []interface{}{chunk}, nil +} + +// generateTraceLink generates a trace link URL +// Uses 'view' mode (clean page without sidebar) for better viewing experience in chat +func generateTraceLink(traceID string, config *AdapterConfig) string { + baseURL := config.BaseURL + if baseURL == "" { + // If no base URL, return a relative link + return fmt.Sprintf("/trace/%s/view", traceID) + } + return fmt.Sprintf("%s/trace/%s/view", baseURL, traceID) +} + // convertImage converts image messages to Markdown image format // Uses ![alt](url) which displays inline in Markdown-supporting clients func convertImage(msg *message.Message, config *AdapterConfig) ([]interface{}, error) { diff --git a/agent/output/adapters/openai/types.go b/agent/output/adapters/openai/types.go index 589b9f0e..0a154bd6 100644 --- a/agent/output/adapters/openai/types.go +++ b/agent/output/adapters/openai/types.go @@ -26,6 +26,19 @@ type AdapterConfig struct { // Model name to include in OpenAI responses Model string + + // Capabilities holds the model capabilities + // Used to determine how to convert certain message types (e.g., stream_start) + Capabilities *ModelCapabilities + + // Locale for internationalization (e.g., "en-US", "zh-CN") + Locale string +} + +// ModelCapabilities is a simplified version of context.ModelCapabilities +// We use a local type to avoid circular dependencies +type ModelCapabilities struct { + Reasoning *bool // Supports reasoning/thinking mode (o1, DeepSeek R1) } // DefaultLinkTemplates provides default Markdown templates for non-text message types diff --git a/agent/output/adapters/openai/writer.go b/agent/output/adapters/openai/writer.go index bdbd83bb..92e29a43 100644 --- a/agent/output/adapters/openai/writer.go +++ b/agent/output/adapters/openai/writer.go @@ -17,8 +17,20 @@ type Writer struct { // NewWriter creates a new OpenAI writer func NewWriter(ctx *context.Context) (*Writer, error) { - // Create adapter with default config - adapter := NewAdapter() + // Get model capabilities from context (set by assistant) + var capabilities *ModelCapabilities + if ctx.Capabilities != nil && ctx.Capabilities.Reasoning != nil { + capabilities = &ModelCapabilities{ + Reasoning: ctx.Capabilities.Reasoning, + } + } + + // Create adapter with capabilities, base URL, and locale + adapter := NewAdapter( + WithCapabilities(capabilities), + WithBaseURL(getBaseURL(ctx)), + WithLocale(ctx.Locale), + ) return &Writer{ ctx: ctx, @@ -27,6 +39,22 @@ func NewWriter(ctx *context.Context) (*Writer, error) { }, nil } +// getBaseURL gets the base URL from context or environment +func getBaseURL(ctx *context.Context) string { + // @todo: get from context metadata + return "http://localhost:8000/__yao_admin_root" + + // // Try to get from context metadata + // if ctx.Metadata != nil { + // if baseURL, ok := ctx.Metadata["base_url"].(string); ok && baseURL != "" { + // return baseURL + // } + // } + + // // TODO: Get from environment variable or config + // return "" +} + // Write writes a single message to the output stream func (w *Writer) Write(msg *message.Message) error { // Convert message to OpenAI format using adapter @@ -67,7 +95,7 @@ func (w *Writer) Write(msg *message.Message) error { } // WriteGroup writes a message group to the output stream -func (w *Writer) WriteGroup(group *message.MessageGroup) error { +func (w *Writer) WriteGroup(group *message.Group) error { // For OpenAI, we don't send group markers // Just send each message individually for _, msg := range group.Messages { diff --git a/agent/output/builtin.go b/agent/output/builtin.go index 1db502e8..bcd83f2f 100644 --- a/agent/output/builtin.go +++ b/agent/output/builtin.go @@ -79,7 +79,7 @@ func NewActionMessage(name string, payload map[string]interface{}) *message.Mess } // NewEventMessage creates an event message -func NewEventMessage(event string, msg string, data map[string]interface{}) *message.Message { +func NewEventMessage(event string, msg string, data interface{}) *message.Message { return &message.Message{ Type: message.TypeEvent, Props: map[string]interface{}{ diff --git a/agent/output/message/interfaces.go b/agent/output/message/interfaces.go index 42f713b9..0d06bff2 100644 --- a/agent/output/message/interfaces.go +++ b/agent/output/message/interfaces.go @@ -9,7 +9,7 @@ type Writer interface { Write(msg *Message) error // WriteGroup writes a group of messages - WriteGroup(group *MessageGroup) error + WriteGroup(group *Group) error // Flush flushes any buffered data Flush() error diff --git a/agent/output/message/types.go b/agent/output/message/types.go index 28999910..8d073fc7 100644 --- a/agent/output/message/types.go +++ b/agent/output/message/types.go @@ -35,8 +35,8 @@ type Metadata struct { TraceID string `json:"trace_id,omitempty"` // Trace ID (for debugging) } -// MessageGroup represents a semantically complete group of messages -type MessageGroup struct { +// Group represents a semantically complete group of messages +type Group struct { ID string `json:"id"` // Message group ID Messages []*Message `json:"messages"` // List of messages Metadata *Metadata `json:"metadata,omitempty"` // Metadata @@ -62,6 +62,14 @@ const ( TypeEvent = "event" // Lifecycle event (stream_start, stream_end, etc.) - CUI only, silent in OpenAI clients ) +// Event types for TypeEvent messages +const ( + EventStreamStart = "stream_start" // Stream started event + EventStreamEnd = "stream_end" // Stream ended event + EventGroupStart = "group_start" // Message group started event + EventGroupEnd = "group_end" // Message group ended event +) + // Standard Props structures for built-in types // TextProps defines the standard structure for text messages diff --git a/agent/output/output.go b/agent/output/output.go index 8d410815..013de262 100644 --- a/agent/output/output.go +++ b/agent/output/output.go @@ -26,7 +26,7 @@ func Send(ctx *context.Context, msg *message.Message) error { } // SendGroup sends a message group using the appropriate writer for the context -func SendGroup(ctx *context.Context, group *message.MessageGroup) error { +func SendGroup(ctx *context.Context, group *message.Group) error { writer, err := GetWriter(ctx) if err != nil { return err diff --git a/openapi/chat/completions.go b/openapi/chat/completions.go index 1ee42ded..2bd83735 100644 --- a/openapi/chat/completions.go +++ b/openapi/chat/completions.go @@ -48,6 +48,7 @@ func GinCreateCompletions(c *gin.Context) { fmt.Println("Chat ID: ", ctx.ChatID) fmt.Println("Assistant ID: ", ctx.AssistantID) fmt.Println("Model: ", completionReq.Model) + fmt.Println("Locale: ", ctx.Locale) fmt.Println("Messages count: ", len(completionReq.Messages)) if completionReq.Temperature != nil { fmt.Println("Temperature: ", *completionReq.Temperature)