在大型AI应用中,如何用Zustand或Redux Toolkit管理多轮对话、生成任务、用户配置等复杂状态?

场景 :"印客学院"的AI学习平台包含多个复杂功能:多轮AI对话、异步生成任务、用户个性化配置等,需要统一的状态管理方案。

Zustand方案 :Zustand适合中小型应用,以更简洁的方式管理状态。

// store/aiStore.ts
import { create } from 'zustand';

interface AIMessage {
  id: string;
  role: 'user' | 'assistant' | 'system';
  content: string;
  status: 'pending' | 'streaming' | 'completed' | 'error';
}

interface AIStore {
  // 对话状态
  conversations: Map<string, AIMessage[]>;
  activeConversationId: string | null;
  
  // 生成任务队列
  generationTasks: Array<{
    id: string;
    type: 'code' | 'text' | 'image';
    status: 'pending' | 'processing' | 'completed';
    progress: number;
  }>;
  
  // 用户配置
  userConfig: {
    model: string;
    temperature: number;
    maxTokens: number;
  };
  
  // 状态更新方法
  addMessage: (conversationId: string, message: AIMessage) => void;
  updateMessageStatus: (messageId: string, status: AIMessage['status']) => void;
  updateConfig: (config: Partial<AIStore['userConfig']>) => void;
}

export const useAIStore = create<AIStore>((set, get) => ({
  conversations: new Map(),
  activeConversationId: null,
  generationTasks: [],
  userConfig: { model: 'gpt-4', temperature: 0.7, maxTokens: 2000 },
  
  addMessage: (conversationId, message) => {
    set((state) => {
      const conversations = new Map(state.conversations);
      const messages = conversations.get(conversationId) || [];
      conversations.set(conversationId, [...messages, message]);
      return { conversations };
    });
  },
  
  updateMessageStatus: (messageId, status) => {
    set((state) => {
      const conversations = new Map(state.conversations);
      for (const [cid, messages] of conversations) {
        const updatedMessages = messages.map(msg =>
          msg.id === messageId ? { ...msg, status } : msg
        );
        conversations.set(cid, updatedMessages);
      }
      return { conversations };
    });
  },
  
  updateConfig: (config) => {
    set((state) => ({
      userConfig: { ...state.userConfig, ...config }
    }));
  }
}));

Redux Toolkit方案 :更适合大型企业级应用,提供更严格的架构约束。

// features/ai/aiSlice.ts
import { createSlice, createAsyncThunk, PayloadAction } from '@reduxjs/toolkit';

interface AIState {
  conversations: Record<string, Conversation>;
  generationQueue: GenerationTask[];
  config: AIConfig;
  status: 'idle' | 'loading' | 'failed';
}

const initialState: AIState = {
  conversations: {},
  generationQueue: [],
  config: { model: 'gpt-4', temperature: 0.7 },
  status: 'idle'
};

export const aiSlice = createSlice({
  name: 'ai',
  initialState,
  reducers: {
    addMessage: (state, action: PayloadAction<AddMessagePayload>) => {
      const { conversationId, message } = action.payload;
      if (!state.conversations[conversationId]) {
        state.conversations[conversationId] = { id: conversationId, messages: [] };
      }
      state.conversations[conversationId].messages.push(message);
    },
    updateConfig: (state, action: PayloadAction<Partial<AIConfig>>) => {
      state.config = { ...state.config, ...action.payload };
    }
  },
  extraReducers: (builder) => {
    builder
      .addCase(startGeneration.pending, (state) => {
        state.status = 'loading';
      })
      .addCase(startGeneration.fulfilled, (state, action) => {
        state.generationQueue.push(action.payload);
        state.status = 'idle';
      });
  }
});

export const startGeneration = createAsyncThunk(
  'ai/startGeneration',
  async (payload: GenerationRequest) => {
    const response = await fetch('/api/inke/ai/generate', {
      method: 'POST',
      body: JSON.stringify(payload)
    });
    return response.json();
  }
);

选择建议

  • Zustand :适合中小型项目,API简洁,学习成本低
  • Redux Toolkit :适合大型企业应用,需要强类型、中间件、DevTools等高级功能
  • "印客学院"实践 :采用Redux Toolkit管理核心业务状态,Zustand管理UI状态

设计一个"状态快照"系统,支持将AI对话的完整状态(包括流式中间结果)序列化保存与恢复

场景 :在"印客学院"的AI对话编辑器中,用户需要保存对话的完整快照,包括正在流式生成的内容,以便后续恢复。

核心设计

interface AISnapshot {
  id: string;
  timestamp: number;
  metadata: {
    conversationId: string;
    userId: string;
    model: string;
  };
  // 完整对话状态
  conversation: {
    messages: Array<{
      id: string;
      role: string;
      content: string;
      streamingContent?: string; // 流式生成的中间内容
      status: 'complete' | 'streaming' | 'pending';
    }>;
  };
  // 应用状态
  uiState: {
    activeMessageId?: string;
    inputText: string;
    config: Record<string, any>;
  };
  // 版本信息
  version: number;
  checksum: string;
}

class InkeSnapshotManager {
  // 创建快照
  async createSnapshot(conversationId: string): Promise<AISnapshot> {
    const state = this.getCurrentState();
    const snapshot: AISnapshot = {
      id: `snap_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`,
      timestamp: Date.now(),
      metadata: {
        conversationId,
        userId: 'user_123',
        model: 'inke-tutor'
      },
      conversation: {
        messages: state.conversation.messages.map(msg => ({
          ...msg,
          // 对流式内容进行特殊处理
          streamingContent: msg.status === 'streaming' ? msg.content : undefined
        }))
      },
      uiState: state.ui,
      version: 2,
      checksum: this.calculateChecksum(state)
    };
    
    // 保存到IndexedDB
    await this.saveToIndexedDB(snapshot);
    
    // 可选:上传到服务器
    await this.uploadToCloud(snapshot);
    
    return snapshot;
  }
  
  // 恢复快照
  async restoreSnapshot(snapshotId: string): Promise<void> {
    const snapshot = await this.loadSnapshot(snapshotId);
    
    // 验证版本兼容性
    if (!this.isVersionCompatible(snapshot.version)) {
      throw new Error('Snapshot version incompatible');
    }
    
    // 恢复对话状态
    this.restoreConversation(snapshot.conversation);
    
    // 恢复UI状态
    this.restoreUIState(snapshot.uiState);
    
    // 如果是流式生成中的消息,继续生成
    await this.resumeStreamingMessages(snapshot.conversation.messages);
  }
  
  // 继续流式生成
  private async resumeStreamingMessages(messages: AISnapshot['conversation']['messages']) {
    const streamingMessages = messages.filter(m => m.status === 'streaming');
    
    for (const msg of streamingMessages) {
      // 获取消息的流式上下文
      const context = this.getMessageContext(msg.id);
      
      // 重新发起流式请求
      await this.continueStreamGeneration({
        messageId: msg.id,
        partialContent: msg.streamingContent || '',
        context
      });
    }
  }
  
  // 增量快照(只保存变化部分)
  createIncrementalSnapshot(previousSnapshotId: string): IncrementalSnapshot {
    const current = this.getCurrentState();
    const previous = this.loadSnapshot(previousSnapshotId);
    
    // 计算差异
    const diff = this.calculateStateDiff(previous, current);
    
    return {
      baseSnapshotId: previousSnapshotId,
      diff,
      timestamp: Date.now()
    };
  }
}

关键特性

  1. 完整序列化 :支持JSON序列化,包含特殊类型处理
  2. 版本控制 :快照包含版本号,支持向后兼容
  3. 流式状态保存 :特别处理流式生成中的中间状态
  4. 恢复机制 :支持从任意快照点恢复
  5. 增量快照 :只保存状态差异,节省存储空间
  6. 完整性验证 :通过checksum确保数据完整性

存储策略

  • 本地:IndexedDB + 压缩
  • 云端:分块上传 + 版本管理
  • 清理策略:LRU + 手动清理

如何用XState或状态图建模AI Agent的完整工作流(包括工具调用、条件分支、错误处理)?

场景 :"印客学院"的AI智能助教需要处理复杂的工作流:接收问题 → 分析意图 → 调用工具 → 处理结果 → 生成回答,包含多种分支和错误处理。

XState实现

import { createMachine, assign } from 'xstate';

const aiAgentMachine = createMachine({
  id: 'aiAgent',
  initial: 'idle',
  context: {
    question: '',
    tools: [],
    result: null,
    error: null,
    retryCount: 0
  },
  states: {
    idle: {
      on: {
        ASK_QUESTION: {
          target: 'analyzing',
          actions: assign({
            question: (_, event) => event.question
          })
        }
      }
    },
    
    analyzing: {
      invoke: {
        src: 'analyzeQuestion',
        onDone: {
          target: 'selectingTool',
          actions: assign({
            tools: (_, event) => event.data.tools
          })
        },
        onError: {
          target: 'error',
          actions: assign({
            error: (_, event) => event.data
          })
        }
      }
    },
    
    selectingTool: {
      always: [
        {
          target: 'callingTool',
          cond: (ctx) => ctx.tools.length === 1
        },
        {
          target: 'toolSelection',
          cond: (ctx) => ctx.tools.length > 1
        },
        {
          target: 'generatingAnswer',
          cond: (ctx) => ctx.tools.length === 0
        }
      ]
    },
    
    toolSelection: {
      on: {
        TOOL_SELECTED: {
          target: 'callingTool',
          actions: assign({
            selectedTool: (_, event) => event.tool
          })
        }
      }
    },
    
    callingTool: {
      invoke: {
        src: 'callTool',
        onDone: {
          target: 'processingResult',
          actions: assign({
            result: (_, event) => event.data
          })
        },
        onError: [
          {
            target: 'retrying',
            cond: (ctx) => ctx.retryCount < 3
          },
          {
            target: 'error',
            actions: assign({
              error: (_, event) => event.data
            })
          }
        ]
      }
    },
    
    retrying: {
      entry: assign({
        retryCount: (ctx) => ctx.retryCount + 1
      }),
      after: {
        1000: 'callingTool' // 1秒后重试
      }
    },
    
    processingResult: {
      invoke: {
        src: 'processResult',
        onDone: {
          target: 'generatingAnswer',
          actions: assign({
            processedResult: (_, event) => event.data
          })
        }
      }
    },
    
    generatingAnswer: {
      invoke: {
        src: 'generateAnswer',
        onDone: {
          target: 'completed',
          actions: assign({
            answer: (_, event) => event.data
          })
        }
      }
    },
    
    completed: {
      type: 'final',
      data: (ctx) => ({
        answer: ctx.answer,
        toolsUsed: ctx.tools
      })
    },
    
    error: {
      on: {
        RETRY: {
          target: 'analyzing',
          actions: assign({
            error: null,
            retryCount: 0
          })
        }
      }
    }
  }
}, {
  services: {
    analyzeQuestion: async (ctx) => {
      // 调用印客学院的意图分析服务
      return fetch('/api/inke/analyze', {
        method: 'POST',
        body: JSON.stringify({ question: ctx.question })
      }).then(r => r.json());
    },
    
    callTool: async (ctx) => {
      const tool = ctx.selectedTool || ctx.tools[0];
      // 调用工具
      return fetch(`/api/inke/tools/${tool.name}`, {
        method: 'POST',
        body: JSON.stringify(tool.parameters)
      }).then(r => r.json());
    },
    
    generateAnswer: async (ctx) => {
      // 生成最终回答
      return fetch('/api/inke/generate', {
        method: 'POST',
        body: JSON.stringify({
          question: ctx.question,
          result: ctx.processedResult
        })
      }).then(r => r.json());
    }
  }
});

状态图优势

  1. 可视化调试 :可用XState Viz工具查看状态流转
  2. 类型安全 :TypeScript强类型支持
  3. 可测试性 :每个状态可独立测试
  4. 持久化 :支持状态序列化/反序列化
  5. 错误恢复 :内置重试和错误处理机制

"印客学院"实践

  • 使用XState管理核心AI工作流
  • 将复杂流程分解为可重用的子状态机
  • 集成Redux进行状态同步
  • 持久化关键状态到本地存储

在微前端架构下,多个AI功能模块需要共享"当前模型版本"状态,如何设计跨应用状态同步方案?

场景 :"印客学院"采用微前端架构,包含聊天、代码生成、文档分析等多个子应用,需要共享AI模型版本等全局状态。

方案一:中心化状态管理 (推荐)

// shared/store/modelStore.ts - 主应用管理
import { createStore } from 'zustand/vanilla';

interface ModelState {
  currentModel: string;
  modelVersion: string;
  availableModels: string[];
  lastUpdated: number;
}

export const modelStore = createStore<ModelState>(() => ({
  currentModel: 'gpt-4',
  modelVersion: 'v2.0',
  availableModels: ['gpt-4', 'claude-3', 'inke-tutor'],
  lastUpdated: Date.now()
}));

// 子应用通过CustomEvent监听状态变化
export function subscribeToModelUpdates(callback: (state: ModelState) => void) {
  return modelStore.subscribe(callback);
}

// 主应用广播状态变化
window.dispatchEvent(new CustomEvent('model-updated', {
  detail: modelStore.getState()
}));

方案二:基于window对象的共享状态

// shared/globalState.ts
class GlobalStateManager {
  private state: Record<string, any> = {};
  private listeners: Map<string, Set<Function>> = new Map();
  
  set(key: string, value: any) {
    const oldValue = this.state[key];
    this.state[key] = value;
    
    // 通知所有监听者
    if (this.listeners.has(key)) {
      this.listeners.get(key)!.forEach(listener => {
        listener(value, oldValue);
      });
    }
    
    // 持久化到localStorage
    this.persistState();
  }
  
  get(key: string) {
    return this.state[key];
  }
  
  subscribe(key: string, callback: Function) {
    if (!this.listeners.has(key)) {
      this.listeners.set(key, new Set());
    }
    this.listeners.get(key)!.add(callback);
    
    return () => this.unsubscribe(key, callback);
  }
  
  private persistState() {
    try {
      localStorage.setItem('inke-global-state', JSON.stringify(this.state));
    } catch (e) {
      console.warn('Failed to persist global state:', e);
    }
  }
  
  loadState() {
    try {
      const saved = localStorage.getItem('inke-global-state');
      if (saved) {
        this.state = JSON.parse(saved);
      }
    } catch (e) {
      console.warn('Failed to load global state:', e);
    }
  }
}

// 挂载到window对象
if (!window.__INKE_GLOBAL_STATE__) {
  window.__INKE_GLOBAL_STATE__ = new GlobalStateManager();
  window.__INKE_GLOBAL_STATE__.loadState();
}

// 子应用使用
const globalState = window.__INKE_GLOBAL_STATE__;

// 设置模型版本
globalState.set('modelVersion', 'v2.1');

// 监听模型版本变化
const unsubscribe = globalState.subscribe('modelVersion', (newVal, oldVal) => {
  console.log(`Model version changed: ${oldVal} -> ${newVal}`);
});

方案三:基于PostMessage的跨应用通信

// shared/eventBus.ts
class MicroFrontendEventBus {
  private targetOrigin = window.location.origin;
  
  // 发送状态更新
  publish(event: string, data: any) {
    window.parent.postMessage({
      type: 'MF_STATE_UPDATE',
      event,
      data,
      source: 'inke-ai-module',
      timestamp: Date.now()
    }, this.targetOrigin);
  }
  
  // 订阅状态更新
  subscribe(event: string, callback: (data: any) => void) {
    const handler = (e: MessageEvent) => {
      if (e.data.type === 'MF_STATE_UPDATE' && e.data.event === event) {
        callback(e.data.data);
      }
    };
    
    window.addEventListener('message', handler);
    return () => window.removeEventListener('message', handler);
  }
}

// 主应用统一协调
window.addEventListener('message', (e) => {
  if (e.data.type === 'MF_STATE_UPDATE') {
    // 验证来源
    if (this.isTrustedSource(e.data.source)) {
      // 更新全局状态
      this.updateGlobalState(e.data.event, e.data.data);
      
      // 广播给其他子应用
      this.broadcastToOtherApps(e.data);
    }
  }
});

最佳实践组合

  1. 核心配置 :通过主应用的中心化Store管理
  2. 运行时状态 :通过window对象共享
  3. 事件通信 :通过PostMessage进行跨应用通知
  4. 持久化 :关键状态保存到localStorage
  5. 版本协商 :子应用声明支持的模型版本,主应用进行兼容性检查

请设计一个"乐观更新"策略,在用户发送AI请求后立即在UI中显示预期结果,再根据实际流式响应逐步修正

场景 :在"印客学院"的AI聊天界面中,用户发送消息后,希望立即看到AI正在思考的提示,而不是等待网络请求完成。

实现方案

class OptimisticUpdateManager {
  private pendingUpdates = new Map<string, {
    optimisticData: any;
    timestamp: number;
    retryCount: number;
  }>();
  
  // 发送消息时的乐观更新
  async sendMessageWithOptimisticUpdate(
    conversationId: string,
    userMessage: string
  ): Promise<string> {
    const messageId = `msg_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
    
    // 1. 立即在UI中添加"AI正在思考"的消息
    this.addOptimisticMessage(conversationId, {
      id: messageId,
      role: 'assistant',
      content: '正在思考...',
      status: 'thinking',
      isOptimistic: true
    });
    
    // 2. 记录乐观更新状态
    this.pendingUpdates.set(messageId, {
      optimisticData: { content: '正在思考...' },
      timestamp: Date.now(),
      retryCount: 0
    });
    
    try {
      // 3. 实际发送请求
      const response = await this.sendToAI(conversationId, userMessage);
      
      // 4. 处理流式响应
      await this.processStreamingResponse(messageId, response);
      
      // 5. 乐观更新成功,移除pending状态
      this.pendingUpdates.delete(messageId);
      
    } catch (error) {
      // 6. 处理失败:显示错误状态
      await this.handleOptimisticFailure(messageId, error);
    }
    
    return messageId;
  }
  
  private addOptimisticMessage(conversationId: string, message: OptimisticMessage) {
    // 更新UI状态
    this.updateConversationState(conversationId, (state) => ({
      ...state,
      messages: [...state.messages, message]
    }));
  }
  
  private async processStreamingResponse(messageId: string, response: Response) {
    const reader = response.body?.getReader();
    if (!reader) return;
    
    const decoder = new TextDecoder();
    let buffer = '';
    
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      
      buffer += decoder.decode(value, { stream: true });
      const lines = buffer.split('\n\n');
      buffer = lines.pop() || '';
      
      for (const line of lines) {
        if (line.startsWith('data: ')) {
          const data = line.slice(6).trim();
          if (data === '[DONE]') return;
          
          try {
            const parsed = JSON.parse(data);
            
            // 逐步更新消息内容
            this.updateMessageContent(messageId, parsed.content);
            
            // 如果是流式生成完成,更新状态
            if (parsed.finish_reason) {
              this.updateMessageStatus(messageId, 'completed');
            }
          } catch (e) {
            console.warn('Failed to parse streaming data:', e);
          }
        }
      }
    }
  }
  
  private updateMessageContent(messageId: string, newContent: string) {
    this.updateConversationState((state) => {
      const updatedMessages = state.messages.map(msg => {
        if (msg.id === messageId) {
          return {
            ...msg,
            content: newContent,
            isOptimistic: false // 标记为非乐观更新
          };
        }
        return msg;
      });
      
      return { ...state, messages: updatedMessages };
    });
  }
  
  private async handleOptimisticFailure(messageId: string, error: any) {
    const pending = this.pendingUpdates.get(messageId);
    if (!pending) return;
    
    if (pending.retryCount < 3) {
      // 重试逻辑
      pending.retryCount++;
      this.pendingUpdates.set(messageId, pending);
      
      // 显示重试提示
      this.updateMessageContent(messageId, `请求失败,正在重试... (${pending.retryCount}/3)`);
      
      // 指数退避重试
      await this.retryWithBackoff(messageId, pending.retryCount);
    } else {
      // 最终失败
      this.updateMessageStatus(messageId, 'error');
      this.updateMessageContent(messageId, '请求失败,请稍后重试');
      this.pendingUpdates.delete(messageId);
    }
  }
  
  // 恢复机制:页面刷新后检查pending的乐观更新
  async recoverPendingUpdates() {
    for (const [messageId, pending] of this.pendingUpdates) {
      if (Date.now() - pending.timestamp > 5 * 60 * 1000) {
        // 超过5分钟,标记为失败
        this.updateMessageStatus(messageId, 'timeout');
        this.pendingUpdates.delete(messageId);
      } else {
        // 尝试重新获取结果
        await this.retryPendingUpdate(messageId);
      }
    }
  }
}

优化策略

  1. 内容预测 :基于历史对话预测AI的可能回答开头
  2. 渐进式显示 :从"正在思考" → "正在生成" → 实际内容
  3. 错误边界 :网络异常时的优雅降级
  4. 重试机制 :自动重试失败的乐观更新
  5. 状态同步 :确保乐观状态与实际状态最终一致

"印客学院"实践

  • 使用专门的 OptimisticUpdateQueue 管理所有乐观更新
  • 集成到Redux middleware中
  • 支持批量乐观更新
  • 提供回滚机制

如何用immer或immutable.js优化AI对话列表的不可变更新,避免深拷贝导致的性能问题?

场景 :"印客学院"的聊天界面需要频繁更新消息列表,每次更新都创建新对象会导致性能问题。

immer方案 (推荐):

import { produce } from 'immer';

class ConversationManager {
  private state = {
    conversations: new Map<string, Conversation>(),
    activeConversationId: null as string | null
  };
  
  // 使用immer优化更新
  addMessage(conversationId: string, message: Message) {
    this.state = produce(this.state, (draft) => {
      if (!draft.conversations.has(conversationId)) {
        draft.conversations.set(conversationId, {
          id: conversationId,
          messages: []
        });
      }
      
      const conversation = draft.conversations.get(conversationId)!;
      conversation.messages.push(message);
    });
  }
  
  // 更新消息内容(流式生成)
  updateMessageContent(conversationId: string, messageId: string, newContent: string) {
    this.state = produce(this.state, (draft) => {
      const conversation = draft.conversations.get(conversationId);
      if (!conversation) return;
      
      const message = conversation.messages.find(m => m.id === messageId);
      if (message) {
        message.content = newContent;
        message.updatedAt = Date.now();
      }
    });
  }
  
  // 批量更新
  batchUpdateMessages(updates: Array<{ conversationId: string; message: Message }>) {
    this.state = produce(this.state, (draft) => {
      for (const update of updates) {
        if (!draft.conversations.has(update.conversationId)) {
          draft.conversations.set(update.conversationId, {
            id: update.conversationId,
            messages: []
          });
        }
        
        const conversation = draft.conversations.get(update.conversationId)!;
        const existingIndex = conversation.messages.findIndex(
          m => m.id === update.message.id
        );
        
        if (existingIndex >= 0) {
          conversation.messages[existingIndex] = update.message;
        } else {
          conversation.messages.push(update.message);
        }
      }
    });
  }
}

immutable.js方案

import { Map, List, Record } from 'immutable';

// 定义不可变记录
const MessageRecord = Record({
  id: '',
  role: '',
  content: '',
  timestamp: 0
});

const ConversationRecord = Record({
  id: '',
  messages: List<MessageRecord>()
});

class ImmutableConversationManager {
  private state = Map<string, ConversationRecord>();
  
  addMessage(conversationId: string, messageData: MessageData) {
    const message = new MessageRecord(messageData);
    
    this.state = this.state.update(conversationId, (conversation) => {
      if (!conversation) {
        return new ConversationRecord({
          id: conversationId,
          messages: List([message])
        });
      }
      
      return conversation.update('messages', (messages) => 
        messages.push(message)
      );
    });
  }
  
  // 高效查找
  getMessage(conversationId: string, messageId: string) {
    const conversation = this.state.get(conversationId);
    if (!conversation) return null;
    
    return conversation.get('messages').find(
      msg => msg.get('id') === messageId
    );
  }
  
  // 分页加载
  getMessagesPaged(conversationId: string, page: number, pageSize: number) {
    const conversation = this.state.get(conversationId);
    if (!conversation) return [];
    
    const messages = conversation.get('messages');
    const start = (page - 1) * pageSize;
    const end = start + pageSize;
    
    return messages.slice(start, end).toJS();
  }
}

性能对比

特性 immer immutable.js
API友好性 ★★★★★ ★★★☆☆
性能 ★★★★☆ ★★★★★
包大小 4KB 16KB
学习曲线 简单 中等
TypeScript支持 优秀 良好

"印客学院"选择

  • 大部分场景使用immer,API更友好
  • 超大列表(1000+消息)使用immutable.js
  • 混合使用:immer处理业务逻辑,immutable.js存储基础数据

设计一个"状态版本控制"系统,支持AI对话历史的任意回退、分支创建与合并

场景 :在"印客学院"的AI协作编辑器中,用户需要像Git一样管理对话历史:创建分支、回退版本、合并修改。

核心设计

interface Commit {
  id: string;
  parentIds: string[]; // 支持合并提交
  message: string;
  timestamp: number;
  author: string;
  data: ConversationSnapshot; // 快照数据
  diff?: ConversationDiff; // 差异数据(可选)
}

class ConversationVersionControl {
  private commits = new Map<string, Commit>();
  private currentBranch = 'main';
  private branches = new Map<string, string>(); // 分支名 -> commitId
  private HEAD = ''; // 当前commitId
  
  // 创建新版本
  commit(message: string, data: ConversationSnapshot): string {
    const commitId = this.generateCommitId();
    const parentId = this.HEAD;
    
    const commit: Commit = {
      id: commitId,
      parentIds: parentId ? [parentId] : [],
      message,
      timestamp: Date.now(),
      author: this.getCurrentUser(),
      data,
      diff: parentId ? this.calculateDiff(parentId, data) : undefined
    };
    
    this.commits.set(commitId, commit);
    this.HEAD = commitId;
    this.branches.set(this.currentBranch, commitId);
    
    return commitId;
  }
  
  // 创建分支
  createBranch(branchName: string, fromCommitId?: string): boolean {
    if (this.branches.has(branchName)) {
      return false; // 分支已存在
    }
    
    const sourceCommitId = fromCommitId || this.HEAD;
    this.branches.set(branchName, sourceCommitId);
    return true;
  }
  
  // 切换分支
  checkout(branchName: string): boolean {
    if (!this.branches.has(branchName)) {
      return false;
    }
    
    this.currentBranch = branchName;
    this.HEAD = this.branches.get(branchName)!;
    return true;
  }
  
  // 回退到指定版本
  revertTo(commitId: string): boolean {
    if (!this.commits.has(commitId)) {
      return false;
    }
    
    // 创建revert提交
    const revertCommit: Commit = {
      id: this.generateCommitId(),
      parentIds: [this.HEAD],
      message: `Revert to ${commitId}`,
      timestamp: Date.now(),
      author: this.getCurrentUser(),
      data: this.commits.get(commitId)!.data
    };
    
    this.commits.set(revertCommit.id, revertCommit);
    this.HEAD = revertCommit.id;
    this.branches.set(this.currentBranch, revertCommit.id);
    
    return true;
  }
  
  // 合并分支
  merge(sourceBranch: string, targetBranch: string = this.currentBranch): MergeResult {
    const sourceCommitId = this.branches.get(sourceBranch);
    const targetCommitId = this.branches.get(targetBranch);
    
    if (!sourceCommitId || !targetCommitId) {
      return { success: false, error: 'Branch not found' };
    }
    
    // 查找最近共同祖先
    const commonAncestor = this.findCommonAncestor(sourceCommitId, targetCommitId);
    
    // 计算差异
    const sourceChanges = this.getChangesSince(commonAncestor, sourceCommitId);
    const targetChanges = this.getChangesSince(commonAncestor, targetCommitId);
    
    // 检测冲突
    const conflicts = this.detectConflicts(sourceChanges, targetChanges);
    
    if (conflicts.length > 0) {
      return {
        success: false,
        conflicts,
        sourceChanges,
        targetChanges
      };
    }
    
    // 自动合并
    const mergedData = this.autoMerge(
      this.commits.get(commonAncestor)!.data,
      sourceChanges,
      targetChanges
    );
    
    // 创建合并提交
    const mergeCommit: Commit = {
      id: this.generateCommitId(),
      parentIds: [sourceCommitId, targetCommitId],
      message: `Merge ${sourceBranch} into ${targetBranch}`,
      timestamp: Date.now(),
      author: this.getCurrentUser(),
      data: mergedData
    };
    
    this.commits.set(mergeCommit.id, mergeCommit);
    this.HEAD = mergeCommit.id;
    this.branches.set(targetBranch, mergeCommit.id);
    
    return { success: true, commitId: mergeCommit.id };
  }
  
  // 查看历史
  getHistory(commitId?: string, limit: number = 50): Commit[] {
    const startCommitId = commitId || this.HEAD;
    const history: Commit[] = [];
    let currentId = startCommitId;
    let count = 0;
    
    while (currentId && count < limit) {
      const commit = this.commits.get(currentId);
      if (!commit) break;
      
      history.push(commit);
      
      // 移动到父提交(支持线性历史,合并提交有多个父提交)
      currentId = commit.parentIds[0];
      count++;
    }
    
    return history;
  }
  
  // 可视化差异
  visualizeDiff(commitId1: string, commitId2: string): DiffVisualization {
    const commit1 = this.commits.get(commitId1);
    const commit2 = this.commits.get(commitId2);
    
    if (!commit1 || !commit2) {
      throw new Error('Commit not found');
    }
    
    const diff = this.calculateDiff(commitId1, commitId2);
    
    return {
      added: diff.added.map(item => ({
        type: 'added',
        content: item,
        context: this.getContext(item, commit2.data)
      })),
      removed: diff.removed.map(item => ({
        type: 'removed',
        content: item,
        context: this.getContext(item, commit1.data)
      })),
      modified: diff.modified.map(change => ({
        type: 'modified',
        from: change.from,
        to: change.to,
        context: this.getContext(change.from, commit1.data)
      }))
    };
  }
}

关键特性

  1. 完整版本控制 :支持commit、branch、merge、revert
  2. 差异存储 :存储完整快照或差异,节省空间
  3. 冲突检测 :自动检测和解决合并冲突
  4. 可视化历史 :提供可交互的版本树
  5. 性能优化 :懒加载历史记录,增量计算差异
  6. 离线支持 :本地存储版本历史

"印客学院"集成

  • 与编辑器深度集成
  • 实时显示版本差异
  • 支持选择性回退
  • 与团队协作功能结合

在离线优先的AI应用中,如何用RxJS或@tanstack/query管理本地缓存与网络状态的同步?

场景 :"印客学院"的移动端应用需要在离线时提供基本的AI功能,在线时同步数据。

RxJS方案 (响应式编程):

import { BehaviorSubject, from, mergeMap, catchError, tap, shareReplay } from 'rxjs';
import { fromFetch } from 'rxjs/fetch';

class OfflineFirstAIService {
  private cache = new Map<string, any>();
  private online$ = new BehaviorSubject(navigator.onLine);
  
  constructor() {
    // 监听网络状态
    window.addEventListener('online', () => this.online$.next(true));
    window.addEventListener('offline', () => this.online$.next(false));
  }
  
  // 获取AI模型列表
  getModels() {
    const cacheKey = 'models';
    const cacheTime = 5 * 60 * 1000; // 5分钟缓存
    
    return this.online$.pipe(
      mergeMap(isOnline => {
        // 检查缓存
        const cached = this.getFromCache(cacheKey, cacheTime);
        if (cached && !this.shouldRefresh(cached)) {
          return from(Promise.resolve(cached.data));
        }
        
        if (isOnline) {
          // 在线:从网络获取并缓存
          return fromFetch('/api/inke/models', {
            selector: response => response.json()
          }).pipe(
            tap(data => this.saveToCache(cacheKey, data)),
            catchError(error => {
              // 网络失败,使用缓存
              if (cached) {
                return from(Promise.resolve(cached.data));
              }
              throw error;
            })
          );
        } else {
          // 离线:使用缓存
          if (cached) {
            return from(Promise.resolve(cached.data));
          }
          throw new Error('Offline and no cache available');
        }
      }),
      shareReplay(1) // 共享结果
    );
  }
  
  // 发送消息(支持离线队列)
  sendMessage(message: AIMessage) {
    return this.online$.pipe(
      mergeMap(isOnline => {
        if (isOnline) {
          // 在线:直接发送
          return fromFetch('/api/inke/chat', {
            method: 'POST',
            body: JSON.stringify(message)
          }).pipe(
            mergeMap(response => response.json()),
            // 清空离线队列中的相关消息
            tap(() => this.removeFromOfflineQueue(message.id))
          );
        } else {
          // 离线:加入队列
          this.addToOfflineQueue(message);
          return from(Promise.resolve({
            id: message.id,
            status: 'queued',
            timestamp: Date.now()
          }));
        }
      })
    );
  }
  
  // 处理离线队列
  private processOfflineQueue() {
    this.online$.pipe(
      mergeMap(isOnline => {
        if (isOnline) {
          const queue = this.getOfflineQueue();
          return from(this.syncQueue(queue));
        }
        return from(Promise.resolve());
      })
    ).subscribe();
  }
}

@tanstack/query方案 (React专属):

import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query';

function useAIModels() {
  return useQuery({
    queryKey: ['ai-models'],
    queryFn: async () => {
      const response = await fetch('/api/inke/models');
      return response.json();
    },
    // 缓存配置
    staleTime: 5 * 60 * 1000, // 5分钟内不使用网络
    cacheTime: 10 * 60 * 1000, // 10分钟后清理缓存
    
    // 网络重试
    retry: 3,
    retryDelay: attemptIndex => Math.min(1000 * 2 ** attemptIndex, 30000),
    
    // 离线支持
    networkMode: 'offlineFirst',
    
    // 初始数据(离线时使用)
    initialData: () => {
      const cached = localStorage.getItem('inke-models-cache');
      return cached ? JSON.parse(cached) : undefined;
    }
  });
}

function useAIChat() {
  const queryClient = useQueryClient();
  
  return useMutation({
    mutationFn: async (message: string) => {
      const response = await fetch('/api/inke/chat', {
        method: 'POST',
        body: JSON.stringify({ message })
      });
      return response.json();
    },
    
    // 乐观更新
    onMutate: async (message) => {
      // 取消可能的重取
      await queryClient.cancelQueries({ queryKey: ['conversation'] });
      
      // 保存当前状态用于回滚
      const previousConversation = queryClient.getQueryData(['conversation']);
      
      // 乐观更新
      queryClient.setQueryData(['conversation'], (old: any) => ({
        ...old,
        messages: [
          ...old.messages,
          { id: Date.now(), content: message, role: 'user', status: 'sending' }
        ]
      }));
      
      return { previousConversation };
    },
    
    // 错误处理
    onError: (err, message, context) => {
      // 回滚乐观更新
      if (context?.previousConversation) {
        queryClient.setQueryData(['conversation'], context.previousConversation);
      }
      
      // 添加到离线队列
      addToOfflineQueue(message);
    },
    
    // 成功处理
    onSuccess: (data) => {
      // 更新对话
      queryClient.setQueryData(['conversation'], (old: any) => ({
        ...old,
        messages: [
          ...old.messages.filter((m: any) => m.status !== 'sending'),
          { id: data.id, content: data.content, role: 'assistant', status: 'sent' }
        ]
      }));
    }
  });
}

// 离线同步管理器
function useOfflineSync() {
  const queryClient = useQueryClient();
  
  // 监听网络状态
  useEffect(() => {
    const handleOnline = () => {
      // 网络恢复时,重新获取所有stale的查询
      queryClient.invalidateQueries();
      
      // 处理离线队列
      processOfflineQueue();
    };
    
    window.addEventListener('online', handleOnline);
    return () => window.removeEventListener('online', handleOnline);
  }, [queryClient]);
}

对比选择

特性 RxJS @tanstack/query
适用场景 通用JavaScript/TypeScript React应用
学习曲线 较陡峭 较平缓
缓存管理 需手动实现 内置完善
离线支持 灵活自定义 内置基础支持
状态同步 响应式流处理 基于Hook的声明式

"印客学院"实践

  • Web端:使用@tanstack/query,简化React状态管理
  • 移动端:使用RxJS,更灵活处理复杂异步流
  • 混合方案:@tanstack/query管理UI状态,RxJS处理业务逻辑流

如何用Recoil或Jotai的原子机制实现AI生成参数的细粒度响应式更新?

场景 :在"印客学院"的AI参数调优面板中,各个参数(温度、top_p、重复惩罚等)相互关联,需要实时响应式更新。

Recoil方案

import { atom, selector, useRecoilState, useRecoilValue } from 'recoil';

// 基础参数原子
const temperatureAtom = atom<number>({
  key: 'temperature',
  default: 0.7,
  effects: [
    // 持久化效果
    ({ onSet, setSelf }) => {
      const saved = localStorage.getItem('inke-temperature');
      if (saved !== null) {
        setSelf(JSON.parse(saved));
      }
      
      onSet((newValue) => {
        localStorage.setItem('inke-temperature', JSON.stringify(newValue));
      });
    },
    
    // 验证效果
    ({ onSet }) => {
      onSet((newValue) => {
        if (newValue < 0 || newValue > 2) {
          console.warn('Temperature should be between 0 and 2');
        }
      });
    }
  ]
});

const topPAtom = atom<number>({
  key: 'topP',
  default: 0.9
});

const frequencyPenaltyAtom = atom<number>({
  key: 'frequencyPenalty',
  default: 0
});

// 派生状态:参数组合
const generationParamsSelector = selector({
  key: 'generationParams',
  get: ({ get }) => {
    const temperature = get(temperatureAtom);
    const topP = get(topPAtom);
    const frequencyPenalty = get(frequencyPenaltyAtom);
    
    return {
      temperature,
      top_p: topP,
      frequency_penalty: frequencyPenalty,
      // 计算其他派生值
      creativity: this.calculateCreativity(temperature, topP),
      randomness: this.calculateRandomness(temperature)
    };
  },
  set: ({ set }, newValue) => {
    // 批量设置多个原子
    if (typeof newValue === 'object') {
      if ('temperature' in newValue) {
        set(temperatureAtom, newValue.temperature);
      }
      if ('top_p' in newValue) {
        set(topPAtom, newValue.top_p);
      }
    }
  }
});

// 参数预设
const parameterPresetsAtom = atom<ParameterPreset[]>({
  key: 'parameterPresets',
  default: [
    { id: 'creative', name: '创意模式', temperature: 0.9, topP: 0.95 },
    { id: 'precise', name: '精确模式', temperature: 0.3, topP: 0.8 },
    { id: 'balanced', name: '平衡模式', temperature: 0.7, topP: 0.9 }
  ]
});

// 当前预设选择器
const currentPresetSelector = selector<ParameterPreset | null>({
  key: 'currentPreset',
  get: ({ get }) => {
    const params = get(generationParamsSelector);
    const presets = get(parameterPresetsAtom);
    
    return presets.find(preset => 
      Math.abs(preset.temperature - params.temperature) < 0.01 &&
      Math.abs(preset.topP - params.top_p) < 0.01
    ) || null;
  },
  set: ({ set }, newPreset) => {
    if (newPreset && typeof newPreset === 'object') {
      set(temperatureAtom, newPreset.temperature);
      set(topPAtom, newPreset.topP);
    }
  }
});

// React组件使用
function ParameterPanel() {
  const [temperature, setTemperature] = useRecoilState(temperatureAtom);
  const [topP, setTopP] = useRecoilState(topPAtom);
  const params = useRecoilValue(generationParamsSelector);
  const currentPreset = useRecoilValue(currentPresetSelector);
  
  return (
    <div>
      <div>
        <label>Temperature: {temperature.toFixed(2)}</label>
        <input
          type="range"
          min="0"
          max="2"
          step="0.1"
          value={temperature}
          onChange={(e) => setTemperature(parseFloat(e.target.value))}
        />
      </div>
      
      <div>
        <label>Top-p: {topP.toFixed(2)}</label>
        <input
          type="range"
          min="0"
          max="1"
          step="0.05"
          value={topP}
          onChange={(e) => setTopP(parseFloat(e.target.value))}
        />
      </div>
      
      <div>Creativity: {(params.creativity * 100).toFixed(0)}%</div>
      <div>Randomness: {(params.randomness * 100).toFixed(0)}%</div>
      
      {currentPreset && (
        <div>Current preset: {currentPreset.name}</div>
      )}
    </div>
  );
}

Jotai方案 (更轻量):

import { atom, useAtom, useAtomValue } from 'jotai';
import { selectAtom } from 'jotai/utils';

// 基础原子
const temperatureAtom = atom(0.7);
const topPAtom = atom(0.9);

// 派生原子
const generationParamsAtom = atom((get) => {
  const temperature = get(temperatureAtom);
  const topP = get(topPAtom);
  
  return {
    temperature,
    top_p: topP,
    creativity: temperature * 0.5 + topP * 0.5,
    randomness: temperature * 0.8
  };
});

// 只读派生原子
const creativityAtom = atom((get) => {
  const params = get(generationParamsAtom);
  return params.creativity;
});

// 异步原子(从API获取默认值)
const defaultParamsAtom = atom(async () => {
  const response = await fetch('/api/inke/default-params');
  return response.json();
});

// 带缓存的原子
const cachedTemperatureAtom = atom(
  (get) => get(temperatureAtom),
  (get, set, newValue: number) => {
    set(temperatureAtom, newValue);
    // 保存到localStorage
    localStorage.setItem('inke-temperature', newValue.toString());
  }
);

// 原子操作
const resetParamsAtom = atom(null, (get, set) => {
  set(temperatureAtom, 0.7);
  set(topPAtom, 0.9);
});

// 选择器(性能优化)
const highTemperatureAtom = selectAtom(
  temperatureAtom,
  (temp) => temp > 1.0
);

// 组件使用
function JotaiParameterPanel() {
  const [temperature, setTemperature] = useAtom(temperatureAtom);
  const [topP, setTopP] = useAtom(topPAtom);
  const params = useAtomValue(generationParamsAtom);
  const isHighTemp = useAtomValue(highTemperatureAtom);
  
  return (
    <div>
      <input
        type="range"
        value={temperature}
        onChange={(e) => setTemperature(parseFloat(e.target.value))}
      />
      {isHighTemp && <span style={{ color: 'red' }}>High temperature warning!</span>}
    </div>
  );
}

性能优化技巧

  1. 原子粒度 :细粒度原子提高复用性
  2. 选择器 :避免不必要的重计算
  3. 记忆化 :对昂贵计算进行缓存
  4. 批量更新 :减少渲染次数
  5. 懒加载 :大型状态按需加载

"印客学院"选择

  • 新项目使用Jotai,API更简洁
  • 大型复杂项目使用Recoil,生态更完善
  • 混合使用:核心状态用Recoil,UI状态用Jotai

设计一个"状态持久化"方案,将AI应用的关键状态自动保存至IndexedDB,并支持跨标签页同步

场景 :"印客学院"的Web应用需要在用户关闭浏览器后保留AI对话历史、用户设置等状态,并支持多标签页同步。

完整实现

interface PersistenceConfig {
  key: string;
  version: number;
  serialize: (state: any) => any;
  deserialize: (data: any) => any;
  migrate?: (oldData: any, oldVersion: number) => any;
  maxSize?: number; // 最大存储大小(字节)
  ttl?: number; // 生存时间(毫秒)
}

class IndexedDBPersistence {
  private dbName = 'InkeAIStateDB';
  private dbVersion = 3;
  private storeName = 'appState';
  private db: IDBDatabase | null = null;
  private configs = new Map<string, PersistenceConfig>();
  private listeners = new Map<string, Set<(data: any) => void>>();
  
  constructor() {
    this.initDB();
    this.setupCrossTabSync();
  }
  
  private async initDB(): Promise<void> {
    return new Promise((resolve, reject) => {
      const request = indexedDB.open(this.dbName, this.dbVersion);
      
      request.onupgradeneeded = (event) => {
        const db = (event.target as IDBOpenDBRequest).result;
        
        if (!db.objectStoreNames.contains(this.storeName)) {
          const store = db.createObjectStore(this.storeName, { keyPath: 'key' });
          store.createIndex('timestamp', 'timestamp', { unique: false });
          store.createIndex('version', 'version', { unique: false });
        }
        
        // 迁移旧数据
        this.handleMigration(db, event.oldVersion);
      };
      
      request.onsuccess = () => {
        this.db = request.result;
        resolve();
      };
      
      request.onerror = () => reject(request.error);
    });
  }
  
  // 注册需要持久化的状态
  register<T>(config: PersistenceConfig) {
    this.configs.set(config.key, config);
    
    // 自动加载已保存的状态
    this.load(config.key).then(data => {
      if (data) {
        this.notifyListeners(config.key, data);
      }
    });
    
    return {
      save: (data: T) => this.save(config.key, data),
      load: () => this.load<T>(config.key),
      subscribe: (listener: (data: T) => void) => 
        this.subscribe(config.key, listener)
    };
  }
  
  // 保存状态
  async save(key: string, data: any): Promise<void> {
    if (!this.db) throw new Error('Database not initialized');
    
    const config = this.configs.get(key);
    if (!config) throw new Error(`No config found for key: ${key}`);
    
    const serialized = config.serialize(data);
    const item = {
      key,
      data: serialized,
      version: config.version,
      timestamp: Date.now(),
      size: new TextEncoder().encode(JSON.stringify(serialized)).length
    };
    
    return new Promise((resolve, reject) => {
      const transaction = this.db!.transaction(this.storeName, 'readwrite');
      const store = transaction.objectStore(this.storeName);
      const request = store.put(item);
      
      request.onsuccess = () => {
        // 通知其他标签页
        this.broadcastChange(key, data);
        
        // 检查存储限制
        this.enforceStorageLimits();
        
        resolve();
      };
      
      request.onerror = () => reject(request.error);
    });
  }
  
  // 加载状态
  async load<T>(key: string): Promise<T | null> {
    if (!this.db) throw new Error('Database not initialized');
    
    const config = this.configs.get(key);
    if (!config) throw new Error(`No config found for key: ${key}`);
    
    return new Promise((resolve, reject) => {
      const transaction = this.db!.transaction(this.storeName, 'readonly');
      const store = transaction.objectStore(this.storeName);
      const request = store.get(key);
      
      request.onsuccess = () => {
        const result = request.result;
        if (!result) {
          resolve(null);
          return;
        }
        
        // 数据迁移
        let data = result.data;
        if (result.version < config.version && config.migrate) {
          data = config.migrate(result.data, result.version);
        }
        
        const deserialized = config.deserialize(data);
        resolve(deserialized);
      };
      
      request.onerror = () => reject(request.error);
    });
  }
  
  // 订阅状态变化
  subscribe(key: string, listener: (data: any) => void): () => void {
    if (!this.listeners.has(key)) {
      this.listeners.set(key, new Set());
    }
    
    this.listeners.get(key)!.add(listener);
    
    return () => {
      const listeners = this.listeners.get(key);
      if (listeners) {
        listeners.delete(listener);
        if (listeners.size === 0) {
          this.listeners.delete(key);
        }
      }
    };
  }
  
  private notifyListeners(key: string, data: any) {
    const listeners = this.listeners.get(key);
    if (listeners) {
      listeners.forEach(listener => listener(data));
    }
  }
  
  // 跨标签页同步
  private setupCrossTabSync() {
    // 使用BroadcastChannel API
    const channel = new BroadcastChannel('inke-state-sync');
    
    channel.onmessage = (event) => {
      if (event.data.type === 'STATE_CHANGE') {
        const { key, data } = event.data;
        this.notifyListeners(key, data);
      }
    };
    
    this.broadcastChange = (key, data) => {
      channel.postMessage({
        type: 'STATE_CHANGE',
        key,
        data,
        timestamp: Date.now(),
        source: 'tab-' + Math.random().toString(36).substr(2, 9)
      });
    };
  }
  
  private broadcastChange: (key: string, data: any) => void = () => {};
  
  // 存储限制管理
  private async enforceStorageLimits() {
    if (!this.db) return;
    
    const transaction = this.db.transaction(this.storeName, 'readwrite');
    const store = transaction.objectStore(this.storeName);
    const index = store.index('timestamp');
    
    // 获取所有记录并按时间排序
    const request = index.getAll();
    
    request.onsuccess = () => {
      const records = request.result as Array<{ key: string; timestamp: number; size: number }>;
      
      // 检查TTL
      const now = Date.now();
      for (const record of records) {
        const config = this.configs.get(record.key);
        if (config?.ttl && now - record.timestamp > config.ttl) {
          store.delete(record.key);
        }
      }
      
      // 检查总大小限制
      const totalSize = records.reduce((sum, record) => sum + (record.size || 0), 0);
      const maxSize = 50 * 1024 * 1024; // 50MB
      
      if (totalSize > maxSize) {
        // 按时间从旧到新删除
        const sorted = records.sort((a, b) => a.timestamp - b.timestamp);
        let removedSize = 0;
        
        for (const record of sorted) {
          if (removedSize >= totalSize - maxSize) break;
          
          store.delete(record.key);
          removedSize += record.size || 0;
        }
      }
    };
  }
  
  // 数据迁移处理
  private handleMigration(db: IDBDatabase, oldVersion: number) {
    if (oldVersion < 2) {
      // 从v1迁移到v2
      const transaction = db.transaction(this.storeName, 'readwrite');
      const store = transaction.objectStore(this.storeName);
      const request = store.openCursor();
      
      request.onsuccess = () => {
        const cursor = request.result;
        if (cursor) {
          const data = cursor.value;
          // 迁移逻辑
          if (data.version === 1) {
            data.version = 2;
            // 数据格式转换
            if (data.key === 'conversations') {
              data.data = this.migrateConversationsV1toV2(data.data);
            }
            cursor.update(data);
          }
          cursor.continue();
        }
      };
    }
  }
}

使用示例

// 配置对话状态持久化
const conversationPersistence = persistence.register({
  key: 'conversations',
  version: 2,
  serialize: (state) => ({
    ...state,
    // 特殊处理流式消息
    messages: state.messages.map(msg => ({
      ...msg,
      // 不保存过大的临时数据
      streamingBuffer: undefined
    }))
  }),
  deserialize: (data) => data,
  migrate: (oldData, oldVersion) => {
    if (oldVersion === 1) {
      // 从v1迁移到v2
      return {
        ...oldData,
        messages: oldData.messages.map((msg: any) => ({
          ...msg,
          // 添加新字段
          metadata: { createdAt: Date.now() }
        }))
      };
    }
    return oldData;
  },
  ttl: 30 * 24 * 60 * 60 * 1000, // 30天
  maxSize: 10 * 1024 * 1024 // 10MB
});

// 在组件中使用
class ConversationManager {
  private persistence = conversationPersistence;
  
  async loadConversations() {
    const saved = await this.persistence.load();
    if (saved) {
      this.restoreState(saved);
    }
  }
  
  async saveConversations() {
    await this.persistence.save(this.state);
  }
  
  // 自动保存
  setupAutoSave() {
    // 防抖保存
    let saveTimeout: NodeJS.Timeout;
    const scheduleSave = () => {
      clearTimeout(saveTimeout);
      saveTimeout = setTimeout(() => this.saveConversations(), 1000);
    };
    
    // 监听状态变化
    this.persistence.subscribe((data) => {
      if (data) {
        this.restoreState(data);
      }
    });
  }
}

高级特性

  1. 增量保存 :只保存变化的部分
  2. 压缩存储 :使用gzip压缩大文本
  3. 加密存储 :敏感数据加密后存储
  4. 冲突解决 :多标签页编辑时的冲突解决
  5. 备份恢复 :定期备份,支持恢复任意版本
  6. 性能监控 :记录存储操作性能指标

在AI可视化编辑器中,如何用Mobx实现画布节点、连接线、属性面板的双向数据绑定?

场景 :在“印客学院”的AI工作流可视化编辑器中,用户通过拖拽创建节点、连线定义流程,并在属性面板调整参数,三者状态需实时同步。

核心答案 :使用Mobx的 observablecomputedaction 构建一个响应式状态模型。将画布节点、连线定义为可观察对象,属性面板绑定到当前选中元素的观察属性。通过 reactionautorun 自动响应状态变化并更新视图,实现“修改属性 → 画布节点更新”和“拖动节点 → 属性面板更新”的双向绑定。

实现步骤

  1. 定义可观察数据模型

    • WorkflowNode 类:包含id、位置、尺寸、类型、配置参数等 observable 属性
    • WorkflowEdge 类:包含id、起点节点ID、终点节点ID、连接点等 observable 属性
    • WorkflowStore 类:管理节点和边的集合,以及当前选中元素的状态
  2. 建立响应式关系

    • 使用 computed 计算画布渲染所需的数据结构(如节点层级、连线路径)
    • 将属性面板的表单字段绑定到 currentSelection 的对应属性
    • 通过 reaction 监听节点位置变化,自动更新相关连线的路径计算
  3. 实现双向更新

    • 属性面板修改 → 通过 action 更新节点配置 → 触发画布重绘
    • 画布拖拽节点 → 通过 action 更新节点位置 → 触发属性面板刷新
    • 删除节点 → 自动删除相关连线(通过 reaction 实现级联清理)
  4. 性能优化

    • 使用 observable.ref 处理大型数据对象
    • 对高频更新操作(如拖拽)使用 transaction 批量处理
    • 通过 computed 缓存计算结果,避免重复计算

关键设计 :Mobx的响应式系统自动追踪状态依赖,无需手动订阅/发布。当任何 observable 属性变化时,所有依赖它的 computed 值和 reaction 都会自动更新,实现细粒度的精准重绘。


如何用Redux-Saga或Redux-Observable处理AI请求的复杂副作用?

场景 :“印客学院”的AI对话系统需处理:1)流式请求需支持中途取消;2)多个AI模型调用需竞态处理;3)失败请求需按策略重试;4)长时间运行任务需轮询状态。

核心答案 :两者都是Redux中间件,用于管理异步副作用。 Redux-Saga 使用Generator函数,以同步方式写异步逻辑,更易读易测试; Redux-Observable 基于RxJS,以响应式流的方式处理,更适合复杂的事件流和组合操作。

Redux-Saga实现复杂副作用的模式

function* watchAIGeneration() {
  // 1. 处理带取消的流式请求
  yield takeLatest('START_AI_GENERATION', withCancel(handleGeneration));
  
  // 2. 处理竞态:多个模型同时请求,取最快响应
  yield takeLatest('COMPARE_MODELS', function* (action) {
    const { fastest } = yield race({
      gpt: call(fetchFromGPT, action.prompt),
      claude: call(fetchFromClaude, action.prompt),
      inke: call(fetchFromInke, action.prompt),
      timeout: delay(10000) // 10秒超时
    });
    if (fastest) yield put(setResult(fastest));
  });
  
  // 3. 指数退避重试
  yield takeEvery('SEND_MESSAGE', withRetry(handleMessage, {
    maxAttempts: 3,
    backoff: exponentialBackoff(1000) // 1秒、2秒、4秒
  }));
  
  // 4. 轮询长任务状态
  yield takeLatest('START_TRAINING', function* (action) {
    yield call(pollTrainingStatus, action.taskId, {
      interval: 2000, // 每2秒轮询
      timeout: 300000 // 5分钟超时
    });
  });
}

Redux-Observable的响应式处理

const aiRequestEpic = (action$, state$) => 
  action$.pipe(
    filter(action => action.type === 'SEND_MESSAGE'),
    // 防抖:避免快速连续发送
    debounceTime(300),
    // 切换Map:新请求自动取消旧请求
    switchMap(action =>
      from(fetchAIResponse(action.payload)).pipe(
        // 重试逻辑
        retryWhen(errors =>
          errors.pipe(
            scan((retryCount, error) => {
              if (retryCount >= 3) throw error;
              return retryCount + 1;
            }, 0),
            delayWhen(retryCount => timer(retryCount * 1000))
          )
        ),
        // 超时处理
        timeout(30000),
        map(response => successAction(response)),
        catchError(error => of(errorAction(error)))
      )
    )
  );

选择建议

  • Redux-Saga :适合步骤明确、需要精细控制的业务流程(如AI工作流引擎)
  • Redux-Observable :适合事件驱动、需要复杂流转换的场景(如实时AI监控面板)
  • “印客学院”实践 :主要业务逻辑用Saga,实时数据流用Observable

设计一个“状态迁移”工具,当AI接口版本升级导致数据结构变化时,自动转换旧版持久化状态

场景 :“印客学院”的AI服务从v1升级到v2,消息数据结构发生变化(如新增 reasoning 字段, confidence 从数字变为对象)。已保存在用户本地的v1格式对话历史需自动迁移到v2格式。

核心答案 :实现一个版本化的迁移管道,包含版本检测、迁移脚本注册、顺序执行、回滚机制。每个迁移脚本负责将数据从N版本升级到N+1版本,支持链式迁移。

实现方案

  1. 版本标识 :在每个持久化状态对象中添加 version 字段
  2. 迁移注册表 :注册从每个旧版本到新版本的迁移函数
  3. 迁移执行器 :检测当前版本与目标版本差异,按顺序执行所需迁移脚本
  4. 验证与回滚 :迁移后验证数据完整性,失败时回滚到上一版本

关键代码

class StateMigrationManager {
  constructor() {
    // 迁移脚本注册表
    this.migrations = new Map();
    this.registerMigration('1.0', '2.0', this.v1ToV2);
    this.registerMigration('2.0', '2.1', this.v2ToV2_1);
  }
  
  // 迁移函数示例:v1到v2
  v1ToV2(data) {
    return {
      ...data,
      version: '2.0',
      messages: data.messages.map(msg => ({
        ...msg,
        // 新增reasoning字段
        reasoning: msg.type === 'assistant' ? '自动迁移生成' : undefined,
        // 转换confidence格式
        confidence: typeof msg.confidence === 'number' 
          ? { score: msg.confidence, factors: [] } 
          : msg.confidence
      }))
    };
  }
  
  // 执行迁移
  async migrate(data, targetVersion) {
    let currentVersion = data.version || '1.0';
    const migrationPath = this.getMigrationPath(currentVersion, targetVersion);
    
    let currentData = { ...data };
    
    for (const { from, to, migrate } of migrationPath) {
      try {
        console.log(`迁移中: ${from} -> ${to}`);
        currentData = await migrate(currentData);
        currentData.version = to;
        
        // 验证迁移结果
        if (!this.validateData(currentData, to)) {
          throw new Error(`迁移验证失败: ${from} -> ${to}`);
        }
        
        // 备份检查点
        await this.createCheckpoint(currentData, from, to);
      } catch (error) {
        // 迁移失败,尝试回滚
        await this.rollbackToCheckpoint(data, from);
        throw new Error(`迁移失败: ${error.message}`);
      }
    }
    
    return currentData;
  }
  
  // 获取迁移路径
  getMigrationPath(fromVersion, toVersion) {
    const path = [];
    let current = fromVersion;
    
    while (current !== toVersion) {
      const migration = this.findMigration(current, toVersion);
      if (!migration) {
        throw new Error(`找不到从 ${current} 到 ${toVersion} 的迁移路径`);
      }
      path.push(migration);
      current = migration.to;
    }
    
    return path;
  }
}

迁移策略

  1. 增量迁移 :使用时迁移,非一次性全部迁移
  2. 向后兼容 :新代码能读取旧格式数据
  3. 数据验证 :迁移后验证数据完整性和业务规则
  4. 性能优化 :大数据集分块迁移,避免阻塞主线程
  5. 用户透明 :迁移过程提供进度反馈,允许用户取消

在AI多租户系统中,如何用上下文(Context)或依赖注入管理不同租户的独立状态实例?

场景 :“印客学院”SaaS平台服务多个教育机构(租户),每个租户有独立的AI模型配置、对话历史、用户权限。需要在前端隔离各租户的状态,避免数据泄露。

核心答案 :使用React Context提供租户感知的状态容器,或通过依赖注入框架为每个租户创建独立的状态实例。在应用入口根据当前租户ID切换状态上下文,所有组件自动获取对应租户的状态。

React Context方案

// 1. 创建租户上下文
const TenantContext = React.createContext({
  tenantId: null,
  config: {},
  aiModels: [],
  conversations: new Map()
});

// 2. 租户状态提供者
function TenantProvider({ tenantId, children }) {
  // 根据tenantId加载租户特定状态
  const [state, setState] = useState(() => 
    loadTenantState(tenantId)
  );
  
  // 状态自动保存到租户隔离的存储
  useEffect(() => {
    saveTenantState(tenantId, state);
  }, [tenantId, state]);
  
  return (
    <TenantContext.Provider value={{ tenantId, ...state, setState }}>
      {children}
    </TenantContext.Provider>
  );
}

// 3. 租户感知的Hook
function useTenantAI() {
  const { tenantId, config, setState } = useContext(TenantContext);
  
  const sendMessage = useCallback(async (message) => {
    // 使用租户特定的API端点
    const response = await fetch(`/api/${tenantId}/ai/chat`, {
      headers: { 'X-Tenant-Config': JSON.stringify(config) }
    });
    // 更新租户特定的状态
    setState(prev => addMessage(prev, response));
  }, [tenantId, config, setState]);
  
  return { sendMessage };
}

// 4. 应用入口:根据路由或身份信息确定租户
function App() {
  const { tenantId } = useAuth(); // 从认证信息获取租户ID
  
  return (
    <TenantProvider tenantId={tenantId}>
      <Dashboard />
    </TenantProvider>
  );
}

依赖注入方案

// 租户作用域的状态容器
class TenantScope {
  private instances = new Map<string, any>();
  
  get<T>(key: string, factory: () => T): T {
    if (!this.instances.has(key)) {
      this.instances.set(key, factory());
    }
    return this.instances.get(key);
  }
  
  clear() {
    this.instances.clear();
  }
}

// 租户状态管理器
class TenantStateManager {
  private scopes = new Map<string, TenantScope>();
  
  getScope(tenantId: string): TenantScope {
    if (!this.scopes.has(tenantId)) {
      this.scopes.set(tenantId, new TenantScope());
    }
    return this.scopes.get(tenantId)!;
  }
  
  // 获取租户特定的AI服务
  getAIService(tenantId: string): AIService {
    const scope = this.getScope(tenantId);
    return scope.get('aiService', () => new AIService({
      endpoint: `/api/${tenantId}/ai`,
      config: this.getTenantConfig(tenantId)
    }));
  }
  
  // 清理租户状态(登出时)
  clearTenant(tenantId: string) {
    this.scopes.get(tenantId)?.clear();
    this.scopes.delete(tenantId);
  }
}

状态隔离策略

  1. 存储隔离 :每个租户数据存储在独立的IndexedDB database或localStorage key中
  2. 内存隔离 :租户状态在内存中完全隔离,通过租户ID索引
  3. 网络隔离 :API请求自动携带租户上下文
  4. 错误隔离 :一个租户的状态错误不影响其他租户
  5. 生命周期 :租户切换时自动清理前租户的状态

如何用Vue3的Composition API或React Hooks封装可复用的AI状态逻辑?

场景 :在“印客学院”的前端项目中,多个页面都需要AI聊天、代码生成、内容总结等功能。需要将这些AI交互的状态逻辑封装为可复用的Hook,避免重复代码。

Vue3 Composition API实现

// useAIChat.js - AI聊天Hook
import { ref, computed, watch } from 'vue';
import { useAIStream } from './useAIStream';

export function useAIChat(options = {}) {
  const {
    initialMessages = [],
    model = 'gpt-4',
    temperature = 0.7
  } = options;
  
  // 状态
  const messages = ref(initialMessages);
  const inputText = ref('');
  const isLoading = ref(false);
  const error = ref(null);
  
  // 依赖注入AI流服务
  const { streamResponse, cancelStream } = useAIStream();
  
  // 计算属性
  const canSend = computed(() => 
    !isLoading.value && inputText.value.trim().length > 0
  );
  
  const lastMessage = computed(() => 
    messages.value[messages.value.length - 1]
  );
  
  // 动作
  const sendMessage = async () => {
    if (!canSend.value) return;
    
    const userMessage = inputText.value.trim();
    inputText.value = '';
    
    // 添加用户消息
    messages.value.push({
      id: generateId(),
      role: 'user',
      content: userMessage,
      timestamp: Date.now()
    });
    
    // 添加AI响应占位符
    const aiMessageId = generateId();
    messages.value.push({
      id: aiMessageId,
      role: 'assistant',
      content: '',
      status: 'thinking',
      timestamp: Date.now()
    });
    
    isLoading.value = true;
    error.value = null;
    
    try {
      // 流式获取AI响应
      await streamResponse({
        messages: messages.value.slice(0, -1), // 不包括刚添加的占位符
        model,
        temperature
      }, {
        onChunk: (chunk) => {
          // 更新AI消息内容
          const index = messages.value.findIndex(m => m.id === aiMessageId);
          if (index !== -1) {
            messages.value[index].content += chunk;
            messages.value[index].status = 'streaming';
          }
        },
        onComplete: () => {
          const index = messages.value.findIndex(m => m.id === aiMessageId);
          if (index !== -1) {
            messages.value[index].status = 'completed';
          }
        }
      });
    } catch (err) {
      error.value = err.message;
      const index = messages.value.findIndex(m => m.id === aiMessageId);
      if (index !== -1) {
        messages.value[index].status = 'error';
      }
    } finally {
      isLoading.value = false;
    }
  };
  
  const clearChat = () => {
    messages.value = [];
    error.value = null;
    cancelStream();
  };
  
  // 自动保存
  watch(messages, (newMessages) => {
    localStorage.setItem('inke-chat-history', JSON.stringify(newMessages));
  }, { deep: true, immediate: true });
  
  return {
    // 状态
    messages,
    inputText,
    isLoading,
    error,
    
    // 计算属性
    canSend,
    lastMessage,
    
    // 动作
    sendMessage,
    clearChat,
    cancelStream
  };
}

React Hooks实现

// useAIChat.ts - AI聊天Hook
import { useState, useCallback, useRef, useEffect } from 'react';
import { useAIStream } from './useAIStream';

interface UseAIChatOptions {
  initialMessages?: Array<{ role: string; content: string }>;
  model?: string;
  temperature?: number;
  autoSave?: boolean;
}

export function useAIChat(options: UseAIChatOptions = {}) {
  const {
    initialMessages = [],
    model = 'gpt-4',
    temperature = 0.7,
    autoSave = true
  } = options;
  
  // 状态
  const [messages, setMessages] = useState(initialMessages);
  const [input, setInput] = useState('');
  const [isLoading, setIsLoading] = useState(false);
  const [error, setError] = useState<string | null>(null);
  
  // 引用
  const messagesRef = useRef(messages);
  const abortControllerRef = useRef<AbortController | null>(null);
  
  // 依赖的Hook
  const { streamResponse } = useAIStream();
  
  // 发送消息
  const sendMessage = useCallback(async () => {
    if (!input.trim() || isLoading) return;
    
    const userMessage = input.trim();
    setInput('');
    
    // 乐观更新:添加用户消息
    const userMsg = {
      id: `msg_${Date.now()}`,
      role: 'user' as const,
      content: userMessage,
      timestamp: Date.now()
    };
    
    // 添加AI消息占位符
    const aiMsgId = `ai_${Date.now()}`;
    const aiMsg = {
      id: aiMsgId,
      role: 'assistant' as const,
      content: '',
      status: 'thinking' as const,
      timestamp: Date.now()
    };
    
    setMessages(prev => [...prev, userMsg, aiMsg]);
    setIsLoading(true);
    setError(null);
    
    // 创建可取消的请求
    abortControllerRef.current = new AbortController();
    
    try {
      await streamResponse(
        {
          messages: [...messagesRef.current, userMsg],
          model,
          temperature
        },
        {
          signal: abortControllerRef.current.signal,
          onChunk: (chunk) => {
            // 更新流式内容
            setMessages(prev => prev.map(msg => 
              msg.id === aiMsgId 
                ? { ...msg, content: msg.content + chunk, status: 'streaming' as const }
                : msg
            ));
          },
          onComplete: () => {
            setMessages(prev => prev.map(msg => 
              msg.id === aiMsgId 
                ? { ...msg, status: 'completed' as const }
                : msg
            ));
          }
        }
      );
    } catch (err: any) {
      if (err.name === 'AbortError') {
        console.log('请求被取消');
      } else {
        setError(err.message);
        setMessages(prev => prev.map(msg => 
          msg.id === aiMsgId 
            ? { ...msg, status: 'error' as const }
            : msg
        ));
      }
    } finally {
      setIsLoading(false);
      abortControllerRef.current = null;
    }
  }, [input, isLoading, model, temperature, streamResponse]);
  
  // 取消当前生成
  const cancelGeneration = useCallback(() => {
    if (abortControllerRef.current) {
      abortControllerRef.current.abort();
      setIsLoading(false);
    }
  }, []);
  
  // 清空对话
  const clearChat = useCallback(() => {
    setMessages([]);
    setError(null);
    cancelGeneration();
  }, [cancelGeneration]);
  
  // 自动保存
  useEffect(() => {
    if (autoSave) {
      messagesRef.current = messages;
      localStorage.setItem('inke-chat-history', JSON.stringify(messages));
    }
  }, [messages, autoSave]);
  
  // 恢复历史
  const loadHistory = useCallback(() => {
    try {
      const saved = localStorage.getItem('inke-chat-history');
      if (saved) {
        setMessages(JSON.parse(saved));
      }
    } catch (err) {
      console.warn('恢复对话历史失败:', err);
    }
  }, []);
  
  return {
    // 状态
    messages,
    input,
    isLoading,
    error,
    
    // 更新函数
    setInput,
    setMessages,
    
    // 动作
    sendMessage,
    cancelGeneration,
    clearChat,
    loadHistory
  };
}

Hook设计原则

  1. 单一职责 :每个Hook只管理一个特定AI功能的状态
  2. 依赖注入 :通过参数注入配置和外部服务
  3. 响应式 :状态变化自动触发副作用
  4. 可组合 :小Hook组合成大Hook(如 useAIChat 使用 useAIStream
  5. 类型安全 :TypeScript提供完整类型定义
  6. 测试友好 :纯逻辑,易于单元测试

设计一个"状态审计"系统,记录AI应用中的所有状态变更,便于调试与回溯

场景 :在"印客学院"的AI调试平台中,需要追踪状态变化的完整历史,以便在出现异常时回溯问题根源,或复现用户操作路径。

核心答案 :实现一个状态变更审计中间件,拦截所有状态更新动作,记录"谁、何时、何故"变更了哪个状态。审计日志包含变更前后的状态差异、调用栈、用户上下文等信息,支持过滤、搜索和时间旅行。

系统设计

interface AuditLogEntry {
  id: string;
  timestamp: number;
  action: {
    type: string;
    payload: any;
    source: 'USER' | 'SYSTEM' | 'AI'; // 变更来源
  };
  stateChange: {
    path: string[]; // 状态路径,如 ['conversations', 'msg_123', 'content']
    oldValue: any;
    newValue: any;
    diff: any; // 结构化差异
  };
  context: {
    userId: string;
    sessionId: string;
    route: string;
    userAgent: string;
  };
  stackTrace?: string; // 开发环境记录调用栈
  performance?: {
    duration: number;
    memoryDelta: number;
  };
}

class StateAuditSystem {
  private logs: AuditLogEntry[] = [];
  private isEnabled = true;
  private maxLogs = 10000; // 最大日志数
  private filters = {
    minDuration: 10, // 只记录超过10ms的状态变更
    ignoredActions: ['MOUSE_MOVE', 'SCROLL'], // 忽略高频低价值动作
    sensitivePaths: ['user.password', 'tokens'] // 敏感路径脱敏
  };
  
  // Redux中间件
  createAuditMiddleware = store => next => action => {
    if (!this.isEnabled || this.filters.ignoredActions.includes(action.type)) {
      return next(action);
    }
    
    const startTime = performance.now();
    const startMemory = performance.memory?.usedJSHeapSize || 0;
    const oldState = store.getState();
    
    // 执行原始action
    const result = next(action);
    
    const endTime = performance.now();
    const endMemory = performance.memory?.usedJSHeapSize || 0;
    const newState = store.getState();
    
    // 计算状态差异
    const diffs = this.calculateStateDiffs(oldState, newState);
    
    // 为每个变化创建审计日志
    diffs.forEach(diff => {
      // 敏感信息脱敏
      if (this.isSensitivePath(diff.path)) {
        diff.oldValue = this.maskSensitiveData(diff.oldValue);
        diff.newValue = this.maskSensitiveData(diff.newValue);
      }
      
      const logEntry: AuditLogEntry = {
        id: `log_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`,
        timestamp: Date.now(),
        action: {
          type: action.type,
          payload: this.sanitizePayload(action.payload),
          source: this.determineActionSource(action)
        },
        stateChange: {
          path: diff.path,
          oldValue: diff.oldValue,
          newValue: diff.newValue,
          diff: diff.diff
        },
        context: this.getCurrentContext(),
        stackTrace: this.isDevelopment() ? new Error().stack : undefined,
        performance: {
          duration: endTime - startTime,
          memoryDelta: endMemory - startMemory
        }
      };
      
      this.addLogEntry(logEntry);
    });
    
    return result;
  };
  
  // 计算状态差异
  private calculateStateDiffs(oldState: any, newState: any) {
    const diffs: Array<{
      path: string[];
      oldValue: any;
      newValue: any;
      diff: any;
    }> = [];
    
    // 递归比较对象
    const compare = (obj1: any, obj2: any, path: string[] = []) => {
      if (obj1 === obj2) return;
      
      if (typeof obj1 !== 'object' || typeof obj2 !== 'object' || 
          obj1 === null || obj2 === null) {
        // 基本类型或null,直接记录变化
        diffs.push({
          path: [...path],
          oldValue: obj1,
          newValue: obj2,
          diff: { type: 'replace', from: obj1, to: obj2 }
        });
        return;
      }
      
      // 比较对象键
      const allKeys = new Set([...Object.keys(obj1), ...Object.keys(obj2)]);
      allKeys.forEach(key => {
        if (obj1[key] !== obj2[key]) {
          compare(obj1[key], obj2[key], [...path, key]);
        }
      });
    };
    
    compare(oldState, newState);
    return diffs;
  }
  
  // 添加日志条目
  private addLogEntry(entry: AuditLogEntry) {
    this.logs.push(entry);
    
    // 限制日志数量
    if (this.logs.length > this.maxLogs) {
      this.logs = this.logs.slice(-this.maxLogs);
    }
    
    // 触发监听器
    this.notifyListeners(entry);
    
    // 可选:发送到远程服务器
    if (this.shouldSendToServer(entry)) {
      this.sendToAnalytics(entry);
    }
  }
  
  // 时间旅行:恢复到特定时间点的状态
  timeTravelTo(timestamp: number) {
    // 找到指定时间前的最后一条日志
    const logsUpToTime = this.logs.filter(log => log.timestamp <= timestamp);
    
    // 重新应用所有action(跳过审计)
    this.isEnabled = false;
    
    // 从初始状态开始
    let state = this.getInitialState();
    
    logsUpToTime.forEach(log => {
      // 重新执行action
      state = this.applyActionToState(state, log.action);
    });
    
    this.isEnabled = true;
    
    return state;
  }
  
  // 查询审计日志
  queryLogs(options: {
    startTime?: number;
    endTime?: number;
    actionTypes?: string[];
    statePaths?: string[];
    userId?: string;
    minDuration?: number;
  }) {
    return this.logs.filter(log => {
      if (options.startTime && log.timestamp < options.startTime) return false;
      if (options.endTime && log.timestamp > options.endTime) return false;
      if (options.actionTypes && !options.actionTypes.includes(log.action.type)) return false;
      if (options.statePaths && !this.pathMatches(log.stateChange.path, options.statePaths)) return false;
      if (options.userId && log.context.userId !== options.userId) return false;
      if (options.minDuration && (!log.performance || log.performance.duration < options.minDuration)) return false;
      return true;
    });
  }
  
  // 可视化审计轨迹
  visualizeAuditTrail(logs: AuditLogEntry[]) {
    return {
      timeline: logs.map(log => ({
        time: new Date(log.timestamp).toISOString(),
        action: log.action.type,
        path: log.stateChange.path.join('.'),
        duration: log.performance?.duration || 0
      })),
      summary: {
        totalChanges: logs.length,
        byActionType: this.groupByActionType(logs),
        byStatePath: this.groupByStatePath(logs),
        performanceStats: this.calculatePerformanceStats(logs)
      }
    };
  }
}

审计策略

  1. 生产环境 :只记录关键业务操作和错误
  2. 开发环境 :记录所有状态变更,包含调用栈
  3. 采样率 :高频操作按采样率记录
  4. 敏感信息 :自动脱敏密码、token等敏感数据
  5. 存储策略 :内存存储近期日志,IndexedDB存储历史日志,远程服务器存储关键审计

在AI实时协作编辑中,如何用CRDT解决多用户同时修改Prompt的冲突?

场景 :在"印客学院"的团队协作功能中,多名教师同时编辑同一个AI提示词(Prompt),需要实时看到彼此的修改,并自动解决编辑冲突。

核心答案 :使用 无冲突复制数据类型 (CRDT)作为底层数据结构,确保分布式系统的一致性。对于文本协作,使用 操作转换 (OT)或 CRDT文本类型 (如Yjs、Automerge)。每个用户的编辑都转换为可交换、可合并的操作,最终所有副本收敛到相同状态。

Yjs实现方案

import * as Y from 'yjs';
import { WebsocketProvider } from 'y-websocket';

class CollaborativePromptEditor {
  constructor(roomId, userId) {
    // 创建共享文档
    this.ydoc = new Y.Doc();
    this.ytext = this.ydoc.getText('prompt');
    
    // 连接协作服务器
    this.provider = new WebsocketProvider(
      'wss://collab.inke.academy',
      roomId,
      this.ydoc
    );
    
    // 监听远程更新
    this.ytext.observe(event => {
      this.handleRemoteChanges(event);
    });
    
    // 绑定到文本编辑器
    this.bindToEditor();
  }
  
  // 处理本地编辑
  handleLocalEdit(delta) {
    // delta格式: { insert: 'text' } 或 { delete: 1 } 或 { retain: 1 }
    Y.transact(this.ydoc, () => {
      let index = 0;
      delta.forEach(op => {
        if (op.insert) {
          this.ytext.insert(index, op.insert);
          index += op.insert.length;
        } else if (op.delete) {
          this.ytext.delete(index, op.delete);
        } else if (op.retain) {
          index += op.retain;
        }
      });
    });
  }
  
  // 处理远程更新
  handleRemoteChanges(event) {
    // 应用远程更改到本地UI
    event.delta.forEach(change => {
      if (change.insert) {
        // 在UI中插入文本
        this.editor.insertText(change.insert, {
          from: event.index,
          attributes: change.attributes
        });
      } else if (change.delete) {
        // 在UI中删除文本
        this.editor.deleteText(event.index, event.index + change.delete);
      }
    });
    
    // 高亮其他用户的选区
    this.updateUserSelections();
  }
  
  // 显示其他用户的光标和选区
  updateUserSelections() {
    const awareness = this.provider.awareness;
    
    // 获取所有在线用户状态
    const states = awareness.getStates();
    
    states.forEach((state, clientId) => {
      if (clientId !== this.provider.awareness.clientID && state.selection) {
        // 在编辑器中高亮其他用户的光标
        this.editor.highlightSelection(state.user, state.selection);
      }
    });
  }
  
  // 共享AI生成结果
  async generateAndShare() {
    const currentPrompt = this.ytext.toString();
    
    // 调用AI生成
    const result = await this.callAI(currentPrompt);
    
    // 将结果插入共享文档
    Y.transact(this.ydoc, () => {
      this.ytext.insert(this.ytext.length, `\n\nAI生成结果:\n${result}`);
    });
    
    return result;
  }
  
  // 解决特殊冲突:多人同时重命名变量
  resolveVariableRenameConflicts() {
    // 检测变量重命名模式
    const text = this.ytext.toString();
    const variableRegex = /\{\{(\w+)\}\}/g;
    const variables = new Set();
    let match;
    
    while ((match = variableRegex.exec(text)) !== null) {
      variables.add(match[1]);
    }
    
    // 如果有冲突的变量名,使用命名空间解决
    const conflicts = this.findVariableConflicts(variables);
    
    conflicts.forEach(conflict => {
      // 为每个用户添加前缀
      Y.transact(this.ydoc, () => {
        const fullText = this.ytext.toString();
        const userPrefix = `user_${conflict.userId}_`;
        
        // 替换变量名
        const newText = fullText.replace(
          new RegExp(`\\{\\{${conflict.varName}\\}\\}`, 'g'),
          `{{${userPrefix}${conflict.varName}}}`
        );
        
        // 更新文档
        this.ytext.delete(0, this.ytext.length);
        this.ytext.insert(0, newText);
      });
    });
  }
}

CRDT类型选择

  1. 文本编辑 :Y.Text、Automerge.Text
  2. JSON对象 :Y.Map、Automerge.Map
  3. 数组操作 :Y.Array、Automerge.List
  4. 富文本 :Y.XmlFragment、Quill with CRDT

冲突解决策略

  1. 最后写入获胜 :简单场景使用时间戳
  2. 操作转换 :文本编辑使用OT算法
  3. CRDT合并 :自动合并无冲突
  4. 语义解决 :特定业务规则(如变量重命名)
  5. 人工干预 :无法自动解决时提示用户

如何用Valtio的代理机制实现AI配置对象的响应式监听,并自动触发相关副作用?

场景 :在"印客学院"的AI实验平台中,用户通过表单调整AI参数(温度、top_p等),需要实时预览参数效果,并在参数变化时自动触发相关计算。

核心答案 :Valtio使用Proxy实现响应式状态。将AI配置对象包装为 proxy ,使用 subscribe 监听变化,使用 snapshot 获取不可变状态,使用 derive 创建派生状态。副作用通过 subscribewatch 自动触发。

实现方案

import { proxy, subscribe, snapshot, derive, watch } from 'valtio';

// 1. 创建响应式AI配置
const aiConfig = proxy({
  // 模型参数
  model: 'gpt-4',
  temperature: 0.7,
  top_p: 0.9,
  max_tokens: 2000,
  frequency_penalty: 0,
  presence_penalty: 0,
  
  // 高级参数
  stop_sequences: ['\n\n', 'Human:', 'AI:'],
  best_of: 1,
  
  // UI状态
  isAdvancedMode: false,
  lastUpdated: Date.now()
});

// 2. 创建派生状态(自动计算)
const derivedConfig = derive({
  // 计算创造力评分
  creativityScore: (get) => {
    const config = get(aiConfig);
    return (config.temperature * 0.6 + config.top_p * 0.4) * 100;
  },
  
  // 计算预估成本
  estimatedCost: (get) => {
    const config = get(aiConfig);
    const modelRates = {
      'gpt-4': 0.03,
      'gpt-3.5-turbo': 0.002,
      'claude-3': 0.025
    };
    return (config.max_tokens / 1000) * (modelRates[config.model] || 0.01);
  },
  
  // 参数有效性检查
  validation: (get) => {
    const config = get(aiConfig);
    const errors = [];
    
    if (config.temperature < 0 || config.temperature > 2) {
      errors.push('温度必须在0-2之间');
    }
    
    if (config.top_p <= 0 || config.top_p > 1) {
      errors.push('top_p必须在0-1之间');
    }
    
    if (config.max_tokens < 1 || config.max_tokens > 8000) {
      errors.push('max_tokens必须在1-8000之间');
    }
    
    return {
      isValid: errors.length === 0,
      errors
    };
  }
});

// 3. 订阅状态变化
const unsubscribe = subscribe(aiConfig, (ops) => {
  // ops包含所有变更操作
  ops.forEach(op => {
    console.log(`字段 ${op.path.join('.')} 从 ${op.oldValue} 变为 ${op.value}`);
    
    // 根据变更字段触发不同副作用
    switch (op.path[0]) {
      case 'temperature':
        // 温度变化时,重新计算示例输出
        updateExampleOutput();
        break;
        
      case 'model':
        // 模型变化时,加载对应模型的默认参数
        loadModelDefaults(op.value);
        break;
        
      case 'max_tokens':
        // token限制变化时,验证当前内容长度
        validateContentLength();
        break;
    }
  });
  
  // 更新最后修改时间
  aiConfig.lastUpdated = Date.now();
});

// 4. 防抖副作用
let debounceTimer;
subscribe(aiConfig, () => {
  clearTimeout(debounceTimer);
  debounceTimer = setTimeout(() => {
    // 保存到本地存储
    saveConfigToStorage(snapshot(aiConfig));
    
    // 发送到预览服务
    sendToPreviewService(snapshot(aiConfig));
  }, 500);
});

// 5. 条件监听
watch((get) => {
  const config = get(aiConfig);
  const derived = get(derivedConfig);
  
  // 只在参数有效时触发AI调用
  if (derived.validation.isValid) {
    // 防抖调用AI预览
    getAIPreview(config);
  }
});

// 6. 批量更新
function updateMultipleParams(updates) {
  // 使用proxy的批量更新
  Object.assign(aiConfig, updates);
  
  // 或者使用事务
  // 注意:Valtio本身没有事务,但可以通过包装函数实现
  batchUpdate(() => {
    Object.keys(updates).forEach(key => {
      aiConfig[key] = updates[key];
    });
  });
}

// 7. 与React集成
function ConfigPanel() {
  const snap = useSnapshot(aiConfig);
  const derived = useSnapshot(derivedConfig);
  
  return (
    <div>
      <input
        type="range"
        min="0"
        max="2"
        step="0.1"
        value={snap.temperature}
        onChange={(e) => {
          aiConfig.temperature = parseFloat(e.target.value);
        }}
      />
      <div>Creativity: {derived.creativityScore.toFixed(0)}%</div>
      {!derived.validation.isValid && (
        <div className="errors">
          {derived.validation.errors.map(err => (
            <div key={err}>{err}</div>
          ))}
        </div>
      )}
    </div>
  );
}

高级模式

  1. 嵌套代理 :深层对象自动转换为proxy
  2. 数组代理 :数组操作自动追踪
  3. 自定义比较 :控制何时触发更新
  4. 中间件 :拦截和转换更新操作
  5. 时间旅行 :配合 subscribe 记录状态历史
  6. 序列化snapshot 获取不可变状态用于持久化

设计一个"状态压缩"算法,对AI对话历史进行无损压缩

场景 :"印客学院"的AI对话历史可能包含数万条消息,占用大量存储空间和内存。需要在保持完整信息的前提下压缩状态大小。

核心答案 :设计多层次压缩策略:1) 语义压缩 :合并相似消息,删除中间状态;2) 结构压缩 :使用数字ID代替重复字符串;3) 二进制压缩 :对最终数据进行gzip或Brotli压缩。

压缩算法设计

interface CompressionStrategy {
  name: string;
  canApply: (data: any) => boolean;
  compress: (data: any) => any;
  decompress: (compressed: any) => any;
  estimatedRatio: number; // 预估压缩比
}

class ConversationCompressor {
  private strategies: CompressionStrategy[] = [
    // 1. 字符串表压缩
    {
      name: 'string-table',
      canApply: (data) => Array.isArray(data) && data.some(d => typeof d === 'string'),
      compress: (messages) => {
        const stringTable = new Map();
        let nextId = 0;
        
        const compressValue = (value: any): any => {
          if (typeof value === 'string') {
            if (!stringTable.has(value)) {
              stringTable.set(value, nextId++);
            }
            return { $s: stringTable.get(value) };
          }
          if (Array.isArray(value)) {
            return value.map(compressValue);
          }
          if (value && typeof value === 'object') {
            const compressed: any = {};
            for (const [key, val] of Object.entries(value)) {
              compressed[key] = compressValue(val);
            }
            return compressed;
          }
          return value;
        };
        
        const compressedMessages = messages.map(compressValue);
        
        return {
          strings: Array.from(stringTable.keys()),
          messages: compressedMessages
        };
      },
      decompress: (compressed) => {
        const { strings, messages } = compressed;
        
        const decompressValue = (value: any): any => {
          if (value && typeof value === 'object' && '$s' in value) {
            return strings[value.$s];
          }
          if (Array.isArray(value)) {
            return value.map(decompressValue);
          }
          if (value && typeof value === 'object') {
            const decompressed: any = {};
            for (const [key, val] of Object.entries(value)) {
              decompressed[key] = decompressValue(val);
            }
            return decompressed;
          }
          return value;
        };
        
        return messages.map(decompressValue);
      },
      estimatedRatio: 0.6
    },
    
    // 2. 时间戳增量编码
    {
      name: 'delta-encoding',
      canApply: (data) => Array.isArray(data) && data.every(d => d.timestamp),
      compress: (messages) => {
        let lastTimestamp = 0;
        
        return messages.map(msg => {
          const delta = msg.timestamp - lastTimestamp;
          lastTimestamp = msg.timestamp;
          
          return {
            ...msg,
            timestamp: delta, // 存储差值而非绝对值
            ts_abs: undefined
          };
        });
      },
      decompress: (messages) => {
        let currentTime = 0;
        
        return messages.map(msg => {
          currentTime += msg.timestamp;
          return {
            ...msg,
            timestamp: currentTime
          };
        });
      },
      estimatedRatio: 0.8
    },
    
    // 3. 消息内容去重
    {
      name: 'content-deduplication',
      canApply: (data) => Array.isArray(data) && data.length > 10,
      compress: (messages) => {
        const contentMap = new Map();
        const deduplicated: any[] = [];
        
        messages.forEach(msg => {
          if (msg.content) {
            const hash = this.hashContent(msg.content);
            if (!contentMap.has(hash)) {
              contentMap.set(hash, msg.content);
            }
            
            deduplicated.push({
              ...msg,
              content: contentMap.has(hash) ? { $ref: hash } : msg.content
            });
          } else {
            deduplicated.push(msg);
          }
        });
        
        return {
          contents: Object.fromEntries(contentMap),
          messages: deduplicated
        };
      },
      decompress: (compressed) => {
        const { contents, messages } = compressed;
        
        return messages.map(msg => {
          if (msg.content && msg.content.$ref) {
            return {
              ...msg,
              content: contents[msg.content.$ref]
            };
          }
          return msg;
        });
      },
      estimatedRatio: 0.5
    },
    
    // 4. 删除中间状态
    {
      name: 'remove-intermediate-states',
      canApply: (data) => Array.isArray(data) && data.some(d => d.status === 'streaming'),
      compress: (messages) => {
        // 只保留最终状态,删除流式生成的中间状态
        const result: any[] = [];
        let lastMessage: any = null;
        
        for (const msg of messages) {
          if (msg.status === 'streaming') {
            if (lastMessage && lastMessage.id === msg.id) {
              // 更新同一个消息的内容
              lastMessage.content = msg.content;
            } else {
              // 新消息
              lastMessage = { ...msg };
              result.push(lastMessage);
            }
          } else {
            result.push(msg);
            lastMessage = null;
          }
        }
        
        return result;
      },
      decompress: (messages) => messages, // 不可逆压缩
      estimatedRatio: 0.3
    }
  ];
  
  // 智能压缩:选择最佳策略组合
  async compressConversation(messages: any[], options = {}): Promise<{
    data: any;
    metadata: {
      originalSize: number;
      compressedSize: number;
      ratio: number;
      strategies: string[];
      lossless: boolean;
    }
  }> {
    const originalSize = JSON.stringify(messages).length;
    let currentData = messages;
    const usedStrategies: string[] = [];
    let isLossless = true;
    
    // 应用所有适用的策略
    for (const strategy of this.strategies) {
      if (strategy.canApply(currentData)) {
        try {
          currentData = strategy.compress(currentData);
          usedStrategies.push(strategy.name);
          
          // 检查是否为有损压缩
          if (strategy.name === 'remove-intermediate-states') {
            isLossless = false;
          }
        } catch (error) {
          console.warn(`策略 ${strategy.name} 失败:`, error);
        }
      }
    }
    
    // 最终二进制压缩
    const jsonStr = JSON.stringify(currentData);
    const compressedBinary = await this.binaryCompress(jsonStr);
    
    const compressedSize = compressedBinary.byteLength;
    
    return {
      data: compressedBinary,
      metadata: {
        originalSize,
        compressedSize,
        ratio: compressedSize / originalSize,
        strategies: usedStrategies,
        lossless: isLossless
      }
    };
  }
  
  // 二进制压缩
  private async binaryCompress(str: string): Promise<ArrayBuffer> {
    // 使用Compression Streams API
    const encoder = new TextEncoder();
    const encoded = encoder.encode(str);
    
    const cs = new CompressionStream('gzip');
    const writer = cs.writable.getWriter();
    writer.write(encoded);
    writer.close();
    
    const compressed = await new Response(cs.readable).arrayBuffer();
    return compressed;
  }
  
  // 解压
  async decompressConversation(compressed: ArrayBuffer, metadata: any): Promise<any> {
    // 二进制解压
    const cs = new DecompressionStream('gzip');
    const writer = cs.writable.getWriter();
    writer.write(compressed);
    writer.close();
    
    const decompressed = await new Response(cs.readable).arrayBuffer();
    const jsonStr = new TextDecoder().decode(decompressed);
    let data = JSON.parse(jsonStr);
    
    // 按相反顺序应用解压策略
    const reversedStrategies = [...this.strategies].reverse();
    
    for (const strategy of reversedStrategies) {
      if (metadata.strategies.includes(strategy.name)) {
        data = strategy.decompress(data);
      }
    }
    
    return data;
  }
}

压缩策略选择

  1. 实时压缩 :每次新增消息时增量压缩
  2. 离线压缩 :定时对历史数据批量压缩
  3. 自适应压缩 :根据数据类型选择最佳算法
  4. 分级存储 :热数据保持可读,冷数据高度压缩
  5. 压缩指标 :监控压缩比、耗时、CPU使用率

在AI工作流引擎中,如何用BPMN或Workflow模型定义状态流转,并前端可视化执行过程?

场景 :"印客学院"的AI工作流设计器,允许用户通过拖拽方式定义复杂的AI处理流程(如:文本输入 → 情感分析 → 条件分支 → 不同回复),并实时可视化执行过程。

核心答案 :使用BPMN 2.0标准定义工作流模型,将AI能力封装为可重用的"服务任务"。前端使用BPMN.js库渲染和编辑工作流图,通过状态机跟踪每个节点的执行状态,实时可视化执行进度、数据流向和错误信息。

实现方案

// 1. BPMN模型定义
const workflowDefinition = {
  processId: 'customer-service-workflow',
  name: '印客学院客服AI工作流',
  nodes: [
    {
      id: 'start',
      type: 'startEvent',
      name: '开始',
      next: 'analyze_sentiment'
    },
    {
      id: 'analyze_sentiment',
      type: 'serviceTask',
      name: '情感分析',
      implementation: 'sentiment-analysis-service',
      config: {
        model: 'emotion-detector-v2',
        language: 'zh-CN'
      },
      next: 'decision_branch'
    },
    {
      id: 'decision_branch',
      type: 'exclusiveGateway',
      name: '情感判断',
      rules: [
        { condition: 'sentiment === "positive"', next: 'positive_response' },
        { condition: 'sentiment === "negative"', next: 'escalate_to_human' },
        { default: 'neutral_response' }
      ]
    },
    {
      id: 'positive_response',
      type: 'serviceTask',
      name: '积极回应',
      implementation: 'ai-chat-service',
      config: {
        prompt_template: '用户情绪积极,给予肯定和鼓励回复',
        temperature: 0.7
      },
      next: 'end'
    },
    {
      id: 'escalate_to_human',
      type: 'userTask',
      name: '转人工客服',
      assignee: 'customer_service_team',
      next: 'end'
    },
    {
      id: 'end',
      type: 'endEvent',
      name: '结束'
    }
  ],
  variables: {
    input_text: '',
    sentiment: '',
    confidence: 0,
    response: ''
  }
};

// 2. 工作流执行引擎
class WorkflowEngine {
  constructor(definition) {
    this.definition = definition;
    this.execution = {
      instanceId: `wf_${Date.now()}`,
      currentNode: null,
      history: [],
      variables: { ...definition.variables },
      status: 'idle' // idle, running, paused, completed, error
    };
    
    // 节点执行器映射
    this.nodeExecutors = {
      startEvent: this.executeStartEvent,
      serviceTask: this.executeServiceTask,
      userTask: this.executeUserTask,
      exclusiveGateway: this.executeExclusiveGateway,
      endEvent: this.executeEndEvent
    };
  }
  
  // 执行工作流
  async start(initialData = {}) {
    this.execution.status = 'running';
    this.execution.variables = { ...this.execution.variables, ...initialData };
    
    // 找到开始节点
    const startNode = this.definition.nodes.find(n => n.type === 'startEvent');
    if (!startNode) throw new Error('No start event found');
    
    // 执行开始节点
    await this.executeNode(startNode);
    
    return this.execution.instanceId;
  }
  
  // 执行单个节点
  async executeNode(node) {
    console.log(`执行节点: ${node.name} (${node.type})`);
    
    // 更新当前节点
    this.execution.currentNode = node;
    
    // 记录历史
    this.execution.history.push({
      nodeId: node.id,
      nodeName: node.name,
      timestamp: Date.now(),
      variables: { ...this.execution.variables },
      status: 'started'
    });
    
    try {
      // 获取执行器
      const executor = this.nodeExecutors[node.type];
      if (!executor) {
        throw new Error(`No executor for node type: ${node.type}`);
      }
      
      // 执行节点
      const result = await executor.call(this, node);
      
      // 更新历史记录状态
      const lastHistory = this.execution.history[this.execution.history.length - 1];
      lastHistory.status = 'completed';
      lastHistory.completedAt = Date.now();
      lastHistory.result = result;
      
      // 确定下一个节点
      let nextNodeId = null;
      
      if (node.type === 'exclusiveGateway') {
        // 条件网关:根据条件选择分支
        nextNodeId = this.evaluateGatewayRules(node, result);
      } else if (node.next) {
        // 直接跳转
        nextNodeId = node.next;
      }
      
      // 执行下一个节点
      if (nextNodeId) {
        const nextNode = this.definition.nodes.find(n => n.id === nextNodeId);
        if (nextNode) {
          // 短暂延迟以便UI更新
          await new Promise(resolve => setTimeout(resolve, 100));
          await this.executeNode(nextNode);
        }
      } else if (node.type === 'endEvent') {
        // 到达结束节点
        this.execution.status = 'completed';
        console.log('工作流执行完成');
      }
      
    } catch (error) {
      console.error(`节点执行失败: ${node.name}`, error);
      
      // 更新历史记录
      const lastHistory = this.execution.history[this.execution.history.length - 1];
      lastHistory.status = 'error';
      lastHistory.error = error.message;
      
      this.execution.status = 'error';
      throw error;
    }
  }
  
  // 执行AI服务任务
  async executeServiceTask(node) {
    const { implementation, config } = node;
    
    switch (implementation) {
      case 'sentiment-analysis-service':
        // 调用情感分析AI
        return await this.callSentimentAnalysis(this.execution.variables.input_text, config);
        
      case 'ai-chat-service':
        // 调用对话AI
        return await this.callAIChat(this.execution.variables.input_text, config);
        
      default:
        throw new Error(`Unknown service implementation: ${implementation}`);
    }
  }
  
  // 评估网关条件
  evaluateGatewayRules(node, context) {
    for (const rule of node.rules) {
      if (rule.condition) {
        // 使用安全的表达式求值
        try {
          const condition = new Function('ctx', `return ${rule.condition}`);
          if (condition(context)) {
            return rule.next;
          }
        } catch (error) {
          console.warn(`条件求值失败: ${rule.condition}`, error);
        }
      } else if (rule.default) {
        return rule.default;
      }
    }
    
    throw new Error(`No matching rule found for gateway: ${node.id}`);
  }
}

// 3. 前端可视化
class WorkflowVisualizer {
  constructor(containerId, workflowEngine) {
    this.engine = workflowEngine;
    this.bpmnViewer = new BpmnJS({ container: `#${containerId}` });
    
    // 加载BPMN图
    this.loadDiagram(workflowEngine.definition);
    
    // 监听执行状态变化
    this.subscribeToExecution();
  }
  
  // 将定义转换为BPMN XML
  async loadDiagram(definition) {
    const bpmnXml = this.convertToBPMNXML(definition);
    
    await this.bpmnViewer.importXML(bpmnXml);
    
    // 设置自定义样式和交互
    this.setupCustomOverlays();
  }
  
  // 订阅执行状态更新
  subscribeToExecution() {
    // 监听节点开始
    this.engine.on('nodeStarted', (node) => {
      this.highlightNode(node.id, 'running');
    });
    
    // 监听节点完成
    this.engine.on('nodeCompleted', (node, result) => {
      this.highlightNode(node.id, 'completed');
      this.showNodeResult(node.id, result);
    });
    
    // 监听节点错误
    this.engine.on('nodeError', (node, error) => {
      this.highlightNode(node.id, 'error');
      this.showNodeError(node.id, error);
    });
    
    // 监听变量变化
    this.engine.on('variablesChanged', (variables) => {
      this.updateVariablesPanel(variables);
    });
  }
  
  // 高亮节点
  highlightNode(nodeId, status) {
    const elementRegistry = this.bpmnViewer.get('elementRegistry');
    const element = elementRegistry.get(nodeId);
    
    if (element) {
      const canvas = this.bpmnViewer.get('canvas');
      
      // 移除旧的高亮
      canvas.removeMarker(element, 'highlight-running');
      canvas.removeMarker(element, 'highlight-completed');
      canvas.removeMarker(element, 'highlight-error');
      
      // 添加新状态的高亮
      canvas.addMarker(element, `highlight-${status}`);
    }
  }
  
  // 显示节点结果
  showNodeResult(nodeId, result) {
    // 在节点旁边显示结果
    const overlay = this.createResultOverlay(nodeId, result);
    this.bpmnViewer.get('overlays').add(nodeId, {
      position: { bottom: 0, right: 0 },
      html: overlay
    });
  }
  
  // 实时数据流可视化
  visualizeDataFlow(data) {
    // 使用连线动画显示数据流向
    const connection = this.findConnection(data.fromNode, data.toNode);
    if (connection) {
      this.animateConnection(connection, data.value);
    }
  }
}

可视化特性

  1. 实时高亮 :当前执行节点、已完成的节点、错误节点
  2. 数据流显示 :在连线上显示传递的数据
  3. 变量监控 :实时显示工作流变量的值
  4. 执行历史 :可回放工作流执行过程
  5. 调试控制 :暂停、继续、单步执行
  6. 性能指标 :显示每个节点的执行时间

如何用SWR或React Query实现AI模型列表、价格、可用性等数据的自动缓存与后台刷新?

场景 :"印客学院"的模型选择器需要显示可用AI模型列表、实时价格和可用性状态。这些数据相对静态但可能变化,需要智能缓存避免重复请求,并在后台静默更新。

核心答案 :使用SWR或React Query的缓存策略:1)内存缓存减少请求;2)后台刷新保持数据新鲜;3)智能重试处理错误;4)依赖请求优化性能。

React Query实现

import { useQuery, useQueries, useQueryClient } from '@tanstack/react-query';

// 1. 获取AI模型列表
function useAIModels(options = {}) {
  return useQuery({
    queryKey: ['ai-models'], // 缓存键
    queryFn: async () => {
      const response = await fetch('https://api.inke.academy/models');
      if (!response.ok) throw new Error('Failed to fetch models');
      return response.json();
    },
    // 缓存配置
    staleTime: 5 * 60 * 1000, // 5分钟内数据视为新鲜
    cacheTime: 10 * 60 * 1000, // 10分钟后清理缓存
    
    // 重试策略
    retry: 3,
    retryDelay: (attemptIndex) => Math.min(1000 * 2 ** attemptIndex, 30000),
    
    // 后台刷新
    refetchOnWindowFocus: true, // 窗口聚焦时刷新
    refetchOnMount: true, // 组件挂载时刷新
    refetchOnReconnect: true, // 网络重连时刷新
    
    // 轮询(价格需要更频繁更新)
    refetchInterval: options.refetchInterval || 2 * 60 * 1000, // 每2分钟
    
    // 错误时使用缓存数据
    initialData: () => {
      const cached = localStorage.getItem('inke-models-cache');
      return cached ? JSON.parse(cached) : undefined;
    },
    
    // 成功时更新本地缓存
    onSuccess: (data) => {
      localStorage.setItem('inke-models-cache', JSON.stringify(data));
    }
  });
}

// 2. 获取单个模型的详细信息和价格
function useAIModelDetails(modelId, options = {}) {
  return useQuery({
    queryKey: ['ai-model', modelId],
    queryFn: async () => {
      const [details, pricing, availability] = await Promise.all([
        fetch(`https://api.inke.academy/models/${modelId}`).then(r => r.json()),
        fetch(`https://api.inke.academy/pricing/${modelId}`).then(r => r.json()),
        fetch(`https://api.inke.academy/availability/${modelId}`).then(r => r.json())
      ]);
      
      return { details, pricing, availability };
    },
    // 依赖查询:只有模型列表中有这个模型才获取
    enabled: options.enabled && !!modelId,
    
    // 价格数据需要更频繁更新
    staleTime: 1 * 60 * 1000, // 1分钟
    refetchInterval: 30 * 1000, // 30秒
    
    // 乐观更新占位符
    placeholderData: () => {
      // 从模型列表中获取基本信息作为占位
      const { data: models } = useAIModels();
      const model = models?.find(m => m.id === modelId);
      return model ? {
        details: model,
        pricing: { input: 0, output: 0 },
        availability: { status: 'loading' }
      } : undefined;
    }
  });
}

// 3. 批量获取多个模型信息
function useAIModelsBatch(modelIds) {
  return useQueries({
    queries: modelIds.map(modelId => ({
      queryKey: ['ai-model', modelId],
      queryFn: () => fetchModelDetails(modelId),
      // 批量请求的优化配置
      staleTime: 2 * 60 * 1000,
    }))
  });
}

// 4. 预加载模型数据
function usePreloadAIModels() {
  const queryClient = useQueryClient();
  
  const preload = async (modelIds) => {
    // 预取模型列表
    await queryClient.prefetchQuery({
      queryKey: ['ai-models'],
      queryFn: fetchModels
    });
    
    // 预取热门模型详情
    const popularModels = modelIds.slice(0, 5);
    await Promise.all(
      popularModels.map(modelId =>
        queryClient.prefetchQuery({
          queryKey: ['ai-model', modelId],
          queryFn: () => fetchModelDetails(modelId)
        })
      )
    );
  };
  
  return { preload };
}

// 5. 缓存失效和更新
function useAIModelCacheManager() {
  const queryClient = useQueryClient();
  
  const invalidateModel = (modelId) => {
    // 使单个模型缓存失效
    queryClient.invalidateQueries({ queryKey: ['ai-model', modelId] });
  };
  
  const invalidateAllModels = () => {
    // 使所有模型相关缓存失效
    queryClient.invalidateQueries({ queryKey: ['ai-models'] });
    queryClient.invalidateQueries({ queryKey: ['ai-model'] });
  };
  
  const updateModelPricing = (modelId, newPricing) => {
    // 乐观更新:立即更新UI,然后在后台验证
    queryClient.setQueryData(['ai-model', modelId], (oldData) => {
      if (!oldData) return oldData;
      return {
        ...oldData,
        pricing: newPricing,
        updatedAt: Date.now()
      };
    });
    
    // 在后台同步到服务器
    syncPricingToServer(modelId, newPricing);
  };
  
  return { invalidateModel, invalidateAllModels, updateModelPricing };
}

// 6. 离线支持
function useOfflineAIModels() {
  const queryClient = useQueryClient();
  const isOnline = useNetworkStatus();
  
  const { data, error, isLoading } = useAIModels({
    // 离线时使用缓存
    networkMode: isOnline ? 'online' : 'always'
  });
  
  // 网络恢复时刷新
  useEffect(() => {
    if (isOnline) {
      queryClient.invalidateQueries({ queryKey: ['ai-models'] });
    }
  }, [isOnline, queryClient]);
  
  return { data, error, isLoading, isOnline };
}

SWR实现 (更轻量):

import useSWR, { useSWRConfig } from 'swr';

// 简单的fetcher函数
const fetcher = (url) => fetch(url).then(r => r.json());

function useAIModelsSWR() {
  const { data, error, isLoading, isValidating, mutate } = useSWR(
    'https://api.inke.academy/models',
    fetcher,
    {
      // 重新验证策略
      revalidateOnFocus: true,
      revalidateOnReconnect: true,
      revalidateIfStale: true,
      
      // 轮询
      refreshInterval: 120000, // 2分钟
      
      // 错误重试
      shouldRetryOnError: true,
      errorRetryCount: 3,
      errorRetryInterval: 5000,
      
      // 缓存
      dedupingInterval: 2000, // 2秒内相同请求去重
      focusThrottleInterval: 5000,
      
      // 回退数据
      fallbackData: JSON.parse(localStorage.getItem('inke-models-fallback') || 'null'),
      
      // 成功回调
      onSuccess: (data) => {
        localStorage.setItem('inke-models-fallback', JSON.stringify(data));
      }
    }
  );
  
  // 手动重新验证
  const refresh = () => mutate();
  
  // 乐观更新
  const addModel = async (newModel) => {
    // 立即更新本地数据
    await mutate([...(data || []), newModel], false);
    
    // 在后台同步到服务器
    try {
      await fetch('https://api.inke.academy/models', {
        method: 'POST',
        body: JSON.stringify(newModel)
      });
      // 成功后重新验证
      mutate();
    } catch (error) {
      // 失败时回滚
      mutate(data, false);
    }
  };
  
  return { 
    models: data, 
    error, 
    isLoading, 
    isValidating, 
    refresh, 
    addModel 
  };
}

缓存策略对比

特性 React Query SWR
缓存粒度 精细的键值缓存 基于URL的缓存
后台刷新 支持 支持
依赖查询 优秀 基础
离线支持 优秀 良好
乐观更新 内置支持 手动实现
预加载 优秀 良好

最佳实践

  1. 分层缓存 :模型列表缓存5分钟,价格缓存1分钟
  2. 智能预取 :用户可能查看的模型提前加载
  3. 离线降级 :网络不可用时显示缓存数据
  4. 批量请求 :合并多个模型的请求
  5. 缓存分区 :按租户、用户分组缓存