// Package atlascloud implements the Atlas Cloud model provider. // // Atlas Cloud is an enterprise AI infrastructure platform offering // high-performance LLM, image, and video APIs. It exposes // OpenAI-compatible endpoints for chat completions and image // generation. // // Usage: // // import _ "go-micro.dev/v6/ai/atlascloud" // // m := ai.New("atlascloud", // ai.WithAPIKey("your-api-key"), // ) // // // Image generation // ig := ai.NewImage("atlascloud", // ai.WithAPIKey("your-api-key"), // ) package atlascloud import ( "bufio" "bytes" "context" "encoding/json" "fmt" "io" "net/http" "os" "strings" "time" "go-micro.dev/v6/ai" ) func init() { ai.Register("atlascloud", func(opts ...ai.Option) ai.Model { return NewProvider(opts...) }) ai.RegisterImage("atlascloud", func(opts ...ai.Option) ai.ImageModel { return NewProvider(opts...) }) ai.RegisterVideo("atlascloud", func(opts ...ai.Option) ai.VideoModel { return NewProvider(opts...) }) ai.RegisterStream("atlascloud") } // Provider implements the ai.Model interface for Atlas Cloud. type Provider struct { opts ai.Options } // NewProvider creates a new Atlas Cloud provider. func NewProvider(opts ...ai.Option) *Provider { options := ai.NewOptions(opts...) if options.Model == "" { // Allow the chat model to be selected via the ATLASCLOUD_MODEL env var // (e.g. to run CI conformance against a stronger tool-use model) without // a code change; fall back to a sensible default otherwise. if m := os.Getenv("ATLASCLOUD_MODEL"); m != "" { options.Model = m } else { options.Model = "deepseek-ai/DeepSeek-V3-0324" } } if options.BaseURL == "" { options.BaseURL = "https://api.atlascloud.ai" } return &Provider{opts: options} } func (p *Provider) Init(opts ...ai.Option) error { for _, o := range opts { o(&p.opts) } return nil } func (p *Provider) Options() ai.Options { return p.opts } func (p *Provider) String() string { return "atlascloud" } func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (*ai.Response, error) { var tools []map[string]any for _, t := range req.Tools { tools = append(tools, map[string]any{ "type": "function", "function": map[string]any{ "name": t.Name, "description": t.Description, "parameters": map[string]any{ "type": "object", "properties": t.Properties, }, }, }) } messages := []map[string]any{ {"role": "system", "content": req.SystemPrompt}, } for _, m := range req.Messages { messages = append(messages, map[string]any{"role": m.Role, "content": m.Content}) } if req.Prompt != "" { messages = append(messages, map[string]any{"role": "user", "content": req.Prompt}) } apiReq := map[string]any{ "model": p.opts.Model, "messages": messages, } if p.opts.MaxTokens > 0 { apiReq["max_tokens"] = p.opts.MaxTokens } if len(tools) > 0 { apiReq["tools"] = tools } resp, rawMessage, err := p.callAPI(ctx, apiReq) if err != nil { return nil, err } if len(resp.ToolCalls) == 0 { return resp, nil } if p.opts.ToolHandler != nil { followUpMessages := append(messages, map[string]any{ "role": "assistant", "content": rawMessage["content"], "tool_calls": rawMessage["tool_calls"], }) for _, tc := range resp.ToolCalls { content := p.opts.ToolHandler(ctx, tc).Content followUpMessages = append(followUpMessages, map[string]any{ "role": "tool", "tool_call_id": tc.ID, "content": content, }) } followUpReq := map[string]any{ "model": p.opts.Model, "messages": followUpMessages, } followUpResp, _, err := p.callAPI(ctx, followUpReq) if err == nil && followUpResp.Reply != "" { resp.Answer = followUpResp.Reply } } return resp, nil } // Stream generates a streaming response from Atlas Cloud's OpenAI-compatible // chat completions endpoint, emitting content deltas as they arrive. func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) { messages := []map[string]any{ {"role": "system", "content": req.SystemPrompt}, } for _, m := range req.Messages { messages = append(messages, map[string]any{"role": m.Role, "content": m.Content}) } if req.Prompt != "" { messages = append(messages, map[string]any{"role": "user", "content": req.Prompt}) } apiReq := map[string]any{ "model": p.opts.Model, "messages": messages, "stream": true, "stream_options": map[string]any{"include_usage": true}, } if p.opts.MaxTokens > 0 { apiReq["max_tokens"] = p.opts.MaxTokens } reqBody, err := json.Marshal(apiReq) if err != nil { return nil, fmt.Errorf("failed to marshal stream request: %w", err) } apiURL := strings.TrimRight(p.opts.BaseURL, "/") + "/v1/chat/completions" httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(reqBody)) if err != nil { return nil, fmt.Errorf("failed to create stream request: %w", err) } httpReq.Header.Set("Content-Type", "application/json") httpReq.Header.Set("Accept", "text/event-stream") httpReq.Header.Set("Authorization", "Bearer "+p.opts.APIKey) httpResp, err := http.DefaultClient.Do(httpReq) if err != nil { return nil, fmt.Errorf("stream API request failed: %w", err) } if httpResp.StatusCode != http.StatusOK { defer httpResp.Body.Close() respBody, _ := io.ReadAll(httpResp.Body) return nil, fmt.Errorf("stream API error (%s): %s", httpResp.Status, string(respBody)) } return &atlasStream{body: httpResp.Body, scanner: bufio.NewScanner(httpResp.Body)}, nil } type atlasStream struct { body io.ReadCloser scanner *bufio.Scanner closed bool } func (s *atlasStream) Recv() (*ai.Response, error) { for s.scanner.Scan() { line := strings.TrimSpace(s.scanner.Text()) if line == "" || strings.HasPrefix(line, ":") { continue } if !strings.HasPrefix(line, "data:") { continue } data := strings.TrimSpace(strings.TrimPrefix(line, "data:")) if data == "[DONE]" { return nil, io.EOF } var chunk struct { Choices []struct { Delta struct { Content string `json:"content"` } `json:"delta"` } `json:"choices"` Usage *struct { PromptTokens int `json:"prompt_tokens"` CompletionTokens int `json:"completion_tokens"` TotalTokens int `json:"total_tokens"` } `json:"usage"` } if err := json.Unmarshal([]byte(data), &chunk); err != nil { return nil, fmt.Errorf("failed to parse stream chunk: %w", err) } if len(chunk.Choices) > 0 && chunk.Choices[0].Delta.Content != "" { return &ai.Response{Reply: chunk.Choices[0].Delta.Content}, nil } // Final chunk (after include_usage) carries token usage and no content. if chunk.Usage != nil { return &ai.Response{Usage: ai.Usage{ InputTokens: chunk.Usage.PromptTokens, OutputTokens: chunk.Usage.CompletionTokens, TotalTokens: chunk.Usage.TotalTokens, }}, nil } continue } if err := s.scanner.Err(); err != nil { return nil, err } return nil, io.EOF } func (s *atlasStream) Close() error { if s.closed { return nil } s.closed = true return s.body.Close() } func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) { reqBody, err := json.Marshal(req) if err != nil { return nil, nil, fmt.Errorf("failed to marshal request: %w", err) } apiURL := strings.TrimRight(p.opts.BaseURL, "/") + "/v1/chat/completions" httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(reqBody)) if err != nil { return nil, nil, fmt.Errorf("failed to create request: %w", err) } httpReq.Header.Set("Content-Type", "application/json") httpReq.Header.Set("Authorization", "Bearer "+p.opts.APIKey) httpResp, err := http.DefaultClient.Do(httpReq) if err != nil { return nil, nil, fmt.Errorf("API request failed: %w", err) } defer httpResp.Body.Close() respBody, _ := io.ReadAll(httpResp.Body) if httpResp.StatusCode != http.StatusOK { return nil, nil, fmt.Errorf("API error (%s): %s", httpResp.Status, string(respBody)) } var chatResp struct { Choices []struct { Message struct { Content string `json:"content"` ToolCalls []struct { ID string `json:"id"` Function struct { Name string `json:"name"` Arguments string `json:"arguments"` } `json:"function"` } `json:"tool_calls"` } `json:"message"` } `json:"choices"` } if err := json.Unmarshal(respBody, &chatResp); err != nil { return nil, nil, fmt.Errorf("failed to parse response: %w", err) } if len(chatResp.Choices) == 0 { return nil, nil, fmt.Errorf("no response from API") } choice := chatResp.Choices[0] response := &ai.Response{ Reply: choice.Message.Content, } for _, tc := range choice.Message.ToolCalls { var input map[string]any if err := json.Unmarshal([]byte(tc.Function.Arguments), &input); err != nil { input = map[string]any{} } response.ToolCalls = append(response.ToolCalls, ai.ToolCall{ ID: tc.ID, Name: tc.Function.Name, Input: input, }) } rawMessage := map[string]any{ "content": choice.Message.Content, "tool_calls": choice.Message.ToolCalls, } return response, rawMessage, nil } const defaultImageModel = "openai/gpt-image-2/text-to-image" // GenerateImage creates an image using Atlas Cloud's async image API. // It submits the job and polls until completion or context cancellation. func (p *Provider) GenerateImage(ctx context.Context, req *ai.ImageRequest, opts ...ai.GenerateOption) (*ai.ImageResponse, error) { model := req.Model if model == "" { model = defaultImageModel } quality := req.Quality if quality == "" { quality = "medium" } outputFmt := req.OutputFormat if outputFmt == "" { outputFmt = "png" } size := req.Size if size == "" { size = "1024x1024" } apiReq := map[string]any{ "model": model, "prompt": req.Prompt, "quality": quality, "output_format": outputFmt, "size": size, "enable_sync_mode": false, "enable_base64_output": false, "moderation": "low", } reqBody, err := json.Marshal(apiReq) if err != nil { return nil, fmt.Errorf("failed to marshal request: %w", err) } apiURL := strings.TrimRight(p.opts.BaseURL, "/") + "/api/v1/model/generateImage" httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(reqBody)) if err != nil { return nil, fmt.Errorf("failed to create request: %w", err) } httpReq.Header.Set("Content-Type", "application/json") httpReq.Header.Set("Authorization", "Bearer "+p.opts.APIKey) httpResp, err := http.DefaultClient.Do(httpReq) if err != nil { return nil, fmt.Errorf("API request failed: %w", err) } defer httpResp.Body.Close() respBody, _ := io.ReadAll(httpResp.Body) if httpResp.StatusCode != http.StatusOK { return nil, fmt.Errorf("API error (%s): %s", httpResp.Status, string(respBody)) } var submitResp struct { Code int `json:"code"` Msg string `json:"message"` Data struct { ID string `json:"id"` Status string `json:"status"` } `json:"data"` } if err := json.Unmarshal(respBody, &submitResp); err != nil { return nil, fmt.Errorf("failed to parse submit response: %w", err) } if submitResp.Code != 200 { return nil, fmt.Errorf("API error: %s", submitResp.Msg) } predictionID := submitResp.Data.ID pollURL := strings.TrimRight(p.opts.BaseURL, "/") + "/api/v1/model/prediction/" + predictionID ticker := time.NewTicker(2 * time.Second) defer ticker.Stop() for { select { case <-ctx.Done(): return nil, ctx.Err() case <-ticker.C: result, err := p.pollPrediction(ctx, pollURL) if err != nil { return nil, err } if result != nil { return result, nil } } } } func (p *Provider) pollPrediction(ctx context.Context, url string) (*ai.ImageResponse, error) { httpReq, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) if err != nil { return nil, err } httpReq.Header.Set("Authorization", "Bearer "+p.opts.APIKey) httpResp, err := http.DefaultClient.Do(httpReq) if err != nil { return nil, fmt.Errorf("poll request failed: %w", err) } defer httpResp.Body.Close() body, _ := io.ReadAll(httpResp.Body) var pollResp struct { Data struct { Status string `json:"status"` Outputs []string `json:"outputs"` Error string `json:"error"` } `json:"data"` } if err := json.Unmarshal(body, &pollResp); err != nil { return nil, fmt.Errorf("failed to parse poll response: %w", err) } switch pollResp.Data.Status { case "completed": resp := &ai.ImageResponse{} for _, output := range pollResp.Data.Outputs { resp.Images = append(resp.Images, ai.Image{URL: output}) } return resp, nil case "failed": return nil, fmt.Errorf("image generation failed: %s", pollResp.Data.Error) default: return nil, nil } } const defaultVideoModel = "google/gemini-omni-flash/image-to-video-developer" // GenerateVideo creates a video using Atlas Cloud's async video API. // Supports text-to-video and image-to-video depending on whether // Images are provided in the request. func (p *Provider) GenerateVideo(ctx context.Context, req *ai.VideoRequest, opts ...ai.GenerateOption) (*ai.VideoResponse, error) { model := req.Model if model == "" { model = defaultVideoModel } duration := req.Duration if duration <= 0 { duration = 6 } aspect := req.AspectRatio if aspect == "" { aspect = "16:9" } resolution := req.Resolution if resolution == "" { resolution = "720p" } apiReq := map[string]any{ "model": model, "prompt": req.Prompt, "duration": duration, "aspect_ratio": aspect, "resolution": resolution, "seed": -1, } if len(req.Images) > 0 { apiReq["images"] = req.Images } reqBody, err := json.Marshal(apiReq) if err != nil { return nil, fmt.Errorf("failed to marshal request: %w", err) } apiURL := strings.TrimRight(p.opts.BaseURL, "/") + "/api/v1/model/generateVideo" httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(reqBody)) if err != nil { return nil, fmt.Errorf("failed to create request: %w", err) } httpReq.Header.Set("Content-Type", "application/json") httpReq.Header.Set("Authorization", "Bearer "+p.opts.APIKey) httpResp, err := http.DefaultClient.Do(httpReq) if err != nil { return nil, fmt.Errorf("API request failed: %w", err) } defer httpResp.Body.Close() respBody, _ := io.ReadAll(httpResp.Body) if httpResp.StatusCode != http.StatusOK { return nil, fmt.Errorf("API error (%s): %s", httpResp.Status, string(respBody)) } var submitResp struct { Code int `json:"code"` Msg string `json:"message"` Data struct { ID string `json:"id"` Status string `json:"status"` } `json:"data"` } if err := json.Unmarshal(respBody, &submitResp); err != nil { return nil, fmt.Errorf("failed to parse submit response: %w", err) } if submitResp.Code != 200 { return nil, fmt.Errorf("API error: %s", submitResp.Msg) } pollURL := strings.TrimRight(p.opts.BaseURL, "/") + "/api/v1/model/prediction/" + submitResp.Data.ID ticker := time.NewTicker(5 * time.Second) defer ticker.Stop() for { select { case <-ctx.Done(): return nil, ctx.Err() case <-ticker.C: result, err := p.pollVideo(ctx, pollURL) if err != nil { return nil, err } if result != nil { return result, nil } } } } func (p *Provider) pollVideo(ctx context.Context, url string) (*ai.VideoResponse, error) { httpReq, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) if err != nil { return nil, err } httpReq.Header.Set("Authorization", "Bearer "+p.opts.APIKey) httpResp, err := http.DefaultClient.Do(httpReq) if err != nil { return nil, fmt.Errorf("poll request failed: %w", err) } defer httpResp.Body.Close() body, _ := io.ReadAll(httpResp.Body) var pollResp struct { Data struct { Status string `json:"status"` Outputs []string `json:"outputs"` Error string `json:"error"` } `json:"data"` } if err := json.Unmarshal(body, &pollResp); err != nil { return nil, fmt.Errorf("failed to parse poll response: %w", err) } switch pollResp.Data.Status { case "completed", "succeeded": if len(pollResp.Data.Outputs) == 0 { return nil, fmt.Errorf("video completed but no outputs returned") } return &ai.VideoResponse{URL: pollResp.Data.Outputs[0]}, nil case "failed": return nil, fmt.Errorf("video generation failed: %s", pollResp.Data.Error) default: return nil, nil } }