aigo

README

aigo

中文说明

aigo is an agent-native Go SDK for multimodal media generation. Describe work as a lightweight workflow graph, route it to 35+ execution engines, and get structured results with error classification, retry hints, and progress callbacks.

Architecture

Agent (LLM / code)
  │
  ▼
AgentTask ──► BuildGraph() ──► workflow.Graph (DAG)
                                    │
              ┌─────────┬──────────┼──────────┬──────────┐
              ▼          ▼          ▼          ▼          ▼
        engine/kling  engine/luma  engine/fal  ...   engine/comfyui
              │          │          │                     │
              ▼          ▼          ▼                     ▼
         Kling API   Luma API   Fal API             ComfyUI WS

Engines

Image Generation

Engine	Backend	Env Var
alibabacloud	Alibaba Cloud DashScope (Qwen, Wan, Z-Image, HappyHorse)	DASHSCOPE_API_KEY
openai	OpenAI Images (gpt-image-2, DALL-E 3, DALL-E 2)	OPENAI_API_KEY
google	Google Imagen	GOOGLE_API_KEY
flux	Black Forest Labs FLUX	BFL_API_KEY
stability	Stability AI (SD3, Ultra, Core)	STABILITY_API_KEY
ideogram	Ideogram	IDEOGRAM_API_KEY
recraft	Recraft V3	RECRAFT_API_KEY
midjourney	Midjourney (via GoAPI)	GOAPI_KEY
jimeng	Jimeng (Volcengine)	JIMENG_API_KEY
liblib	LibLibAI (HMAC-SHA1 auth)	LIBLIB_ACCESS_KEY / LIBLIB_SECRET_KEY
ark	Volcengine Ark (Seedream image)	ARK_API_KEY

Video Generation

Engine	Backend	Env Var
alibabacloud	DashScope HappyHorse (t2v, i2v, r2v, video-edit)	DASHSCOPE_API_KEY
alibabacloud	DashScope Wan (t2v, i2v, r2v, video-edit)	DASHSCOPE_API_KEY
alibabacloud	DashScope Kling V3 (video, omni-video)	DASHSCOPE_API_KEY
ark	Volcengine Ark Seedance 2.0 / 2.0 Fast / 1.0 Lite	ARK_API_KEY
kling	Kling AI (v1/v1.5/v2/v2.1)	KLING_API_KEY
hailuo	Hailuo / MiniMax Video	HAILUO_API_KEY
luma	Luma Dream Machine	LUMA_API_KEY
runway	Runway Gen-3/Gen-4	RUNWAY_API_KEY
pika	Pika Labs	PIKA_API_KEY
hedra	Hedra (talking head video)	HEDRA_API_KEY

Audio / Music / Speech

Engine	Backend	Env Var
alibabacloud	DashScope TTS (Qwen3-TTS Flash / Instruct Flash)	DASHSCOPE_API_KEY
alibabacloud	DashScope Voice Design (Qwen Voice Design)	DASHSCOPE_API_KEY
alibabacloud	DashScope ASR (Qwen3-ASR Flash / Filetrans)	DASHSCOPE_API_KEY
alibabacloud	DashScope Fun-Music v1 (music generation)	DASHSCOPE_API_KEY
elevenlabs	ElevenLabs TTS	ELEVENLABS_API_KEY
minimax	MiniMax TTS & Music	MINIMAX_API_KEY
suno	Suno Music Generation	SUNO_API_KEY
volcvoice	Volcengine Speech	VOLC_SPEECH_ACCESS_TOKEN

3D Generation

Engine	Backend	Env Var
meshy	Meshy (text/image to 3D)	MESHY_API_KEY
alibabacloud	Alibaba Cloud Bailian — Tripo P1.0 / Tripo H3.1 (text / image / multi-image to 3D, .glb)	DASHSCOPE_API_KEY

Multi-Modal Understanding

Engine	Backend	Env Var
gemini	Google Gemini (vision + text)	GEMINI_API_KEY
gpt4o	OpenAI GPT-4o Vision	OPENAI_API_KEY

Multi-Backend / Gateway

Engine	Backend	Env Var
newapi	Multi-route gateway (OpenAI, Kling, Jimeng, Sora, Qwen, Gemini, Seedance)	NEWAPI_API_KEY
openrouter	OpenRouter (multi-provider routing)	OPENROUTER_API_KEY
fal	Fal.ai (generic model runner)	FAL_KEY
replicate	Replicate (generic model runner)	REPLICATE_API_TOKEN
comfydeploy	ComfyDeploy (hosted ComfyUI)	COMFYDEPLOY_API_TOKEN
comfyui	ComfyUI server (WebSocket)	COMFY_CLOUD_API_KEY
runninghub	RunningHub (ComfyUI cloud)	RH_API_KEY

Embedding

Engine	Backend	Env Var
embed/openai	OpenAI Embeddings	OPENAI_API_KEY
embed/gemini	Google Gemini Embeddings	GEMINI_API_KEY
embed/alibabacloud	DashScope Embeddings (text + multimodal)	DASHSCOPE_API_KEY
embed/jina	Jina Embeddings	JINA_API_KEY
embed/voyage	Voyage AI Embeddings	VOYAGE_API_KEY

Install

go get github.com/godeps/aigo

Quick Start

Simple prompt

client := aigo.NewClient()

_ = client.RegisterEngine("img", alibabacloud.New(alibabacloud.Config{
    Model: alibabacloud.ModelQwenImage,
}))

result, err := client.ExecutePrompt(ctx, "img", "A shiba inu riding a vintage motorcycle")
fmt.Println(result.Value)   // URL or data URI
fmt.Println(result.Kind)    // aigo.OutputURL, OutputDataURI, etc.
fmt.Println(result.Engine)  // "img"
fmt.Println(result.Elapsed) // execution duration

Rich Result type

Every execution method returns aigo.Result:

type Result struct {
    Value    string         // raw output (URL, data URI, JSON, etc.)
    Kind     OutputKind     // authoritative classification
    Engine   string         // which engine produced this
    Elapsed  time.Duration  // wall-clock execution time
    Metadata map[string]any // engine-specific data (optional)
}

fmt.Println(result) // Result implements String(), prints Value

Structured task

result, err := client.ExecuteTask(ctx, "video", aigo.AgentTask{
    Prompt:   "Turn this product scene into a short ad",
    Duration: 2,
    Structured: &aigo.AgentTaskStructured{
        VideoSize: "1280*720",
        ImageSize: "1024*1024",
    },
    References: []aigo.ReferenceAsset{
        {Type: aigo.ReferenceTypeVideo, URL: "https://example.com/input.mp4"},
        {Type: aigo.ReferenceTypeImage, URL: "https://example.com/style.png"},
    },
})

TTS (text-to-speech)

result, err := client.ExecuteTask(ctx, "tts", aigo.AgentTask{
    Prompt: "Welcome to our product launch event",
    TTS: &aigo.TTSOptions{
        Voice:        "zhiyan",
        LanguageType: "zh",
    },
})

Voice design

result, err := client.ExecuteTask(ctx, "vd", aigo.AgentTask{
    Prompt: "design a voice",
    VoiceDesign: &aigo.VoiceDesignOptions{
        VoicePrompt:   "A warm, friendly female voice",
        PreviewText:   "Hello, welcome!",
        TargetModel:   "cosyvoice-v2",
        PreferredName: "custom-voice-01",
    },
})

Batch Registration

Register multiple engines at once

client.RegisterAll(map[string]engine.Engine{
    "img":   alibabacloud.New(alibabacloud.Config{Model: alibabacloud.ModelQwenImage}),
    "video": kling.New(kling.Config{Model: kling.ModelKlingV2Master}),
})

Conditional registration (skip if env var is missing)

client.RegisterAllIfKey([]aigo.EngineEntry{
    {Name: "kling-video", Engine: klingEngine, EnvVars: []string{"KLING_API_KEY"}},
    {Name: "luma-video",  Engine: lumaEngine,  EnvVars: []string{"LUMA_API_KEY"}},
    {Name: "local",       Engine: localEngine}, // always registered
})

Provider grouping

Register all engines from a vendor in one call. Engines whose required env vars are not set are silently skipped:

import "github.com/godeps/aigo/engine/alibabacloud"

registered, _ := client.RegisterProvider(alibabacloud.DefaultProvider())
// registered: ["alibabacloud-image", "alibabacloud-video", "alibabacloud-tts"]

Every engine package exports DefaultProvider() with sensible presets.

Config-file-driven setup

Declare engines in a JSON file instead of code:

{
  "engines": [
    {"name": "img",   "provider": "alibabacloud", "model": "qwen-image"},
    {"name": "video", "provider": "kling",         "model": "kling-v2-master"},
    {"name": "tts",   "provider": "elevenlabs"},
    {"name": "backup","provider": "runway",         "enabled": false}
  ]
}

cfg, _ := aigo.LoadConfig("engines.json")
registered, _ := client.ApplyConfig(cfg)

Each engine package registers its factory via init(), so importing the package is sufficient:

import (
    _ "github.com/godeps/aigo/engine/alibabacloud"
    _ "github.com/godeps/aigo/engine/kling"
    _ "github.com/godeps/aigo/engine/elevenlabs"
    _ "github.com/godeps/aigo/engine/runway"
)

Model i18n Metadata

Every model registered via engine packages carries i18n display names and descriptions:

import _ "github.com/godeps/aigo/engine/kling"

info, ok := client.ModelInfo("kling-v2-master")
fmt.Println(info.DisplayName["en"]) // "Kling V2 Master"
fmt.Println(info.DisplayName["zh"]) // "可灵 V2 大师版"
fmt.Println(info.Description["zh"]) // "最高画质视频和图片生成"
fmt.Println(info.Intro["zh"])       // "可灵旗舰视频生成模型，支持文生视频和图生视频..."
fmt.Println(info.DocURL)            // "https://docs.qingque.cn/..."
fmt.Println(info.Capability)        // "video"

List all registered models:

for _, m := range client.AllModelInfos() {
    fmt.Printf("%-25s %-8s %s\n", m.Name, m.Capability, m.DisplayName["zh"])
}

You can also use engine.LookupModelInfo(name) and engine.AllModelInfos() directly.

Filter models by capability

videoModels := client.ModelInfosByCapability("video")
for _, m := range videoModels {
    fmt.Printf("%s — %s\n", m.Name, m.DisplayName["en"])
}

Query models by provider (engine)

klingModels := client.ModelInfosByProvider("kling")
// Returns all ModelInfo entries registered by the kling engine package

Engine-level metadata

meta := engine.LookupEngineMetadata("kling")
fmt.Println(meta.DisplayName["en"]) // "Kling AI"
fmt.Println(meta.Intro["en"])       // detailed introduction
fmt.Println(meta.DocURL)            // official documentation URL

Agent-Native Features

Structured errors with retry classification

Errors from all engines are classified so agents can make retry decisions:

import "github.com/godeps/aigo/engine/aigoerr"

_, err := client.ExecutePrompt(ctx, "img", "...")
if aigoerr.IsRetryable(err) {
    // safe to retry (429, 5xx, timeout)
}

code, ok := aigoerr.GetCode(err)
// aigoerr.CodeRateLimit, CodeServerError, CodeInvalidInput, etc.

var ae *aigoerr.Error
if errors.As(err, &ae) {
    fmt.Println(ae.StatusCode)  // original HTTP status
    fmt.Println(ae.RetryAfter)  // parsed Retry-After header
}

JSON Schema tool definitions

Register aigo tools with any agent framework (OpenAI, Anthropic, LangChain, Vercel AI SDK):

import "github.com/godeps/aigo/tooldef"

tools := tooldef.AllTools()
// generate_image, generate_video, generate_3d, text_to_speech,
// design_voice, edit_image, edit_video, transcribe_audio, generate_music

Engine registry

Centralized engine registration, lookup, and capability-based discovery:

import "github.com/godeps/aigo/engine"

reg := engine.NewRegistry()
reg.Register("kling", engine.Entry{
    Name:   "kling",
    Engine: klingEngine,
    ConfigSchemaFunc:   kling.ConfigSchema,
    ModelsByCapability: kling.ModelsByCapability,
})

// Find all engines that can generate video
videoEngines := reg.FindByCapability("video")

// Get all models grouped by engine and capability
allModels := reg.AllModels()

Engine capability discovery

Query what engines can do — for dynamic tool selection:

cap, _ := client.EngineCapabilities("alibabacloud-img")
// cap.MediaTypes  → ["image"]
// cap.Models      → ["qwen-image"]
// cap.SupportsSync, cap.SupportsPoll

// Find all engines that handle video:
videoEngines := client.AvailableFor("video")

Engine controls

Dynamically enable, disable, or conditionally register engines:

// Disable an engine without removing it
client.DisableEngine("runway")

// Re-enable it later
client.EnableEngine("runway")

// Remove an engine entirely
client.UnregisterEngine("old-engine")

// Register only if the API key is available
client.RegisterEngineIfKey("kling", klingEngine, "KLING_API_KEY")

// Check if an engine is active
if client.IsEnabled("kling") { ... }

Selective tool definitions

Filter tool definitions by media type for your agent framework:

import "github.com/godeps/aigo/tooldef"

// All tools
tools := tooldef.AllTools() // 9 tools

// Only image tools (generate_image, edit_image)
imageTools := tooldef.ToolsFor("image")

// Multiple categories
mediaTools := tooldef.ToolsFor("video", "audio", "music")

Progress reporting

Monitor long-running tasks:

result, err := client.Execute(ctx, "video", graph, aigo.WithProgress(func(e aigo.ProgressEvent) {
    fmt.Printf("[%s] elapsed=%s\n", e.Phase, e.Elapsed)
    // Phase: "submitted", "completed"
}))

Result caching

Cache results to avoid redundant API calls:

import "github.com/godeps/aigo/engine"

cached := engine.WithCache(myEngine, 10*time.Minute, 100) // TTL + max entries
// Identical workflow graphs return cached results

HTTP retry & rate limiting

Built-in HTTP transports for resilient API calls:

import "github.com/godeps/aigo/engine/httpx"

// Auto-retry on 429/5xx with exponential backoff
retryClient := httpx.NewRetryClient(httpx.RetryOptions{
    MaxRetries: 3,
    BaseDelay:  time.Second,
})

// Token bucket rate limiting
rateLimitedClient := httpx.NewRateLimitedClient(10, 20, 30*time.Second) // 10 RPS, burst 20

Middleware

Add cross-cutting concerns (logging, retry, timing):

client.Use(aigo.WithLogging(os.Stderr))
client.Use(aigo.WithRetry(3)) // retry retryable errors up to 3 times

Pipeline chaining

Chain multi-step workflows where each step feeds the next:

p := aigo.NewPipeline("img", aigo.AgentTask{Prompt: "a cat"}).
    Then(func(prev aigo.Result) (aigo.AgentTask, string) {
        return aigo.AgentTask{
            Prompt:     "animate this image",
            References: []aigo.ReferenceAsset{{Type: aigo.ReferenceTypeImage, URL: prev.Value}},
        }, "video"
    })

results, err := client.ExecutePipeline(ctx, p)
// results[0] = image result, results[1] = video result

DryRun estimation

Check what would happen without executing:

dr, err := client.DryRun("video", aigo.AgentTask{Prompt: "..."})
// dr.WillPoll       — whether the engine will poll
// dr.EstimatedTime  — human-readable time estimate
// dr.Warnings       — potential issues

Auto-routing with selector

Let the LLM inside your agent choose the engine:

result, err := client.ExecuteTaskAuto(ctx, selector, aigo.AgentTask{
    Prompt:   "make a 2 second product video",
    Duration: 2,
})
// result.Engine       — which engine was selected
// result.Reason       — why it was selected
// result.Output.Value — the generation result

Capability-aware routing (RichSelector)

RichSelector receives engine capability metadata so the LLM (or rules) can make informed decisions:

// Query all engine capabilities
infos := client.EngineInfos()
// []EngineInfo{{Name: "kling", Capability: {MediaTypes: ["video"], MaxDuration: 10, ...}}, ...}

// RichSelector automatically receives capabilities — no extra code needed
result, err := client.ExecuteTaskAuto(ctx, myRichSelector, task)

Rule-based pre-filtering

Filter incompatible engines before LLM selection — by media type, size, duration, and voice:

filter := &aigo.RuleFilter{}
candidates := filter.Filter(task, client.EngineInfos())
// Only engines matching the task's constraints remain

Priority selector (no LLM needed)

Pick the first compatible engine from a priority-ordered list:

selector := &aigo.PrioritySelector{
    Priority: []string{"kling", "luma", "runway"},
    Filter:   &aigo.RuleFilter{}, // optional constraint filtering
}
result, err := client.ExecuteTaskAuto(ctx, selector, task)

Infer media type from task

Automatically detect what kind of media a task needs:

mediaType := aigo.InferMediaType(task)
// "video" if Duration > 0, "audio" if TTS set, "music" if Music set, "image" otherwise

Fallback across engines

Try multiple engines in order; first success wins:

result, err := client.ExecuteWithFallback(ctx, []string{"primary", "backup"}, graph)
// result.Engine       — which engine succeeded
// result.Output.Value — the result
// result.Skipped      — engines that failed (with errors)

Auto-routing with fallback

Combine selector-based routing with automatic failover:

result, err := client.ExecuteTaskAutoWithFallback(ctx, selector, task)
// Selector picks the best engine; if it fails, tries remaining candidates

Async execution

Non-blocking execution via channel:

ch := client.ExecuteAsync(ctx, "video", graph)
// ... do other work ...
ar := <-ch
if ar.Err != nil { ... }
fmt.Println(ar.Result.Value)

Low-Level API

If your agent already emits workflow graphs, call Execute directly:

graph := workflow.Graph{
    "1": {
        ClassType: "CLIPTextEncode",
        Inputs:    map[string]any{"text": "A cinematic lighthouse in a storm"},
    },
    "2": {
        ClassType: "EmptyLatentImage",
        Inputs:    map[string]any{"width": 1536, "height": 1024},
    },
}

result, err := client.Execute(ctx, "img", graph)

Internal Packages

Package	Purpose
workflow	Workflow graph types and validation
workflow/resolve	Graph resolution (prompt extraction, option helpers, link following)
engine/poll	Unified polling with backoff, progress callbacks, and status mapping
engine/httpx	HTTP client defaults, retry transport, rate limiting, file upload
engine/aigoerr	Structured error classification for agent retry logic
engine/embed	Embedding engine implementations (OpenAI, Gemini, Jina, Voyage, Aliyun)
tooldef	JSON Schema tool definitions for agent frameworks

Examples

# Alibaba Cloud
go run ./examples/alibabacloud_qwen_image
go run ./examples/alibabacloud_wan_image
go run ./examples/alibabacloud_zimage
go run ./examples/alibabacloud_wan_t2v
go run ./examples/alibabacloud_wan_r2v
go run ./examples/alibabacloud_wan_videoedit
go run ./examples/alibabacloud_qwen_tts
go run ./examples/alibabacloud_qwen_voice_design
ENGINE=text  go run ./examples/alibabacloud_tripo_3d  # also: ENGINE=image | ENGINE=multi

# New API gateway
go run ./examples/newapi_image
go run ./examples/newapi_speech
go run ./examples/newapi_video
go run ./examples/newapi_seedance_video

# Auto-routing
go run ./examples/agent_auto_router

Notes

• Alibaba Cloud result URLs are temporary OSS links. Persist them immediately.
• All async engines support WaitForCompletion mode for synchronous use and Resume() for reconnecting to running tasks.
• All engines use unified API key resolution via engine.ResolveKey — configure via struct field or environment variable.

Documentation

Index

• Constants
• Variables
• func BuildGraph(task AgentTask) workflow.Graph
• func InferMediaType(task AgentTask) string
• type AgentTask
• type AgentTaskStructured
• type AsyncResult
• type Client
  • func NewClient(opts ...ClientOption) *Client
  • func (c *Client) AllModelInfos() []engine.ModelInfo
  • func (c *Client) ApplyConfig(cfg FileConfig) ([]string, error)
  • func (c *Client) AvailableFor(mediaType string) []string
  • func (c *Client) DisableEngine(name string) error
  • func (c *Client) DryRun(engineName string, task AgentTask) (engine.DryRunResult, error)
  • func (c *Client) EnableEngine(name string) error
  • func (c *Client) EngineCapabilities(name string) (engine.Capability, error)
  • func (c *Client) EngineInfos() []EngineInfo
  • func (c *Client) EngineNames() []string
  • func (c *Client) Execute(ctx context.Context, engineName string, graph workflow.Graph, ...) (Result, error)
  • func (c *Client) ExecuteAsync(ctx context.Context, engineName string, graph workflow.Graph) <-chan AsyncResult
  • func (c *Client) ExecutePipeline(ctx context.Context, p *Pipeline) ([]Result, error)
  • func (c *Client) ExecutePrompt(ctx context.Context, engineName string, prompt string) (Result, error)
  • func (c *Client) ExecutePromptAuto(ctx context.Context, selector Selector, prompt string) (RoutedResult, error)
  • func (c *Client) ExecuteTask(ctx context.Context, engineName string, task AgentTask, opts ...ExecuteOption) (Result, error)
  • func (c *Client) ExecuteTaskAuto(ctx context.Context, selector Selector, task AgentTask) (RoutedResult, error)
  • func (c *Client) ExecuteTaskAutoWithFallback(ctx context.Context, selector Selector, task AgentTask) (RoutedResult, error)
  • func (c *Client) ExecuteTaskWithFallback(ctx context.Context, engines []string, task AgentTask, opts ...ExecuteOption) (FallbackResult, error)
  • func (c *Client) ExecuteWithFallback(ctx context.Context, engines []string, graph workflow.Graph, ...) (FallbackResult, error)
  • func (c *Client) ExportModelCatalog() ([]byte, error)
  • func (c *Client) IsEnabled(name string) bool
  • func (c *Client) ModelInfo(model string) (engine.ModelInfo, bool)
  • func (c *Client) ModelInfosByCapability(cap string) []engine.ModelInfo
  • func (c *Client) ModelInfosByProvider(provider string) []engine.ModelInfo
  • func (c *Client) RecoverPending() ([]TaskRecord, error)
  • func (c *Client) RegisterAll(engines map[string]engine.Engine) error
  • func (c *Client) RegisterAllIfKey(entries []EngineEntry) ([]string, error)
  • func (c *Client) RegisterEngine(name string, e engine.Engine) error
  • func (c *Client) RegisterEngineIfKey(name string, e engine.Engine, envVars ...string) (bool, error)
  • func (c *Client) RegisterProvider(p engine.Provider) ([]string, error)
  • func (c *Client) Resume(ctx context.Context, taskID string) (Result, error)
  • func (c *Client) SearchModelInfos(query string) []engine.ModelInfo
  • func (c *Client) Submit(ctx context.Context, engineName string, graph workflow.Graph, ...) (string, error)
  • func (c *Client) UnregisterEngine(name string) error
  • func (c *Client) Use(mw ...Middleware)
• type ClientOption
  • func WithDefaultStore() (ClientOption, error)
  • func WithStore(s TaskStore) ClientOption
• type EngineEntry
• type EngineInfo
• type ExecuteOption
  • func WithProgress(fn func(ProgressEvent)) ExecuteOption
• type FallbackError
  • func (e FallbackError) Error() string
  • func (e FallbackError) Unwrap() error
• type FallbackResult
• type FileConfig
  • func LoadConfig(path string) (FileConfig, error)
• type FileTaskStore
  • func DefaultFileTaskStore() (*FileTaskStore, error)
  • func NewFileTaskStore(path string) (*FileTaskStore, error)
  • func (s *FileTaskStore) All() ([]TaskRecord, error)
  • func (s *FileTaskStore) Delete(id string) error
  • func (s *FileTaskStore) Load(id string) (TaskRecord, error)
  • func (s *FileTaskStore) Purge(maxAge time.Duration) error
  • func (s *FileTaskStore) Save(record TaskRecord) error
• type Middleware
  • func WithLogging(w io.Writer) Middleware
  • func WithRetry(maxRetries int) Middleware
• type MusicOptions
• type OutputKind
  • func InterpretOutputKind(s string) OutputKind
• type Pipeline
  • func NewPipeline(engineName string, task AgentTask) *Pipeline
  • func (p *Pipeline) Then(fn PipelineStep) *Pipeline
• type PipelineStep
• type PrioritySelector
  • func (s *PrioritySelector) SelectEngine(_ context.Context, task AgentTask, engines []string) (Selection, error)
  • func (s *PrioritySelector) SelectEngineFromCandidates(_ context.Context, task AgentTask, candidates []EngineInfo) (Selection, error)
• type ProgressEvent
• type ReferenceAsset
• type ReferenceType
• type Result
  • func (r Result) String() string
• type RichSelector
• type RoutedResult
• type RuleFilter
  • func (f *RuleFilter) Filter(task AgentTask, candidates []EngineInfo) []EngineInfo
• type Selection
• type Selector
• type TTSOptions
• type TaskRecord
• type TaskStore
• type VoiceDesignOptions

Constants

const (
	OutputUnknown   = engine.OutputUnknown
	OutputURL       = engine.OutputURL
	OutputDataURI   = engine.OutputDataURI
	OutputJSON      = engine.OutputJSON
	OutputPlainText = engine.OutputPlainText
)

const (
	TaskStatusPending   = "pending"
	TaskStatusCompleted = "completed"
	TaskStatusFailed    = "failed"
)

TaskStatus constants.

Variables

var (
	ErrEngineNil       = errors.New("aigo: engine is nil")
	ErrEngineExists    = errors.New("aigo: engine already registered")
	ErrEngineNotFound  = errors.New("aigo: engine not found")
	ErrEngineNameEmpty = errors.New("aigo: engine name is empty")
	ErrEngineDisabled  = errors.New("aigo: engine is disabled")
)

var (
	ErrStoreNotConfigured = errors.New("aigo: task store not configured")
	ErrTaskNotFound       = errors.New("aigo: task not found")
	ErrResumeNotSupported = errors.New("aigo: engine does not support resume")
)

Functions

func BuildGraph

func BuildGraph(task AgentTask) workflow.Graph

BuildGraph compiles a high-level agent task into the SDK's workflow graph format.

func InferMediaType

func InferMediaType(task AgentTask) string

InferMediaType guesses the required media type from an AgentTask's fields.

Types

type AgentTask

type AgentTask struct {
	Prompt         string
	NegativePrompt string
	Width          int
	Height         int
	Size           string
	Duration       int
	Watermark      *bool
	References     []ReferenceAsset

	TTS         *TTSOptions
	VoiceDesign *VoiceDesignOptions
	Music       *MusicOptions

	// Structured groups image/video options separately for finer control.
	Structured *AgentTaskStructured
}

AgentTask is a graph-free request shape for agents.

type AgentTaskStructured

type AgentTaskStructured struct {
	ImageSize        string
	ImageAspectRatio string
	ImageResolution  string // "1K", "2K", "4K"
	ImageCameraAngle string
	ImageWatermark   *bool
	VideoDuration    int
	VideoSize        string
	VideoWatermark   *bool
	VideoAspectRatio string
	VideoResolution  string // "480P", "720P", "1080P"
	VideoAudio       *bool
}

AgentTaskStructured 将图像与视频选项分组；便于扩展而无需继续增大 AgentTask 扁平字段。

type AsyncResult

type AsyncResult struct {
	Result Result
	Err    error
}

AsyncResult delivers an asynchronous execution outcome.

type Client

type Client struct {
	// contains filtered or unexported fields
}

Client routes a workflow graph to a registered execution engine.

func NewClient

func NewClient(opts ...ClientOption) *Client

NewClient creates a new SDK client.

func (*Client) AllModelInfos

func (c *Client) AllModelInfos() []engine.ModelInfo

AllModelInfos returns all registered model i18n metadata, sorted by name.

func (*Client) ApplyConfig

func (c *Client) ApplyConfig(cfg FileConfig) ([]string, error)

ApplyConfig registers engines from a FileConfig using registered factories. Disabled entries are silently skipped. Returns the names of successfully registered engines.

func (*Client) AvailableFor

func (c *Client) AvailableFor(mediaType string) []string

AvailableFor returns enabled engine names whose capabilities include the given media type. Disabled engines are excluded. Engines that do not implement Describer are included (assumed capable).

func (*Client) DisableEngine

func (c *Client) DisableEngine(name string) error

DisableEngine temporarily disables a registered engine without removing it. Disabled engines are excluded from Execute, EngineNames, EngineInfos, and AvailableFor. Returns ErrEngineNotFound if the name is not registered.

func (*Client) DryRun

func (c *Client) DryRun(engineName string, task AgentTask) (engine.DryRunResult, error)

DryRun checks what would happen without actually executing the task. Returns an estimation if the engine implements DryRunner; otherwise returns a basic result based on Describer capabilities.

func (*Client) EnableEngine

func (c *Client) EnableEngine(name string) error

EnableEngine re-enables a previously disabled engine. Returns ErrEngineNotFound if the name is not registered.

func (*Client) EngineCapabilities

func (c *Client) EngineCapabilities(name string) (engine.Capability, error)

EngineCapabilities returns the capabilities of a registered engine. If the engine does not implement Describer, an empty Capability is returned.

func (*Client) EngineInfos

func (c *Client) EngineInfos() []EngineInfo

EngineInfos returns capability metadata for all enabled engines, sorted by name. Disabled engines are excluded. Engines that do not implement Describer get an empty Capability.

func (*Client) EngineNames

func (c *Client) EngineNames() []string

EngineNames returns enabled engine names in deterministic order. Disabled engines are excluded.

func (*Client) Execute

func (c *Client) Execute(ctx context.Context, engineName string, graph workflow.Graph, opts ...ExecuteOption) (Result, error)

Execute dispatches the graph to the named engine. Returns ErrEngineNotFound if the engine is not registered, or ErrEngineDisabled if it is disabled.

func (*Client) ExecuteAsync

func (c *Client) ExecuteAsync(ctx context.Context, engineName string, graph workflow.Graph) <-chan AsyncResult

ExecuteAsync runs Execute in a goroutine and delivers the result on the returned channel. The channel is closed after sending exactly one value. Cancelling ctx stops the work.

func (*Client) ExecutePipeline

func (c *Client) ExecutePipeline(ctx context.Context, p *Pipeline) ([]Result, error)

ExecutePipeline runs each step in sequence, feeding results forward.

func (*Client) ExecutePrompt

func (c *Client) ExecutePrompt(ctx context.Context, engineName string, prompt string) (Result, error)

ExecutePrompt runs the simplest agent request: a single prompt.

func (*Client) ExecutePromptAuto

func (c *Client) ExecutePromptAuto(ctx context.Context, selector Selector, prompt string) (RoutedResult, error)

ExecutePromptAuto lets a selector choose the engine for a prompt-driven request.

func (*Client) ExecuteTask

func (c *Client) ExecuteTask(ctx context.Context, engineName string, task AgentTask, opts ...ExecuteOption) (Result, error)

ExecuteTask converts an agent task into a workflow graph and routes it to the target engine.

func (*Client) ExecuteTaskAuto

func (c *Client) ExecuteTaskAuto(ctx context.Context, selector Selector, task AgentTask) (RoutedResult, error)

ExecuteTaskAuto lets a selector choose the engine for a structured agent task. If the selector implements RichSelector, capability metadata is collected and passed automatically.

func (*Client) ExecuteTaskAutoWithFallback

func (c *Client) ExecuteTaskAutoWithFallback(ctx context.Context, selector Selector, task AgentTask) (RoutedResult, error)

ExecuteTaskAutoWithFallback selects an engine and retries with alternatives on failure. If the selector is a RichSelector, all candidates receive capability metadata.

func (*Client) ExecuteTaskWithFallback

func (c *Client) ExecuteTaskWithFallback(ctx context.Context, engines []string, task AgentTask, opts ...ExecuteOption) (FallbackResult, error)

ExecuteTaskWithFallback is the AgentTask variant of ExecuteWithFallback.

func (*Client) ExecuteWithFallback

func (c *Client) ExecuteWithFallback(ctx context.Context, engines []string, graph workflow.Graph, opts ...ExecuteOption) (FallbackResult, error)

ExecuteWithFallback tries each engine in order; the first success wins. All engines that fail are recorded in FallbackResult.Skipped.

func (*Client) ExportModelCatalog

func (c *Client) ExportModelCatalog() ([]byte, error)

ExportModelCatalog returns the complete model catalog as JSON bytes.

func (*Client) IsEnabled

func (c *Client) IsEnabled(name string) bool

IsEnabled reports whether a registered engine is currently enabled.

func (*Client) ModelInfo

func (c *Client) ModelInfo(model string) (engine.ModelInfo, bool)

ModelInfo returns the i18n metadata for a specific model name.

func (*Client) ModelInfosByCapability

func (c *Client) ModelInfosByCapability(cap string) []engine.ModelInfo

ModelInfosByCapability returns all registered model metadata matching the given capability.

func (*Client) ModelInfosByProvider

func (c *Client) ModelInfosByProvider(provider string) []engine.ModelInfo

ModelInfosByProvider returns all registered model metadata for the given provider/engine name.

func (*Client) RecoverPending

func (c *Client) RecoverPending() ([]TaskRecord, error)

RecoverPending returns all tasks that are still pending (not completed or failed). Callers should iterate and call Resume for each.

func (*Client) RegisterAll

func (c *Client) RegisterAll(engines map[string]engine.Engine) error

RegisterAll registers multiple engines at once. Stops and returns the first error encountered.

func (*Client) RegisterAllIfKey

func (c *Client) RegisterAllIfKey(entries []EngineEntry) ([]string, error)

RegisterAllIfKey registers multiple engines, each gated by env vars. Engines whose env vars are not set are silently skipped. Returns the names of engines that were actually registered.

func (*Client) RegisterEngine

func (c *Client) RegisterEngine(name string, e engine.Engine) error

RegisterEngine registers an engine under a logical name.

func (*Client) RegisterEngineIfKey

func (c *Client) RegisterEngineIfKey(name string, e engine.Engine, envVars ...string) (bool, error)

RegisterEngineIfKey registers an engine only if the required API key is available. It checks the environment variables in order; if any is set, the engine is registered. Returns true if the engine was registered, false if skipped (no key found).

func (*Client) RegisterProvider

func (c *Client) RegisterProvider(p engine.Provider) ([]string, error)

RegisterProvider registers all engines from a provider. Engines whose required env vars are not set are silently skipped. Returns the names of engines that were actually registered.

func (*Client) Resume

func (c *Client) Resume(ctx context.Context, taskID string) (Result, error)

Resume polls a previously submitted async task to completion. The taskID is the aigo-side ID returned by Submit.

func (*Client) SearchModelInfos

func (c *Client) SearchModelInfos(query string) []engine.ModelInfo

SearchModelInfos returns models whose Name or DisplayName contains the query (case-insensitive).

func (*Client) Submit

func (c *Client) Submit(ctx context.Context, engineName string, graph workflow.Graph, opts ...ExecuteOption) (string, error)

Submit executes the graph and persists the task for crash recovery. If the engine completes synchronously, the result is stored as completed and an empty task ID is returned. If the engine returns a remote task ID (WaitForCompletion=false), the record is stored as pending and the aigo-side task ID is returned for later Resume.

func (*Client) UnregisterEngine

func (c *Client) UnregisterEngine(name string) error

UnregisterEngine removes a previously registered engine. Returns ErrEngineNotFound if the name is not registered.

func (*Client) Use

func (c *Client) Use(mw ...Middleware)

Use appends middleware that wraps every engine on each Execute call. Middleware is applied in the order added (first added = outermost wrapper).

type ClientOption

type ClientOption func(*Client)

ClientOption configures the Client at construction time.

func WithDefaultStore

func WithDefaultStore() (ClientOption, error)

WithDefaultStore attaches a FileTaskStore at .aigo/tasks.json (current working directory). Returns an error if the store directory cannot be created.

func WithStore

func WithStore(s TaskStore) ClientOption

WithStore attaches a TaskStore for async task persistence and crash recovery.

type EngineEntry

type EngineEntry struct {
	Name    string
	Engine  engine.Engine
	EnvVars []string // required env vars; if empty, always register
}

EngineEntry describes an engine to register with optional env-var gating.

type EngineInfo

type EngineInfo struct {
	Name        string             `json:"name"`
	DisplayName engine.DisplayName `json:"display_name"`
	Intro       engine.DisplayName `json:"intro,omitempty"`
	DocURL      string             `json:"doc_url,omitempty"`
	Capability  engine.Capability  `json:"capability"`
	Models      []engine.ModelInfo `json:"models,omitempty"`
}

EngineInfo describes a candidate engine's capabilities, provided to RichSelector for informed decisions.

type ExecuteOption

type ExecuteOption func(*executeConfig)

ExecuteOption configures optional Execute behavior.

func WithProgress

func WithProgress(fn func(ProgressEvent)) ExecuteOption

WithProgress registers a callback for execution progress events.

type FallbackError

type FallbackError struct {
	Engine string
	Err    error
}

FallbackError records a single engine failure during fallback execution.

func (FallbackError) Error

func (e FallbackError) Error() string

func (FallbackError) Unwrap

func (e FallbackError) Unwrap() error

type FallbackResult

type FallbackResult struct {
	Engine  string
	Output  Result
	Skipped []FallbackError
}

FallbackResult is the outcome of a fallback-enabled execution.

type FileConfig

type FileConfig struct {
	Engines []engine.EngineConfig `json:"engines"`
}

FileConfig is the top-level structure for JSON-based engine configuration.

func LoadConfig

func LoadConfig(path string) (FileConfig, error)

LoadConfig reads engine configuration from a JSON file.

type FileTaskStore

type FileTaskStore struct {
	// contains filtered or unexported fields
}

FileTaskStore implements TaskStore using a single JSON file. It uses sync.Mutex for in-process safety and flock for cross-process safety.

func DefaultFileTaskStore

func DefaultFileTaskStore() (*FileTaskStore, error)

DefaultFileTaskStore creates a file-backed task store at .aigo/tasks.json relative to the current working directory.

func NewFileTaskStore

func NewFileTaskStore(path string) (*FileTaskStore, error)

NewFileTaskStore creates a file-backed task store at the given path.

func (*FileTaskStore) All

func (s *FileTaskStore) All() ([]TaskRecord, error)

func (*FileTaskStore) Delete

func (s *FileTaskStore) Delete(id string) error

func (*FileTaskStore) Load

func (s *FileTaskStore) Load(id string) (TaskRecord, error)

func (*FileTaskStore) Purge

func (s *FileTaskStore) Purge(maxAge time.Duration) error

Purge removes completed and failed records older than maxAge.

func (*FileTaskStore) Save

func (s *FileTaskStore) Save(record TaskRecord) error

type Middleware

type Middleware func(name string, next engine.Engine) engine.Engine

Middleware wraps an engine to add cross-cutting behavior (logging, timing, retry, etc.).

func WithLogging

func WithLogging(w io.Writer) Middleware

WithLogging returns middleware that logs engine calls to the given writer.

func WithRetry

func WithRetry(maxRetries int) Middleware

WithRetry returns middleware that retries on retryable errors up to maxRetries times.

type MusicOptions

type MusicOptions struct {
	Lyrics          string
	IsInstrumental  *bool
	LyricsOptimizer *bool
	OutputFormat    string // "url" or "hex"
	SampleRate      int
	Bitrate         int
	Format          string // "mp3", "wav", "flac"
}

MusicOptions groups music generation parameters.

type OutputKind

type OutputKind = engine.OutputKind

OutputKind is an alias for engine.OutputKind.

func InterpretOutputKind

func InterpretOutputKind(s string) OutputKind

InterpretOutputKind 根据内容前缀做轻量分类；任务 id、纯数字等会落在 OutputPlainText。

type Pipeline

type Pipeline struct {
	// contains filtered or unexported fields
}

Pipeline chains multiple engine executions where each step's input depends on the previous output.

func NewPipeline

func NewPipeline(engineName string, task AgentTask) *Pipeline

NewPipeline starts a pipeline with an initial engine and task.

func (*Pipeline) Then

func (p *Pipeline) Then(fn PipelineStep) *Pipeline

Then appends a step that transforms the previous result into the next task.

type PipelineStep

type PipelineStep func(prev Result) (AgentTask, string)

PipelineStep transforms a previous result into the next task and target engine.

type PrioritySelector

type PrioritySelector struct {
	Priority []string    // Engine names in preference order.
	Filter   *RuleFilter // Optional constraint filter.
}

PrioritySelector picks the first compatible engine from a priority-ordered list. It optionally pre-filters candidates using a RuleFilter.

func (*PrioritySelector) SelectEngine

func (s *PrioritySelector) SelectEngine(_ context.Context, task AgentTask, engines []string) (Selection, error)

SelectEngine implements Selector using the priority list against flat engine names.

func (*PrioritySelector) SelectEngineFromCandidates

func (s *PrioritySelector) SelectEngineFromCandidates(_ context.Context, task AgentTask, candidates []EngineInfo) (Selection, error)

SelectEngineFromCandidates implements RichSelector with capability-aware filtering.

type ProgressEvent

type ProgressEvent struct {
	Phase   string        // "submitted", "polling", "completed"
	Attempt int           // poll attempt number (0 for non-polling phases)
	Elapsed time.Duration // wall-clock time since execution start
}

ProgressEvent reports execution progress to the caller.

type ReferenceAsset

type ReferenceAsset struct {
	Type ReferenceType
	URL  string
}

ReferenceAsset describes an externally reachable media input.

type ReferenceType

type ReferenceType string

ReferenceType identifies the kind of remote asset to attach to an agent task.

const (
	ReferenceTypeImage ReferenceType = "image"
	ReferenceTypeVideo ReferenceType = "video"
)

type Result

type Result struct {
	Value    string         // Raw output from the engine.
	Kind     OutputKind     // Authoritative classification from the engine.
	Engine   string         // Name of the engine that produced the result.
	Elapsed  time.Duration  // Wall-clock execution time.
	Metadata map[string]any // Engine-specific data (optional).
}

Result is the public outcome of every Client execution method.

func (Result) String

func (r Result) String() string

String returns the raw output value, allowing fmt.Sprint(result) to work naturally.

type RichSelector

type RichSelector interface {
	Selector
	SelectEngineFromCandidates(ctx context.Context, task AgentTask, candidates []EngineInfo) (Selection, error)
}

RichSelector is an enhanced selector that receives engine capability metadata. Implementations can use capability information to make better routing decisions. A RichSelector is also a Selector — ExecuteTaskAuto auto-detects the interface.

type RoutedResult

type RoutedResult struct {
	Engine string
	Reason string
	Output Result
}

RoutedResult is the result of a selector-driven execution.

type RuleFilter

type RuleFilter struct{}

RuleFilter filters candidates based on task constraints.

func (*RuleFilter) Filter

func (f *RuleFilter) Filter(task AgentTask, candidates []EngineInfo) []EngineInfo

Filter returns candidates compatible with the task. It checks media type, size, duration, and voice constraints.

type Selection

type Selection struct {
	Engine string
	Reason string
}

Selection is the selector's routing decision.

type Selector

type Selector interface {
	SelectEngine(ctx context.Context, task AgentTask, engines []string) (Selection, error)
}

Selector decides which registered engine should handle a task.

type TTSOptions

type TTSOptions struct {
	Voice                string
	LanguageType         string
	Instructions         string
	OptimizeInstructions *bool
}

TTSOptions groups text-to-speech parameters.

type TaskRecord

type TaskRecord struct {
	ID         string     `json:"id"`
	EngineName string     `json:"engine"`
	RemoteID   string     `json:"remote_id"`
	Status     string     `json:"status"`
	CreatedAt  time.Time  `json:"created_at"`
	UpdatedAt  time.Time  `json:"updated_at"`
	ResultVal  string     `json:"result_value,omitempty"`
	ResultKind OutputKind `json:"result_kind,omitempty"`
	ErrMsg     string     `json:"error,omitempty"`
}

TaskRecord represents a persisted async task.

type TaskStore

type TaskStore interface {
	Save(record TaskRecord) error
	Load(id string) (TaskRecord, error)
	All() ([]TaskRecord, error)
	Delete(id string) error
}

TaskStore persists async task records for crash recovery.

type VoiceDesignOptions

type VoiceDesignOptions struct {
	VoicePrompt    string
	PreviewText    string
	TargetModel    string
	PreferredName  string
	Language       string
	SampleRate     int
	ResponseFormat string
	OmitPreview    bool
}

VoiceDesignOptions groups voice design parameters.