AI赋能业务工作流：极简架构下的落地策略与工程实践

星辰AI

1262人浏览 · 2026-06-17 13:08:14

星辰AI · 2026-06-17 13:08:14 发布

AI赋能业务工作流：极简架构下的落地策略与工程实践

cover

一、AI落地的两难：技术先进性与工程简洁性的冲突

AI能力接入业务系统时，架构师面临一个核心矛盾：技术团队倾向于引入最先进的模型和框架，而业务团队只关心"能不能用、稳不稳定、贵不贵"。一个RAG系统，用LangChain+Chroma+OpenAI可以快速搭出原型，但生产部署时发现依赖链太长、调试困难、成本不可控。另一个极端是直接调API拼Prompt，简单粗暴但缺乏可维护性。

真正的挑战在于：如何在保持架构极简的前提下，将AI能力稳定地融入现有业务工作流？极简不是偷工减料，而是用最少的组件实现最核心的能力，同时为未来的扩展留出接口。

二、AI工作流集成的极简架构：三层解耦模型

将AI能力集成到业务系统，最有效的架构是三层解耦：接入层、编排层、执行层。每层只做一件事，层与层之间通过明确的接口通信。

graph TB
    subgraph 接入层
        A1[REST API] --> B[AI网关]
        A2[消息队列] --> B
        A3[定时任务] --> B
    end

    subgraph 编排层
        B --> C[Prompt模板引擎]
        C --> D[上下文管理器]
        D --> E[结果校验器]
    end

    subgraph 执行层
        E --> F1[OpenAI]
        E --> F2[Claude]
        E --> F3[本地模型]
    end

    subgraph 可观测性
        G[Token计量] -.-> B
        H[延迟监控] -.-> E
        I[质量评分] -.-> E
    end

    style B fill:#722ed1,color:#fff
    style C fill:#1890ff,color:#fff
    style D fill:#faad14,color:#fff
    style E fill:#52c41a,color:#fff

接入层统一所有AI请求的入口，无论来自API调用、消息消费还是定时触发，都经过AI网关进行鉴权、限流和路由。编排层是核心，负责Prompt模板渲染、上下文窗口管理和输出校验。执行层是多模型适配层，屏蔽不同LLM提供商的API差异。

这种架构的关键在于：业务代码不直接调用LLM API，而是通过AI网关的统一接口。LLM提供商的切换、Prompt的优化、上下文策略的调整，都在编排层完成，业务层无感知。

三、极简AI网关的Go实现

以下是一个生产可用的AI网关核心实现，强调模型切换、Prompt模板化和结果校验：

package gateway

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"sync"
	"text/template"
	"time"
)

// LLMProvider 定义模型提供商接口
type LLMProvider interface {
	Name() string
	Chat(ctx context.Context, req *ChatRequest) (*ChatResponse, error)
	EstimateTokens(text string) int
}

// ChatRequest 统一的聊天请求
type ChatRequest struct {
	Messages []Message `json:"messages"`
	Model    string    `json:"model"`
	// 最大生成Token数
	MaxTokens int `json:"max_tokens,omitempty"`
	// 温度参数
	Temperature float64 `json:"temperature,omitempty"`
}

// ChatResponse 统一的聊天响应
type ChatResponse struct {
	Content     string `json:"content"`
	Model       string `json:"model"`
	TokensUsed  int    `json:"tokens_used"`
	LatencyMs   int64  `json:"latency_ms"`
	FinishReason string `json:"finish_reason"`
}

// Message 对话消息
type Message struct {
	Role    string `json:"role"`
	Content string `json:"content"`
}

// PromptTemplate Prompt模板定义
type PromptTemplate struct {
	Name     string
	Template string
	// 模板参数的JSON Schema，用于校验输入
	ParamSchema json.RawMessage
}

// AIGateway AI网关
type AIGateway struct {
	providers map[string]LLMProvider
	templates map[string]*PromptTemplate
	// 默认模型提供商
	defaultProvider string
	mu              sync.RWMutex
	// Token计量
	tokenCounter *TokenCounter
}

// TokenCounter Token消耗计量器
type TokenCounter struct {
	dailyUsage map[string]int // key: date+provider, value: token count
	mu         sync.Mutex
}

// NewAIGateway 创建AI网关
func NewAIGateway(defaultProvider string) *AIGateway {
	return &AIGateway{
		providers:      make(map[string]LLMProvider),
		templates:      make(map[string]*PromptTemplate),
		defaultProvider: defaultProvider,
		tokenCounter:   &TokenCounter{dailyUsage: make(map[string]int)},
	}
}

// RegisterProvider 注册模型提供商
func (g *AIGateway) RegisterProvider(provider LLMProvider) {
	g.mu.Lock()
	defer g.mu.Unlock()
	g.providers[provider.Name()] = provider
}

// RegisterTemplate 注册Prompt模板
func (g *AIGateway) RegisterTemplate(tpl *PromptTemplate) {
	g.mu.Lock()
	defer g.mu.Unlock()
	g.templates[tpl.Name()] = tpl
}

// Execute 执行AI请求，支持模板渲染和模型切换
func (g *AIGateway) Execute(ctx context.Context, req *ExecuteRequest) (*ChatResponse, error) {
	// 1. 渲染Prompt模板
	messages, err := g.renderTemplate(req.TemplateName, req.TemplateParams)
	if err != nil {
		return nil, fmt.Errorf("模板渲染失败: %w", err)
	}

	// 2. 注入上下文（如果有）
	if req.Context != "" {
		messages = g.injectContext(messages, req.Context)
	}

	// 3. 选择模型提供商
	providerName := req.Provider
	if providerName == "" {
		providerName = g.defaultProvider
	}

	g.mu.RLock()
	provider, exists := g.providers[providerName]
	g.mu.RUnlock()

	if !exists {
		return nil, fmt.Errorf("未注册的模型提供商: %s", providerName)
	}

	// 4. Token预算检查
	estimatedTokens := 0
	for _, msg := range messages {
		estimatedTokens += provider.EstimateTokens(msg.Content)
	}
	if req.MaxTokens > 0 && estimatedTokens > req.MaxTokens {
		// 截断上下文以适应Token预算
		messages = g.truncateMessages(messages, provider, req.MaxTokens)
	}

	// 5. 调用模型
	startTime := time.Now()
	chatReq := &ChatRequest{
		Messages:    messages,
		Model:       req.Model,
		MaxTokens:   req.MaxOutputTokens,
		Temperature: req.Temperature,
	}

	resp, err := provider.Chat(ctx, chatReq)
	if err != nil {
		// 主模型失败，尝试降级到备用模型
		if req.FallbackProvider != "" {
			return g.executeFallback(ctx, req.FallbackProvider, chatReq)
		}
		return nil, fmt.Errorf("模型调用失败: %w", err)
	}
	resp.LatencyMs = time.Since(startTime).Milliseconds()

	// 6. 结果校验
	if req.Validator != nil {
		if err := req.Validator(resp.Content); err != nil {
			return nil, fmt.Errorf("结果校验失败: %w", err)
		}
	}

	// 7. 记录Token消耗
	g.tokenCounter.Record(providerName, resp.TokensUsed)

	return resp, nil
}

// renderTemplate 渲染Prompt模板
func (g *AIGateway) renderTemplate(name string, params map[string]any) ([]Message, error) {
	g.mu.RLock()
	tpl, exists := g.templates[name]
	g.mu.RUnlock()

	if !exists {
		return nil, fmt.Errorf("模板不存在: %s", name)
	}

	tmpl, err := template.New(name).Parse(tpl.Template)
	if err != nil {
		return nil, fmt.Errorf("模板解析失败: %w", err)
	}

	var buf strings.Builder
	if err := tmpl.Execute(&buf, params); err != nil {
		return nil, fmt.Errorf("模板执行失败: %w", err)
	}

	// 解析渲染后的内容为消息列表
	var messages []Message
	if err := json.Unmarshal([]byte(buf.String()), &messages); err != nil {
		// 如果不是JSON格式，作为单条用户消息处理
		messages = []Message{{Role: "user", Content: buf.String()}}
	}

	return messages, nil
}

// injectContext 将业务上下文注入到消息列表
func (g *AIGateway) injectContext(messages []Message, ctx string) []Message {
	contextMsg := Message{
		Role:    "system",
		Content: fmt.Sprintf("以下是相关的业务上下文信息：\n%s", ctx),
	}
	// 插入到系统消息之后、用户消息之前
	result := make([]Message, 0, len(messages)+1)
	inserted := false
	for _, msg := range messages {
		result = append(result, msg)
		if msg.Role == "system" && !inserted {
			result = append(result, contextMsg)
			inserted = true
		}
	}
	if !inserted {
		result = append([]Message{contextMsg}, result...)
	}
	return result
}

// truncateMessages 按Token预算截断消息
func (g *AIGateway) truncateMessages(
	messages []Message, provider LLMProvider, budget int,
) []Message {
	totalTokens := 0
	result := make([]Message, 0, len(messages))
	// 从最早的消息开始丢弃，保留最近的
	for i := len(messages) - 1; i >= 0; i-- {
		tokens := provider.EstimateTokens(messages[i].Content)
		if totalTokens+tokens > budget {
			break
		}
		totalTokens += tokens
		result = append([]Message{messages[i]}, result...)
	}
	return result
}

// executeFallback 降级到备用模型
func (g *AIGateway) executeFallback(
	ctx context.Context, providerName string, req *ChatRequest,
) (*ChatResponse, error) {
	g.mu.RLock()
	provider, exists := g.providers[providerName]
	g.mu.RUnlock()

	if !exists {
		return nil, fmt.Errorf("备用模型未注册: %s", providerName)
	}
	return provider.Chat(ctx, req)
}

// ExecuteRequest 执行请求参数
type ExecuteRequest struct {
	TemplateName    string         // Prompt模板名称
	TemplateParams  map[string]any // 模板参数
	Context         string         // 业务上下文
	Provider        string         // 模型提供商
	Model           string         // 具体模型
	FallbackProvider string        // 降级模型提供商
	MaxTokens       int            // Token预算上限
	MaxOutputTokens int            // 最大输出Token数
	Temperature     float64        // 温度参数
	Validator       func(string) error // 结果校验函数
}

// Record 记录Token消耗
func (tc *TokenCounter) Record(provider string, tokens int) {
	tc.mu.Lock()
	defer tc.mu.Unlock()
	key := fmt.Sprintf("%s:%s", time.Now().Format("2006-01-02"), provider)
	tc.dailyUsage[key] += tokens
}

四、极简架构的边界：何时需要更重的方案

三层解耦模型在中小规模场景下非常有效，但当系统规模增长到一定程度时，它的局限性会暴露出来。

缺乏流式输出支持。当前实现只支持同步请求-响应模式，不支持SSE流式输出。对于长文本生成场景（如文章撰写、代码生成），用户需要等待整个响应完成才能看到结果，体验很差。要支持流式输出，需要将ChatResponse改为channel模式，这会增加编排层的复杂度。

不支持多轮对话的持久化。上下文管理器只维护单次请求的上下文，不负责跨请求的对话历史持久化。如果你的业务需要多轮对话（如客服机器人），需要在外部维护对话session，并在每次请求时传入完整历史。

Prompt版本管理缺失。模板以代码形式硬编码，没有版本管理和A/B测试能力。在生产环境中，Prompt的调优是持续进行的，你需要能够灰度发布新Prompt、对比不同版本的效果。这需要一个独立的Prompt管理服务。

禁用场景：需要实时流式输出的对话产品——当前架构不支持SSE；需要严格合规审核的AI输出（如金融、医疗）——校验器过于简单，无法满足合规要求；多租户场景下需要Prompt隔离和用量配额——当前网关没有租户概念。

五、总结

AI赋能业务工作流的核心策略是"三层解耦"：接入层统一入口、编排层管理Prompt和上下文、执行层屏蔽模型差异。这种极简架构的优势在于：业务代码与AI实现完全解耦，模型切换和Prompt优化可以在不影响业务的情况下进行。Go实现的AI网关提供了模板渲染、Token预算控制、降级切换和结果校验等生产级能力。但极简架构有其边界——当需要流式输出、多轮对话持久化或Prompt版本管理时，就需要引入更重的组件。架构的演进应该是渐进的，从极简开始，按需加码。