flowchart TB
    subgraph Clients["客户端层"]
        WebClient["Web 应用<br/>React/Vue"]
        MobileClient["移动应用<br/>iOS/Android"]
        APIClient["第三方 API 客户端"]
    end

    subgraph Gateway["API 网关层"]
        direction LR
        AuthGW["认证网关<br/>JWT 验证<br/>验证延迟 < 5ms"]
        RateLimiter["限流器<br/>令牌桶算法<br/>100 req/min/user"]
    end

    subgraph Orchestrator["AI Orchestrator 服务层 (Kratos)"]
        direction TB

        subgraph ServerLayer["Server 层（传输层）"]
            HTTPServer["HTTP Server<br/>:8000<br/>Kratos HTTP Transport"]
            GRPCServer["gRPC Server<br/>:9000<br/>Kratos gRPC Transport"]
        end

        subgraph ServiceLayer["Service 层（应用服务层）"]
            OrchestratorSvc["OrchestratorService<br/>对外服务接口<br/>gRPC/HTTP 双协议"]
        end

        subgraph BizLayer["Biz 层（业务用例层）"]
            TaskUC["TaskUsecase<br/>任务生命周期管理<br/>Pipeline 编排"]
            Metrics["Metrics 指标<br/>Prometheus 采集<br/>5 类核心指标"]
        end

        subgraph DomainLayer["Domain 层（领域模型层）"]
            TaskAgg["Task 聚合根<br/>状态机管理<br/>5 种状态"]
            RAGPipe["RAG Pipeline<br/>检索 + 生成<br/>2 步流程"]
            AgentPipe["Agent Pipeline<br/>ReAct 循环<br/>最多 10 次迭代"]
            VoicePipe["Voice Pipeline<br/>ASR/TTS<br/>语音处理"]
            PipelineIntf["Pipeline 接口<br/>统一执行协议"]
        end

        subgraph DataLayer["Data 层（数据访问层）"]
            TaskRepo["TaskRepository<br/>GORM 实现<br/>JSONB 存储"]
            ServiceClient["ServiceClient<br/>gRPC 客户端<br/>熔断 + 重试"]
            DB["DB Connection<br/>PostgreSQL<br/>连接池管理"]
        end

        subgraph InfraLayer["Infrastructure 层（基础设施）"]
            CircuitBreaker["熔断器<br/>gobreaker<br/>失败率 60% 触发"]
            RetryMech["重试机制<br/>指数退避<br/>最多 3 次"]
            Tracing["分布式追踪<br/>OpenTelemetry<br/>Jaeger 导出"]
        end

        HTTPServer --> OrchestratorSvc
        GRPCServer --> OrchestratorSvc
        OrchestratorSvc --> TaskUC
        TaskUC --> TaskAgg
        TaskUC --> RAGPipe
        TaskUC --> AgentPipe
        TaskUC --> VoicePipe
        TaskUC --> TaskRepo
        TaskUC --> Metrics
        RAGPipe -.实现.-> PipelineIntf
        AgentPipe -.实现.-> PipelineIntf
        VoicePipe -.实现.-> PipelineIntf
        RAGPipe --> ServiceClient
        AgentPipe --> ServiceClient
        VoicePipe --> ServiceClient
        TaskRepo --> DB
        ServiceClient --> CircuitBreaker
        ServiceClient --> RetryMech
        OrchestratorSvc --> Tracing
    end

    subgraph AIEngines["AI 引擎集群（Python 服务）"]
        RAGEngine["RAG Engine<br/>:8001<br/>FastAPI<br/>检索增强生成"]
        AgentEngine["Agent Engine<br/>:8003<br/>FastAPI<br/>ReAct Agent"]
        VoiceEngine["Voice Engine<br/>:8006<br/>FastAPI<br/>ASR + TTS"]
        MultimodalEngine["Multimodal Engine<br/>:8007<br/>FastAPI<br/>多模态理解"]
    end

    subgraph SupportServices["支撑服务"]
        RetrievalService["Retrieval Service<br/>:9001<br/>混合检索<br/>向量 + 全文 + RRF"]
        ModelRouter["Model Router<br/>:9005<br/>智能路由<br/>成本优化 30%"]
    end

    subgraph Storage["存储层"]
        PostgreSQL["PostgreSQL 14+<br/>ai_orchestrator.tasks 表<br/>JSONB 字段<br/>复合索引优化"]
        Redis["Redis 7+<br/>任务缓存（未来支持）<br/>TTL: 1h"]
    end

    subgraph Monitoring["可观测性"]
        Prometheus["Prometheus<br/>指标采集<br/>10s 频率"]
        Jaeger["Jaeger<br/>分布式追踪<br/>采样率 1%"]
        Grafana["Grafana<br/>监控大盘<br/>实时刷新"]
    end

    Clients --> Gateway
    Gateway --> HTTPServer
    Gateway --> GRPCServer

    ServiceClient --> RAGEngine
    ServiceClient --> AgentEngine
    ServiceClient --> VoiceEngine
    ServiceClient --> MultimodalEngine
    ServiceClient --> RetrievalService
    ServiceClient --> ModelRouter

    DB --> PostgreSQL
    TaskRepo -.未来.-> Redis

    Metrics --> Prometheus
    Tracing --> Jaeger
    Prometheus --> Grafana

    style Clients fill:#e3f2fd
    style Gateway fill:#fff9c4
    style Orchestrator fill:#fff3e0
    style AIEngines fill:#f3e5f5
    style SupportServices fill:#e8f5e9
    style Storage fill:#e0f2f1
    style Monitoring fill:#f1f8e9

flowchart TB
    subgraph Client["客户端"]
        HTTPClient["HTTP 客户端"]
        GRPCClient["gRPC 客户端"]
    end

    subgraph ServerLayer["Server 层（传输层）"]
        HTTPServer["HTTP Server<br/>Kratos HTTP Transport<br/>:8000"]
        GRPCServer["gRPC Server<br/>Kratos gRPC Transport<br/>:9000"]
    end

    subgraph Middleware["Kratos 中间件链"]
        Recovery["Recovery<br/>恢复 panic"]
        Tracing["Tracing<br/>追踪上报"]
        Logging["Logging<br/>请求日志"]
        Validate["Validate<br/>参数验证（仅 gRPC）"]
    end

    subgraph ServiceLayer["Service 层（应用服务）"]
        OrchestratorSvc["OrchestratorService<br/>・CreateTask()<br/>・ExecuteTask()<br/>・GetTask()<br/>・CancelTask()<br/>・CreateAndExecuteTask()"]
    end

    subgraph BizLayer["Biz 层（业务用例）"]
        TaskUC["TaskUsecase<br/>・CreateTask()<br/>・ExecuteTask()<br/>・GetTask()<br/>・CancelTask()<br/>・ProcessPendingTasks()"]
        MetricsCollector["Metrics Collector<br/>・TaskCreatedTotal<br/>・TaskExecutionDuration<br/>・TaskStatusTotal<br/>・TaskTokensUsed<br/>・TaskCostUSD"]
    end

    subgraph DomainLayer["Domain 层（领域模型）"]
        TaskAgg["Task 聚合根<br/>・NewTask()<br/>・Start() / Complete()<br/>・Fail() / Cancel()<br/>・AddStep()"]
        RAGPipeline["RAG Pipeline<br/>・Execute()<br/>检索 + 生成"]
        AgentPipeline["Agent Pipeline<br/>・Execute()<br/>ReAct 循环"]
        VoicePipeline["Voice Pipeline<br/>・Execute()<br/>ASR 识别"]
        PipelineIntf["Pipeline 接口<br/>Execute(task) → output"]
    end

    subgraph DataLayer["Data 层（数据访问）"]
        TaskRepo["TaskRepository<br/>GORM 实现<br/>・Create() / Update()<br/>・GetByID()<br/>・ListPending()"]
        ServiceClient["GRPCServiceClient<br/>・Call(service, method, input)<br/>・熔断 + 重试 + 超时"]
        DBConn["DB Connection<br/>PostgreSQL<br/>连接池"]
    end

    subgraph InfraLayer["Infrastructure（基础设施）"]
        CircuitBreaker["Circuit Breaker<br/>gobreaker<br/>失败率 60% 触发"]
        RetryMech["Retry Mechanism<br/>指数退避<br/>最多 3 次"]
        OTelTracer["OpenTelemetry Tracer<br/>Span 创建<br/>Jaeger 导出"]
    end

    subgraph ExternalServices["外部服务"]
        RetrievalSvc["Retrieval Service<br/>:9001<br/>混合检索"]
        RAGEngine["RAG Engine<br/>:8001<br/>生成答案"]
        AgentEngine["Agent Engine<br/>:8003<br/>ReAct 执行"]
        VoiceEngine["Voice Engine<br/>:8006<br/>ASR/TTS"]
    end

    subgraph Storage["存储"]
        PostgreSQL["PostgreSQL<br/>ai_orchestrator.tasks<br/>JSONB 字段"]
    end

    subgraph Observability["可观测性"]
        Prometheus["Prometheus<br/>指标采集"]
        Jaeger["Jaeger<br/>追踪存储"]
    end

    %% 请求流
    HTTPClient --> HTTPServer
    GRPCClient --> GRPCServer

    HTTPServer --> Recovery
    GRPCServer --> Recovery
    Recovery --> Tracing
    Tracing --> Logging
    Logging --> Validate
    Validate --> OrchestratorSvc

    OrchestratorSvc --> TaskUC

    TaskUC --> TaskAgg
    TaskUC --> TaskRepo
    TaskUC --> RAGPipeline
    TaskUC --> AgentPipeline
    TaskUC --> VoicePipeline
    TaskUC --> MetricsCollector

    RAGPipeline -.实现.-> PipelineIntf
    AgentPipeline -.实现.-> PipelineIntf
    VoicePipeline -.实现.-> PipelineIntf

    RAGPipeline --> ServiceClient
    AgentPipeline --> ServiceClient
    VoicePipeline --> ServiceClient

    TaskRepo --> DBConn
    DBConn --> PostgreSQL

    ServiceClient --> CircuitBreaker
    CircuitBreaker --> RetryMech
    RetryMech --> RetrievalSvc
    RetryMech --> RAGEngine
    RetryMech --> AgentEngine
    RetryMech --> VoiceEngine

    Tracing --> OTelTracer
    OTelTracer --> Jaeger

    MetricsCollector --> Prometheus

    style Client fill:#e3f2fd
    style ServerLayer fill:#fff3e0
    style Middleware fill:#fff9c4
    style ServiceLayer fill:#e8f5e9
    style BizLayer fill:#c8e6c9
    style DomainLayer fill:#f8bbd0
    style DataLayer fill:#b2dfdb
    style InfraLayer fill:#ffe0b2
    style ExternalServices fill:#e1bee7
    style Storage fill:#b3e5fc
    style Observability fill:#f0f4c3

sequenceDiagram
    autonumber
    participant Client as 客户端
    participant HTTPServer as HTTP Server<br/>:8000
    participant Middleware as Kratos 中间件链<br/>Recovery→Tracing→Logging
    participant OrcSvc as OrchestratorService<br/>CreateAndExecuteTask()
    participant TaskUC as TaskUsecase<br/>CreateTask() / ExecuteTask()
    participant TaskAgg as Task 聚合根<br/>NewTask() / Start() / Complete()
    participant TaskRepo as TaskRepository<br/>GORM 实现
    participant RAGPipe as RAGPipeline<br/>Execute()
    participant SvcClient as GRPCServiceClient<br/>Call()
    participant CB as Circuit Breaker<br/>gobreaker
    participant Retry as Retry Mechanism<br/>指数退避
    participant Retrieval as Retrieval Service<br/>:9001
    participant RAGEngine as RAG Engine<br/>:8001
    participant DB as PostgreSQL<br/>tasks 表
    participant Metrics as Prometheus<br/>指标采集
    participant Tracer as Jaeger<br/>分布式追踪

    Client->>HTTPServer: POST /api/v1/orchestrator/rag<br/>Content-Type: application/json<br/>{"conversation_id": "conv_123", "message": "什么是DDD?", ...}

    Note over HTTPServer: HTTP/1.1 或 HTTP/2
    HTTPServer->>Middleware: 请求进入中间件链

    Note over Middleware: 1. Recovery 捕获 panic
    Note over Middleware: 2. Tracing 创建 Span（TraceID: abc123）
    Note over Middleware: 3. Logging 记录请求日志
    Middleware->>Tracer: 创建 Span<br/>operation: POST /api/v1/orchestrator/rag<br/>attributes: method, path

    Middleware->>OrcSvc: CreateAndExecuteTask(ctx, req)

    Note over OrcSvc: 参数转换<br/>CreateTaskRequest → domain.TaskInput
    OrcSvc->>OrcSvc: input = &domain.TaskInput{<br/>  Content: "什么是DDD?"<br/>  Mode: "rag"<br/>}

    Note over OrcSvc: ===== 阶段 1: 创建任务 =====
    OrcSvc->>TaskUC: CreateTask(ctx, TaskTypeRAG, conv_id, user_id, tenant_id, input)

    TaskUC->>TaskAgg: NewTask(TaskTypeRAG, conv_id, user_id, tenant_id, input)
    TaskAgg->>TaskAgg: 生成任务 ID<br/>id = "task_" + uuid.New()
    TaskAgg->>TaskAgg: 初始化状态<br/>Status = TaskStatusPending<br/>Priority = TaskPriorityNormal
    TaskAgg-->>TaskUC: task（内存对象）

    TaskUC->>TaskRepo: Create(ctx, task)
    TaskRepo->>TaskRepo: toTaskPO(task)<br/>领域对象 → 持久化对象<br/>JSON 序列化 Input/Output/Steps
    TaskRepo->>DB: INSERT INTO ai_orchestrator.tasks<br/>(id, type, status, priority, input, ...)<br/>VALUES (...)
    DB-->>TaskRepo: ok（10-20ms）
    TaskRepo-->>TaskUC: nil

    TaskUC->>Metrics: TaskCreatedTotal.WithLabelValues("rag", tenant_id).Inc()
    Metrics-->>TaskUC:
    TaskUC-->>OrcSvc: task

    Note over OrcSvc: ===== 阶段 2: 执行任务 =====
    OrcSvc->>TaskUC: ExecuteTask(ctx, task.ID)

    TaskUC->>TaskRepo: GetByID(ctx, task.ID)
    TaskRepo->>DB: SELECT * FROM tasks WHERE id = ?
    DB-->>TaskRepo: TaskPO 对象（10-20ms）
    TaskRepo->>TaskRepo: toDomainTask(po)<br/>持久化对象 → 领域对象<br/>JSON 反序列化
    TaskRepo-->>TaskUC: task

    TaskUC->>TaskUC: 检查状态<br/>if task.Status != Pending { error }

    TaskUC->>TaskUC: 获取 Pipeline<br/>pipeline = pipelines[TaskTypeRAG]<br/>→ RAGPipeline 实例

    TaskUC->>TaskAgg: task.Start()
    TaskAgg->>TaskAgg: Status = Running<br/>StartedAt = time.Now()
    TaskAgg-->>TaskUC:

    TaskUC->>TaskRepo: Update(ctx, task)
    TaskRepo->>DB: UPDATE tasks SET status='running', started_at=NOW() WHERE id=?
    DB-->>TaskRepo: ok（10-20ms）
    TaskRepo-->>TaskUC:

    Note over TaskUC: 执行 Pipeline

    TaskUC->>RAGPipe: Execute(task)

    Note over RAGPipe: ===== 步骤 1: 检索文档 =====
    RAGPipe->>TaskAgg: step1 = NewTaskStep("检索文档", "retrieval-service")
    RAGPipe->>TaskAgg: task.AddStep(step1)

    RAGPipe->>SvcClient: Call("retrieval-service", "retrieve", {<br/>  "query": "什么是DDD?",<br/>  "tenant_id": "tenant_1",<br/>  "top_k": 20<br/>})

    SvcClient->>CB: cb.Execute(func() { ... })

    Note over CB: 检查熔断器状态<br/>state = Closed（正常）
    CB->>Retry: callWithRetry(...)

    Note over Retry: 重试循环（最多 3 次）
    Retry->>Retry: 尝试 1/3
    Retry->>Retrieval: gRPC Call: Retrieve(query, top_k=20)

    Note over Retrieval: 混合检索<br/>1. 向量检索（Milvus）<br/>2. 全文检索（ES）<br/>3. RRF 重排序
    Retrieval-->>Retry: {"chunks": [...], "count": 20}（50-200ms）

    Retry-->>CB: result
    CB-->>SvcClient: result

    SvcClient-->>RAGPipe: {"chunks": [...]}

    RAGPipe->>TaskAgg: step1.CompleteStep({"chunks": [...]})
    TaskAgg->>TaskAgg: step1.Status = Completed<br/>step1.CompletedAt = time.Now()

    Note over RAGPipe: ===== 步骤 2: RAG 生成 =====
    RAGPipe->>TaskAgg: step2 = NewTaskStep("RAG生成", "rag-engine")
    RAGPipe->>TaskAgg: task.AddStep(step2)

    RAGPipe->>SvcClient: Call("rag-engine", "generate", {<br/>  "query": "什么是DDD?",<br/>  "context": chunks,<br/>  "history": []<br/>})

    SvcClient->>CB: cb.Execute(func() { ... })
    CB->>Retry: callWithRetry(...)

    Retry->>Retry: 尝试 1/3
    Retry->>RAGEngine: gRPC Call: Generate(query, context, history)

    Note over RAGEngine: 1. 构建 Prompt<br/>2. 调用 LLM<br/>3. 流式生成（简化为同步）
    RAGEngine-->>Retry: {<br/>  "answer": "DDD是领域驱动设计...",<br/>  "tokens_used": 500,<br/>  "model": "gpt-4-turbo",<br/>  "cost_usd": 0.005,<br/>  "latency_ms": 2000<br/>}（1-3s）

    Retry-->>CB: result
    CB-->>SvcClient: result

    SvcClient-->>RAGPipe: {"answer": "...", "tokens_used": 500, ...}

    RAGPipe->>TaskAgg: step2.CompleteStep({"answer": "...", ...})

    Note over RAGPipe: 构建 TaskOutput
    RAGPipe->>RAGPipe: output = &TaskOutput{<br/>  Content: ragResult["answer"],<br/>  TokensUsed: 500,<br/>  Model: "gpt-4-turbo",<br/>  CostUSD: 0.005,<br/>  LatencyMS: 2000,<br/>  Metadata: ragResult<br/>}

    RAGPipe-->>TaskUC: output

    Note over TaskUC: ===== 阶段 3: 完成任务 =====
    TaskUC->>TaskAgg: task.Complete(output)
    TaskAgg->>TaskAgg: Status = Completed<br/>Output = output<br/>CompletedAt = time.Now()
    TaskAgg-->>TaskUC:

    TaskUC->>TaskRepo: Update(ctx, task)
    TaskRepo->>DB: UPDATE tasks SET<br/>  status='completed',<br/>  output=jsonb,<br/>  steps=jsonb,<br/>  completed_at=NOW()<br/>WHERE id=?
    DB-->>TaskRepo: ok（10-20ms）
    TaskRepo-->>TaskUC:

    Note over TaskUC: 采集指标
    TaskUC->>Metrics: TaskStatusTotal.WithLabelValues("rag", "completed", tenant_id).Inc()
    TaskUC->>Metrics: TaskExecutionDuration.WithLabelValues("rag", "completed", tenant_id).Observe(2.5)
    TaskUC->>Metrics: TaskTokensUsed.WithLabelValues("rag", "gpt-4-turbo", tenant_id).Observe(500)
    TaskUC->>Metrics: TaskCostUSD.WithLabelValues("rag", "gpt-4-turbo", tenant_id).Observe(0.005)
    Metrics-->>TaskUC:

    TaskUC-->>OrcSvc: output

    Note over OrcSvc: 响应转换<br/>domain.Task → TaskResponse<br/>domain.TaskOutput → TaskOutputResponse
    OrcSvc->>OrcSvc: response = &ExecuteTaskResponse{<br/>  Task: toTaskResponse(task),<br/>  Output: toTaskOutputResponse(output)<br/>}

    OrcSvc-->>Middleware: response

    Middleware->>Tracer: Span.End()<br/>记录延迟: 2.5s
    Tracer-->>Middleware:

    Middleware->>Middleware: Logging 记录完成日志<br/>status: 200, duration: 2.5s

    Middleware-->>HTTPServer: response

    HTTPServer-->>Client: HTTP 200 OK<br/>Content-Type: application/json<br/>{<br/>  "task": {...},<br/>  "output": {<br/>    "content": "DDD是领域驱动设计...",<br/>    "tokens_used": 500,<br/>    "cost_usd": 0.005,<br/>    "latency_ms": 2000<br/>  }<br/>}

    Note over Client: 端到端延迟: 2.5-3秒<br/>（检索 150ms + 生成 2s + DB 40ms + 其他 310ms）

sequenceDiagram
    autonumber
    participant Client as 客户端
    participant HTTP as HTTP Server
    participant Pipeline as 处理管道
    participant AgentEx as AgentChatExecutor
    participant CB as CircuitBreaker
    participant OrcUC as OrchestratorUsecase
    participant Agent as Agent Engine
    participant Tools as 工具服务
    participant DB as PostgreSQL
    participant Stream as SSE Stream

    Client->>HTTP: POST /api/v1/orchestrator/process<br/>{mode: "agent", message: "帮我查天气并发邮件"}

    HTTP->>Pipeline: Handle(request)
    Note over Pipeline: 日志、验证、指标 < 5ms

    Pipeline->>AgentEx: Execute(ctx, request, stream)

    AgentEx->>CB: ExecuteWithContext(fn)
    Note over CB: 熔断检查 Closed

    Note over AgentEx: 创建任务
    CB->>OrcUC: CreateTask(type=Agent)
    OrcUC->>DB: INSERT INTO tasks
    DB-->>OrcUC: ok (10-20ms)

    Note over AgentEx: 调用 Agent Engine
    OrcUC->>Agent: Execute(task, tools, max_iterations=5)

    Note over Agent: ReAct 循环 - Iteration 1
    Agent->>Agent: LLM Reasoning<br/>"需要先查询天气"
    AgentEx->>Stream: Send({type: "thinking", content: "分析任务..."})
    Stream-->>Client: SSE: thinking

    Agent->>Agent: 决策：调用 weather_api
    AgentEx->>Stream: Send({type: "tool_call", tool: "weather_api"})
    Stream-->>Client: SSE: tool_call

    Agent->>Tools: weather_api.call({city: "北京"})
    Tools-->>Agent: {temp: 15, condition: "晴"} (200-500ms)

    Agent->>Agent: 记录 Observation<br/>"北京 15°C 晴"

    Note over Agent: ReAct 循环 - Iteration 2
    Agent->>Agent: LLM Reasoning<br/>"天气已获取，需要发邮件"
    AgentEx->>Stream: Send({type: "thinking", content: "准备发送邮件..."})
    Stream-->>Client: SSE: thinking

    Agent->>Agent: 决策：调用 email_api
    AgentEx->>Stream: Send({type: "tool_call", tool: "email_api"})
    Stream-->>Client: SSE: tool_call

    Agent->>Tools: email_api.send({to: "user", body: "..."})
    Tools-->>Agent: {success: true} (500-1000ms)

    Note over Agent: ReAct 循环 - Iteration 3
    Agent->>Agent: LLM Reasoning<br/>"任务完成"
    Agent->>Agent: 生成最终答案<br/>"已查询天气并发送邮件"

    Agent-->>OrcUC: {<br/>  final_answer: "...",<br/>  steps: 3,<br/>  tools_called: ["weather_api", "email_api"],<br/>  tokens: 800,<br/>  cost: 0.008<br/>} (3-10s)

    OrcUC->>DB: UPDATE tasks (steps, output)

    AgentEx->>Stream: Send({type: "text", content: final_answer})
    Stream-->>Client: SSE: text

    AgentEx->>Stream: Send({type: "final", metrics: {...}})
    Stream-->>Client: SSE: final

    CB->>CB: afterRequest(nil)<br/>failures = 0
    CB-->>AgentEx: nil

    AgentEx-->>Pipeline: nil
    Pipeline-->>HTTP: 200 OK
    HTTP-->>Client: 连接关闭

// RAG Pipeline - 两步执行
func (p *RAGPipeline) Execute(task *Task) (*TaskOutput, error) {
    // 步骤 1: 检索文档
    retrievalResult, err := p.retrievalClient.Call("retrieval-service", "retrieve", ...)

    // 步骤 2: RAG 生成
    ragResult, err := p.ragEngineClient.Call("rag-engine", "generate", ...)

    return &TaskOutput{Content: ragResult["answer"].(string), ...}, nil
}

// Agent Pipeline - 单步执行（内部多次迭代）
func (p *AgentPipeline) Execute(task *Task) (*TaskOutput, error) {
    // 单步：Agent 执行（内部包含 ReAct 循环）
    result, err := p.agentClient.Call("agent-engine", "execute", map[string]interface{}{
        "task": task.Input.Content,
        "max_iterations": 10,
    })

    return &TaskOutput{Content: result["result"].(string), ...}, nil
}

classDiagram
    class Task {
        +string ID
        +TaskType Type
        +TaskStatus Status
        +TaskPriority Priority
        +string ConversationID
        +string UserID
        +string TenantID
        +TaskInput Input
        +TaskOutput Output
        +[]TaskStep Steps
        +map Metadata
        +time CreatedAt
        +time UpdatedAt
        +Start()
        +Complete(output)
        +Fail(error)
        +Cancel()
        +AddStep(step)
    }

    class TaskInput {
        +string Content
        +string Mode
        +map Context
        +map Options
    }

    class TaskOutput {
        +string Content
        +map Metadata
        +int TokensUsed
        +string Model
        +float64 CostUSD
        +int LatencyMS
    }

    class TaskStep {
        +string ID
        +string Name
        +string Service
        +map Input
        +map Output
        +TaskStatus Status
        +string Error
        +time StartedAt
        +time CompletedAt
        +int DurationMS
        +CompleteStep(output)
        +FailStep(error)
    }

    class TaskType {
        <<enumeration>>
        RAG
        AGENT
        CHAT
        VOICE
        MULTIMODAL
    }

    class TaskStatus {
        <<enumeration>>
        PENDING
        RUNNING
        COMPLETED
        FAILED
        CANCELLED
    }

    class TaskPriority {
        <<enumeration>>
        LOW = 0
        NORMAL = 5
        HIGH = 10
    }

    Task "1" *-- "1" TaskInput
    Task "1" *-- "0..1" TaskOutput
    Task "1" *-- "*" TaskStep
    Task --> TaskType
    Task --> TaskStatus
    Task --> TaskPriority
    TaskStep --> TaskStatus

type CreateTaskRequest struct {
    TaskType       string              // "rag" / "agent" / "voice"
    ConversationID string              // 会话 ID
    UserID         string              // 用户 ID
    TenantID       string              // 租户 ID
    Input          *TaskInputRequest
}

type TaskInputRequest struct {
    Content string                      // 用户输入内容
    Mode    string                      // 模式
    Context map[string]interface{}      // 上下文（如历史消息）
    Options map[string]interface{}      // 选项配置
}

type ExecuteTaskResponse struct {
    Task   *TaskResponse         // 任务信息
    Output *TaskOutputResponse   // 执行输出
}

type TaskResponse struct {
    ID             string                    // 任务 ID
    Type           string                    // 任务类型
    Status         string                    // 任务状态
    Priority       int32                     // 优先级
    ConversationID string                    // 会话 ID
    UserID         string                    // 用户 ID
    TenantID       string                    // 租户 ID
    CreatedAt      *timestamppb.Timestamp    // 创建时间
    UpdatedAt      *timestamppb.Timestamp    // 更新时间
}

type TaskOutputResponse struct {
    Content    string                   // 输出内容（如 RAG 答案）
    Metadata   map[string]interface{}   // 元数据
    TokensUsed int32                    // Token 使用量
    Model      string                   // 使用的模型
    CostUSD    float64                  // 成本（美元）
    LatencyMS  int32                    // 延迟（毫秒）
}

func (s *OrchestratorService) CreateAndExecuteTask(
    ctx context.Context,
    req *CreateTaskRequest,
) (*ExecuteTaskResponse, error) {
    // 转换输入
    input := &domain.TaskInput{
        Content: req.Input.Content,
        Mode:    req.Input.Mode,
        Context: req.Input.Context,
        Options: req.Input.Options,
    }

    // 创建并执行（调用 TaskUsecase）
    task, output, err := s.taskUC.CreateAndExecuteTask(
        ctx,
        domain.TaskType(req.TaskType),
        req.ConversationID,
        req.UserID,
        req.TenantID,
        input,
    )
    if err != nil {
        return nil, err
    }

    // 构建响应
    return &ExecuteTaskResponse{
        Task:   s.toTaskResponse(task),
        Output: s.toTaskOutputResponse(output),
    }, nil
}

{
  "task_type": "rag",
  "conversation_id": "conv_123",
  "user_id": "user_456",
  "tenant_id": "tenant_789",
  "input": {
    "content": "什么是领域驱动设计？",
    "mode": "rag",
    "context": {},
    "options": {}
  }
}

{
  "task": {
    "id": "task_uuid",
    "type": "rag",
    "status": "completed",
    "conversation_id": "conv_123",
    "created_at": "2025-10-28T12:00:00Z"
  },
  "output": {
    "content": "领域驱动设计（DDD）是一种软件开发方法...",
    "tokens_used": 500,
    "model": "gpt-4-turbo",
    "cost_usd": 0.005,
    "latency_ms": 2500
  }
}

func (uc *OrchestratorUsecase) selectMode(message, preferredMode string) string {
    // 如果指定了模式，直接使用
    if preferredMode != "" && preferredMode != "auto" {
        return preferredMode
    }

    // 基于关键词的启发式规则
    message = strings.ToLower(message)

    // Agent 模式关键词
    agentPatterns := []string{
        "帮我.*查询",
        "执行.*操作",
        "搜索.*并.*发送",
        "完成.*任务",
    }
    for _, pattern := range agentPatterns {
        if matched, _ := regexp.MatchString(pattern, message); matched {
            return "agent"
        }
    }

    // RAG 模式关键词
    ragPatterns := []string{
        "什么是",
        "如何.*",
        "介绍.*",
        "文档中.*",
        ".*的定义",
    }
    for _, pattern := range ragPatterns {
        if matched, _ := regexp.MatchString(pattern, message); matched {
            return "rag"
        }
    }

    // 默认使用 RAG（知识问答）
    return "rag"
}

func (uc *OrchestratorUsecase) selectMode(message, preferredMode string) string {
    // 如果用户显式指定模式，直接使用
    if preferredMode != "" && preferredMode != "auto" {
        return preferredMode
    }

    // 基于关键词的启发式规则
    message = strings.ToLower(message)

    // Agent 模式关键词检测
    agentPatterns := []string{
        "帮我.*查询",
        "执行.*操作",
        "搜索.*并.*发送",
        "完成.*任务",
        "调用.*工具",
    }
    for _, pattern := range agentPatterns {
        if matched, _ := regexp.MatchString(pattern, message); matched {
            return "agent"
        }
    }

    // RAG 模式关键词检测
    ragPatterns := []string{
        "什么是",
        "如何.*",
        "介绍.*",
        "文档中.*",
        ".*的定义",
        "解释.*",
    }
    for _, pattern := range ragPatterns {
        if matched, _ := regexp.MatchString(pattern, message); matched {
            return "rag"
        }
    }

    // 默认使用 RAG（知识问答场景）
    return "rag"
}

用户消息输入
    ↓
显式指定 mode？
    ├─ 是 → 使用指定模式
    └─ 否 → 自动检测
        ↓
包含 Agent 关键词？（帮我、执行、调用工具）
    ├─ 是 → Agent 模式
    └─ 否 → 继续
        ↓
包含 RAG 关键词？（什么是、如何、介绍）
    ├─ 是 → RAG 模式
    └─ 否 → 默认 RAG 模式

sequenceDiagram
    autonumber
    participant Client
    participant HTTP as HTTPServer
    participant Pipeline as 处理管道
    participant Usecase as OrchestratorUsecase
    participant Engine as AI Engine
    participant DB as PostgreSQL
    participant Cache as Redis

    Client->>HTTP: POST /api/v1/orchestrator/process
    HTTP->>Pipeline: Handle(request)

    Note over Pipeline: 日志、验证、指标

    Pipeline->>Usecase: ExecuteChat(req)
    Usecase->>DB: CreateTask()
    DB-->>Usecase: task_id

    Usecase->>Usecase: selectMode(message)

    alt mode == "rag"
        Usecase->>Engine: RAG.Generate()
        Engine-->>Usecase: result
    else mode == "agent"
        Usecase->>Engine: Agent.Execute()
        Engine-->>Usecase: result
    end

    Usecase->>DB: UpdateTask(status, output)
    Usecase->>Cache: CacheTask(task)

    Usecase-->>Pipeline: response
    Pipeline-->>HTTP: response
    HTTP-->>Client: 200 OK + SSE Stream

type GetTaskResponse struct {
    Task         *Task          `json:"task"`
    Status       string         `json:"status"`
    Progress     float64        `json:"progress"`      // 0.0-1.0
    CurrentStep  *TaskStep      `json:"current_step"`
    ElapsedTime  float64        `json:"elapsed_time"`
    EstimatedTime float64       `json:"estimated_time"`
}

func (uc *OrchestratorUsecase) GetTask(
    ctx context.Context,
    taskID string,
) (*GetTaskResponse, error) {
    // 1. 从缓存获取
    if cached, err := uc.cache.Get(ctx, "task:"+taskID); err == nil {
        return cached.(*GetTaskResponse), nil
    }

    // 2. 从数据库获取
    task, err := uc.taskRepo.GetByID(ctx, taskID)
    if err != nil {
        return nil, fmt.Errorf("task not found: %w", err)
    }

    // 3. 构建响应
    response := &GetTaskResponse{
        Task:         task,
        Status:       string(task.Status),
        Progress:     calculateProgress(task),
        CurrentStep:  getCurrentStep(task),
        ElapsedTime:  task.Duration().Seconds(),
        EstimatedTime: estimateRemainingTime(task),
    }

    // 4. 缓存结果
    if task.IsCompleted() {
        _ = uc.cache.Set(ctx, "task:"+taskID, response, 1*time.Hour)
    }

    return response, nil
}

# 服务配置
HOST=0.0.0.0
PORT=8000
GRPC_PORT=9000
WORKERS=4

# AI 引擎地址
AGENT_ENGINE_URL=http://agent-engine:8003
RAG_ENGINE_URL=http://rag-engine:8001
VOICE_ENGINE_URL=http://voice-engine:8006
MULTIMODAL_ENGINE_URL=http://multimodal-engine:8007

# 任务配置
MAX_CONCURRENT_TASKS=100
TASK_TIMEOUT=300s
TASK_RETRY_TIMES=3

# 数据库配置
DB_HOST=localhost
DB_PORT=5432
DB_NAME=voiceassistant

# Redis 配置
REDIS_ADDR=localhost:6379
REDIS_DB=0

service:
  name: ai-orchestrator
  version: 1.0.0

server:
  host: 0.0.0.0
  http_port: 8000
  grpc_port: 9000
  workers: 4

engines:
  agent:
    url: http://agent-engine:8003
    timeout: 60s
  rag:
    url: http://rag-engine:8001
    timeout: 30s
  voice:
    url: http://voice-engine:8006
    timeout: 30s
  multimodal:
    url: http://multimodal-engine:8007
    timeout: 60s

routing:
  default_mode: auto
  intent_detection_enabled: true

task_manager:
  max_concurrent_tasks: 100
  task_timeout: 300s
  retry_times: 3
  priority_queue_enabled: true

circuit_breaker:
  failure_threshold: 5
  timeout: 30s
  reset_timeout: 60s

// HTTP Server 中间件配置
var opts = []http.ServerOption{
    http.Middleware(
        recovery.Recovery(),      // 1. 捕获 panic
        tracing.Server(),         // 2. 分布式追踪
        logging.Server(logger),   // 3. 请求日志
    ),
}

// gRPC Server 中间件配置
var opts = []grpc.ServerOption{
    grpc.Middleware(
        recovery.Recovery(),      // 1. 捕获 panic
        tracing.Server(),         // 2. 分布式追踪
        logging.Server(logger),   // 3. 请求日志
        validate.Validator(),     // 4. 参数验证（仅 gRPC）
    ),
}

// 熔断器配置（每个外部服务独立熔断器）
settings := gobreaker.Settings{
    Name:        service,                  // 服务名称
    MaxRequests: 3,                        // 半开状态最大请求数
    Interval:    10 * time.Second,         // 统计周期
    Timeout:     60 * time.Second,         // 开启状态持续时间
    ReadyToTrip: func(counts gobreaker.Counts) bool {
        // 触发条件：失败率 >= 60% 且请求数 >= 3
        failureRatio := float64(counts.TotalFailures) / float64(counts.Requests)
        return counts.Requests >= 3 && failureRatio >= 0.6
    },
    OnStateChange: func(name string, from gobreaker.State, to gobreaker.State) {
        log.Infof("circuit breaker state changed: service=%s, from=%s, to=%s", name, from, to)
    },
}

// 使用熔断器
result, err := cb.Execute(func() (interface{}, error) {
    return c.callWithRetry(service, method, input)  // 调用实际服务
})

func (c *GRPCServiceClient) callWithRetry(service, method string, input map[string]interface{}) (map[string]interface{}, error) {
    var lastErr error

    for attempt := 1; attempt <= c.maxRetries; attempt++ {  // maxRetries = 3
        ctx, cancel := context.WithTimeout(context.Background(), c.timeout)  // timeout = 30s
        defer cancel()

        result, err := c.doCall(ctx, service, method, input)
        if err == nil {
            return result, nil  // 成功，立即返回
        }

        lastErr = err
        if attempt < c.maxRetries {
            backoff := time.Duration(attempt) * 100 * time.Millisecond  // 100ms, 200ms, 300ms
            time.Sleep(backoff)
        }
    }

    return nil, fmt.Errorf("service call failed after %d attempts: %w", c.maxRetries, lastErr)
}

type GRPCServiceClient struct {
    connections map[string]*grpc.ClientConn  // 连接池
}

func (c *GRPCServiceClient) getConnection(service string) (*grpc.ClientConn, error) {
    if conn, exists := c.connections[service]; exists {
        return conn, nil  // 复用现有连接
    }

    // 创建新连接
    conn, err := grpc.Dial(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
    c.connections[service] = conn
    return conn, nil
}

// 5 类核心指标
TaskCreatedTotal = promauto.NewCounterVec(...)           // 任务创建计数
TaskExecutionDuration = promauto.NewHistogramVec(...)    // 执行时长分布
TaskStatusTotal = promauto.NewCounterVec(...)            // 状态计数
TaskTokensUsed = promauto.NewHistogramVec(...)           // Token 使用量
TaskCostUSD = promauto.NewHistogramVec(...)              // 成本分布

// 采集示例
TaskCreatedTotal.WithLabelValues("rag", tenant_id).Inc()
TaskExecutionDuration.WithLabelValues("rag", "completed", tenant_id).Observe(2.5)

type TaskPO struct {
    ID      string `gorm:"primaryKey"`
    Input   string `gorm:"type:jsonb"`  // JSON 序列化
    Output  string `gorm:"type:jsonb"`
    Steps   string `gorm:"type:jsonb"`
    Metadata string `gorm:"type:jsonb"`
}

// 序列化
inputJSON, _ := json.Marshal(task.Input)
po.Input = string(inputJSON)

// 反序列化
var input domain.TaskInput
json.Unmarshal([]byte(po.Input), &input)

// Pipeline 接口
type Pipeline interface {
    Execute(task *Task) (*TaskOutput, error)
    Name() string
}

// 三种实现
type RAGPipeline struct { ... }
type AgentPipeline struct { ... }
type VoicePipeline struct { ... }

// 映射表
pipelines := map[domain.TaskType]domain.Pipeline{
    domain.TaskTypeRAG:   ragPipeline,
    domain.TaskTypeAgent: agentPipeline,
    domain.TaskTypeVoice: voicePipeline,
}

// Kratos 自动创建 Span
http.Middleware(tracing.Server())

// 自定义 Span（可选）
ctx, span := otel.Tracer("orchestrator").Start(ctx, "ExecuteTask")
defer span.End()
span.SetAttributes(
    attribute.String("task_id", taskID),
    attribute.String("task_type", taskType),
)

// wire.go
func wireApp(*Config, log.Logger) (*kratos.App, func(), error) {
    panic(wire.Build(
        data.NewDB,
        data.NewData,
        data.NewTaskRepo,
        data.NewServiceClient,
        biz.NewTaskUsecase,
        service.NewOrchestratorService,
        server.NewGRPCServer,
        server.NewHTTPServer,
        newApp,
    ))
}

type TaskStatus string

const (
    TaskStatusPending   TaskStatus = "pending"
    TaskStatusRunning   TaskStatus = "running"
    TaskStatusCompleted TaskStatus = "completed"
    TaskStatusFailed    TaskStatus = "failed"
    TaskStatusCancelled TaskStatus = "cancelled"
)

// 状态转换
task.Start()      // Pending → Running
task.Complete()   // Running → Completed
task.Fail()       // Running → Failed
task.Cancel()     // Pending/Running → Cancelled