RAGFlow源码解析-13、十周技术之旅总结

# 双模式切换机制@app.route('/api/chat', methods=['POST'])@timeitdef chat():  # Flask 同步版本    pass@app.route('/api/chat/stream', methods=['POST'])@timeitasync def chat_stream():  # Quart 异步版本    pass

class Base(ABC):    @abstractmethod    def encode(self, texts: list[str]) -> np.ndarray:        passclass OpenAIEmbed(Base):    def encode(self, texts):        return self.client.embeddings.create(            model=self.model_name,            input=texts        ).data[0].embedding

Redis Queue → Task Executor → Component Pipeline → ES/Infinity                ↓           Callback Updates

# 正则匹配变量表达式pat = re.compile(r"\{\{([a-zA-Z:0-9]+@[A-Za-z0-9_.-]+|sys\.[A-Za-z0-9_.]+)\}\}")# 递归路径解析def get_variable_param_value(obj, path):    for key in path.split('.'):        if isinstance(obj, dict):            obj = obj.get(key)        elif isinstance(obj, list):            obj = obj[int(key)]        else:            obj = getattr(obj, key, None)    return obj

# 流式工具调用累积final_tool_calls = {}for delta in response:    if delta.tool_calls:        for tool_call in delta.tool_calls:            idx = tool_call.index            if idx not in final_tool_calls:                final_tool_calls[idx] = tool_call            else:                # 累积 arguments                final_tool_calls[idx].function.arguments += tool_call.function.arguments

# 统一错误处理def get_data_error_result(message):    return {"code": 400, "message": message}# 统一成功响应def get_json_result(data):    return {"code": 0, "data": data}

class RetryingPooledMySQLDatabase(PooledMySQLDatabase):    def execute_sql(self, sql, params=None):        for attempt in range(self.max_retries + 1):            try:                return super().execute_sql(sql, params)            except (OperationalError, InterfaceError) as e:                if should_retry and attempt < self.max_retries:                    time.sleep(self.retry_delay * (2 ** attempt))  # 指数退避                else:                    raise

Base (抽象基类)  ├── GptV4 (OpenAI)  ├── GeminiCV (Google)  ├── QWenCV (阿里 - 支持视频处理)  ├── AnthropicCV (Claude - 支持扩展思考)  └── OllamaCV (本地部署)

# 组件工厂cpn = component_class("LLM")(canvas, cpn_id, param)# 模型工厂model = MODEL_MAP[factory_name](api_key, model_name, **kwargs)

# 解析器策略PARSER_MAP = {    "naive": Naive,    "paper": Paper,    "book": Book,    "laws": Laws}parser = PARSER_MAP[parser_id](**config)

class ComponentBase:    async def invoke(self, **kwargs):        # 模板方法：固定流程        self._pre_process()        result = await self._execute()        self._post_process()        return result    def _pre_process(self): pass    def _execute(self): raise NotImplementedError    def _post_process(self): pass

# OpenAI 格式return response.choices[0].message.content# Anthropic 格式return response.content[0].text# Gemini 格式return response.text

async def async_chat_streamly(self, ...):    async for chunk in response:        yield chunk  # 流式通知

Frontend Canvas UI       ↓ (DSL JSON)    Graph/Canvas       ↓Component Pipeline       ↓  LLM/Retrieval/Tools       ↓    Output Stream

LLM Response (stream)    ↓Delta Tool Call (index=0, args="...")    ↓Accumulate: final_tool_calls[0].args += "..."    ↓Stream Complete: Parse & Execute    ↓Tool Result → History → Continue

Document Chunks      ↓  Entity Extraction (LLM)      ↓  Relation Extraction (LLM)      ↓  Graph Storage (Neo4j/ES)      ↓  Graph Query + Vector Retrieval

class Base(ABC):    def describe(self, image): ...    def transcription(self, audio): ...# 统一调用model = GptV4(api_key, "gpt-4-vision-preview")text, tokens = model.describe(image_bytes)

# 指数退避重试time.sleep(retry_delay * (2 ** attempt))# 自动重连def _handle_connection_loss(self):    self.close()    self.connect()

# 1. 预过滤（减少向量计算量）filters = {"kb_id": kb_id, "status": "valid"}# 2. HNSW 索引优化index_params = {    "M": 16,  # 连接数    "ef_construction": 200  # 构建时搜索范围}# 3. Reranking 仅对 Top 100ranks = retrieval(question, top_k=100)reranked = rerank(question, ranks[:100])

# 并发执行独立组件import asyncioasync def run_parallel():    results = await asyncio.gather(        retrieval_0.invoke(),        categorize_0.invoke(),        retrieval_1.invoke()    )    return results

# Redis 缓存检索结果cache_key = f"retrieval:{hash(question)}:{kb_ids}"cached = REDIS_CONN.get(cache_key)if cached:    return json.loads(cached)result = retrieval(question, kb_ids)REDIS_CONN.setex(cache_key, 3600, json.dumps(result))

# 错误写法cursor = db.execute_sql(sql, params, commit=True)# 正确写法cursor = db.execute_sql(sql, params)# Peewee 自动提交

# 错误写法await component.invoke(**kwargs)# 正确写法async for chunk in component.invoke(**kwargs):    yield chunk  # 必须消费

import json_repair# 容错解析try:    args = json.loads(tool_call.function.arguments)except:    args = json_repair.loads(tool_call.function.arguments)

# 更严格的正则pat = re.compile(r"\{\{([a-zA-Z:0-9]+@[A-Za-z0-9_.-]+|sys\.[A-Za-z0-9_.]+|env\.[A-Za-z0-9_.]+)\}\}")# 提供转义语法# 用户输入 \{\{literal\}\} → {{literal}}