# 來自 backend/airweave/platform/sync/orchestrator.py class SyncOrchestrator: """協調資料同步的核心類別，採用拉取式並發控制""" def __init__(self, entity_processor, worker_pool, sync_context): self.entity_processor = entity_processor self.worker_pool = worker_pool self.sync_context = sync_context self.stream_buffer_size = 100 # 緩衝區大小 async def _process_entities(self): """使用拉取式並發控制處理實體資料""" async with AsyncSourceStream( self.sync_context.source.generate_entities(), queue_size=self.stream_buffer_size ) as stream: async for entity in stream.get_entities(): if getattr(entity, "should_skip", False): await self.sync_context.progress.increment("skipped") continue # 提交至工作池進行並發處理 task = await self.worker_pool.submit( self.entity_processor.process, entity=entity, source_node=source_node, sync_context=self.sync_context )

// 來自 frontend/src/lib/stores/collections.ts import { create } from 'zustand'; interface CollectionsState { collections: Collection[]; isLoading: boolean; sourceConnections: Record<string, SourceConnection[]>; fetchCollections: (forceRefresh?: boolean) => Promise<Collection[]>; subscribeToEvents: () => () => void; clearCollections: () => void; } export const useCollectionsStore = create<CollectionsState>((set, get) => ({ collections: [], isLoading: false, sourceConnections: {}, fetchCollections: async (forceRefresh = false) => { // 智能快取：如果已有資料且非強制刷新，直接返回快取 const { collections, isLoading } = get(); if (collections.length > 0 && !forceRefresh) { console.log("🔍 使用快取資料，略過 API 呼叫"); return collections; } set({ isLoading: true, error: null }); const response = await apiClient.get('/collections'); // 處理回應資料... } }));

# 來自 backend/airweave/platform/transformers/optimized_file_chunker.py async def _try_chunk_size( file: FileEntity, text_content: str, chunk_size: int ) -> tuple[bool, list[ChunkEntity]]: """嘗試特定大小的分塊，驗證是否符合 OpenAI 限制""" final_chunk_texts = await _chunk_text_adaptive(text_content, chunk_size) # 建立父實體 parent = FileParentClass(**file.model_dump()) all_chunks_fit = True test_chunks = [] for i, chunk_text in enumerate(final_chunk_texts): chunk = _create_chunk_entity(file, parent, chunk_text, i) # 檢查序列化後的實際大小 actual_size = calculate_entity_token_size(chunk) if actual_size > OPENAI_TOKEN_LIMIT: logger.warning(f"分塊 {i + 1} 過大: {actual_size} tokens") all_chunks_fit = False break else: test_chunks.append(chunk) return all_chunks_fit, [parent] + test_chunks if all_chunks_fit else []