RAGFlow源码解析-5、RAG Pipeline核心流程深度解析(第三周)

class Pipeline(Graph):  # 第28行：继承Graph，获得组件编排能力    def __init__(self, dsl: str|dict, tenant_id=None, doc_id=None, task_id=None, flow_id=None):        # 第30-31行：DSL可以是字典或字符串，自动转换为JSON字符串        if isinstance(dsl, dict):            dsl = json.dumps(dsl, ensure_ascii=False)        # 第32行：调用父类Graph.__init__初始化组件图        super().__init__(dsl, tenant_id, task_id)        # 第33-34行：调试文档ID特殊处理（Canvas调试模式）        if doc_id == CANVAS_DEBUG_DOC_ID:            doc_id = None        self._doc_id = doc_id  # 文档ID（用于文档处理任务）        self._flow_id = flow_id  # 流程ID（用于日志追踪）        # 第37-41行：关联知识库ID，如果文档不属于任何知识库则无效化doc_id        self._kb_id = None        if self._doc_id:            self._kb_id = DocumentService.get_knowledgebase_id(doc_id)            if not self._kb_id:                self._doc_id = None  # 文档不属于任何知识库，取消处理

def callback(self, component_name: str, progress: float | int | None = None, message: str = "") -> None:    from common.exceptions import TaskCanceledException    # 第45行：Redis日志键，格式为"{flow_id}-{task_id}-logs"    log_key = f"{self._flow_id}-{self.task_id}-logs"    timestamp = timer()  # 第46行：获取当前时间戳    # 第47-49行：检查任务是否被取消，如果取消则标记进度为-1    if has_canceled(self.task_id):        progress = -1        message += "[CANCEL]"    try:        # 第51-52行：从Redis获取已有日志对象        bin = REDIS_CONN.get(log_key)        obj = json.loads(bin.encode("utf-8"))        if obj:            # 第54-63行：如果是同一组件的后续调用，追加trace记录            if obj[-1]["component_id"] == component_name:                obj[-1]["trace"].append({                    "progress": progress,  # 进度值（0-1或-1表示错误）                    "message": message,  # 进度消息                    "datetime": datetime.datetime.now().strftime("%H:%M:%S"),                    "timestamp": timestamp,                    "elapsed_time": timestamp - obj[-1]["trace"][-1]["timestamp"],  # 计算耗时                })            else:                # 第65-70行：如果是新组件，创建新的component记录                obj.append({                    "component_id": component_name,                    "trace": [{"progress": progress, "message": message, ...}],                })        else:            # 第72-77行：首次调用，创建初始日志结构            obj = [{                "component_id": component_name,                "trace": [{"progress": progress, "message": message, ...}],            }]        # 第78-95行：如果是文档处理任务，更新数据库进度        if component_name != "END" and self._doc_id and self.task_id:            percentage = 1.0 / len(self.components.items())  # 每个组件平均进度占比            finished = 0.0            # 第81-88行：累加所有组件的进度            for o in obj:                for t in o["trace"]:                    if t["progress"] < 0:  # 遇到错误，立即终止                        finished = -1                        break                if finished < 0:                    break                finished += o["trace"][-1]["progress"] * percentage            # 第90-95行：构建进度消息并更新数据库            msg = ""            if len(obj[-1]["trace"]) == 1:  # 新组件首次调用                msg += f"\n-------------------------------------\n[{self.get_component_name(o['component_id'])}]:\n"            t = obj[-1]["trace"][-1]            msg += "%s: %s\n" % (t["datetime"], t["message"])            TaskService.update_progress(self.task_id, {"progress": finished, "progress_msg": msg})        # 第96-97行：如果是END组件且无doc_id，记录最终DSL状态        elif component_name == "END" and not self._doc_id:            obj[-1]["trace"][-1]["dsl"] = json.loads(str(self))        # 第98行：将日志对象存入Redis，有效期30分钟        REDIS_CONN.set_obj(log_key, obj, 60 * 30)    except Exception as e:        logging.exception(e)  # 第101行：异常仅记录日志，不中断流程    # 第103-104行：如果任务取消，抛出异常终止Pipeline    if has_canceled(self.task_id):        raise TaskCanceledException(message)

async def run(self, **kwargs):    # 第118-122行：初始化Redis日志为空数组，有效期10分钟    log_key = f"{self._flow_id}-{self.task_id}-logs"    try:        REDIS_CONN.set_obj(log_key, [], 60 * 10)    except Exception as e:        logging.exception(e)    self.error = ""  # 第123行：清空错误信息    # 第124-130行：如果path为空，初始化第一个组件为"File"    if not self.path:        self.path.append("File")  # File组件是默认入口        cpn_obj = self.get_component_obj(self.path[0])        await cpn_obj.invoke(**kwargs)  # 异步调用第一个组件        # 第128-130行：如果第一个组件报错，记录错误并回调        if cpn_obj.error():            self.error = "[ERROR]" + cpn_obj.error()            self.callback(cpn_obj.component_name, -1, self.error)    # 第132-136行：如果是文档处理任务，更新初始进度（0-5%随机值）    if self._doc_id:        TaskService.update_progress(self.task_id, {            "progress": random.randint(0, 5) / 100.0,  # 随机0-5%进度            "progress_msg": "Start the pipeline...",            "begin_at": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")        })    # 第138-141行：获取当前path最后一个组件，并扩展其下游组件    idx = len(self.path) - 1    cpn_obj = self.get_component_obj(self.path[idx])    idx += 1    self.path.extend(cpn_obj.get_downstream())  # 获取下游组件列表    # 第143-164行：循环执行下游组件链    while idx < len(self.path) and not self.error:        last_cpn = self.get_component_obj(self.path[idx - 1])  # 上一个组件        cpn_obj = self.get_component_obj(self.path[idx])  # 当前组件        # 第147-153行：定义异步invoke函数，传递上一个组件的output        async def invoke():            nonlocal last_cpn, cpn_obj            await cpn_obj.invoke(**last_cpn.output())  # 当前组件接收上一个组件的输出        # 第155-157行：创建异步任务并执行        tasks = []        tasks.append(asyncio.create_task(invoke()))        await asyncio.gather(*tasks)  # 等待任务完成        # 第159-162行：检查当前组件是否报错        if cpn_obj.error():            self.error = "[ERROR]" + cpn_obj.error()            self.callback(cpn_obj._id, -1, self.error)            break  # 报错则终止循环        # 第163-164行：更新索引，继续扩展下游组件        idx += 1        self.path.extend(cpn_obj.get_downstream())    # 第166行：调用END回调，记录最终状态（成功progress=1，失败progress=-1）    self.callback("END", 1 if not self.error else -1, json.dumps(self.get_component_obj(self.path[-1]).output(), ensure_ascii=False))    # 第168-169行：如果无错误，返回最后一个组件的输出    if not self.error:        return self.get_component_obj(self.path[-1]).output()    # 第171-173行：如果错误，更新数据库进度为-1，返回空字典    TaskService.update_progress(self.task_id, {        "progress": -1,        "progress_msg": f"[ERROR]: {self.error}"    })    return {}  # 第174行：错误时返回空字典

class Base(ABC):  # 第37行：抽象基类，定义统一接口    def __init__(self, key, model_name, **kwargs):        """        Constructor for abstract base class.        Parameters are accepted for interface consistency but are not stored.        Subclasses should implement their own initialization as needed.        """        pass  # 第44行：基类不存储参数，由子类自行实现    def encode(self, texts: list):  # 第46-47行：文档编码方法        raise NotImplementedError("Please implement encode method!")    def encode_queries(self, text: str):  # 第49-50行：查询编码方法        raise NotImplementedError("Please implement encode method!")

class BuiltinEmbed(Base):  # 第53行：继承Base，实现内置模型    _FACTORY_NAME = "Builtin"  # 第54行：工厂名称标识    MAX_TOKENS = {"Qwen/Qwen3-Embedding-0.6B": 30000, "BAAI/bge-m3": 8000, "BAAI/bge-small-en-v1.5": 500}  # 第55行：模型token限制映射    _model = None  # 第56行：全局模型实例（单例）    _model_name = ""  # 第57行：模型名称    _max_tokens = 500  # 第58行：默认token限制    _model_lock = threading.Lock()  # 第59行：线程锁，保护单例初始化    def __init__(self, key, model_name, **kwargs):        # 第62-63行：从settings读取配置        logging.info(f"Initialize BuiltinEmbed according to settings.EMBEDDING_CFG: {settings.EMBEDDING_CFG}")        embedding_cfg = settings.EMBEDDING_CFG        # 第64-68行：如果全局模型未初始化且启用TEI服务，创建单例        if not BuiltinEmbed._model and "tei-" in os.getenv("COMPOSE_PROFILES", ""):            with BuiltinEmbed._model_lock:  # 第65行：线程锁保护                BuiltinEmbed._model_name = settings.EMBEDDING_MDL                BuiltinEmbed._max_tokens = BuiltinEmbed.MAX_TOKENS.get(settings.EMBEDDING_MDL, 500)                # 第68行：创建HuggingFace TEI客户端                BuiltinEmbed._model = HuggingFaceEmbed(                    embedding_cfg["api_key"],                     settings.EMBEDDING_MDL,                     base_url=embedding_cfg["base_url"]                )        # 第69-71行：每个实例共享全局模型（节省内存）        self._model = BuiltinEmbed._model        self._model_name = BuiltinEmbed._model_name        self._max_tokens = BuiltinEmbed._max_tokens    def encode(self, texts: list):  # 第73-85行：批量文档编码        batch_size = 16  # 第74行：TEI推荐batch size        token_count = 0  # 第76行：累计token数        ress = None        # 第78-84行：分批编码，自动截断超长文本        for i in range(0, len(texts), batch_size):            embeddings, token_count_delta = self._model.encode(texts[i : i + batch_size])            token_count += token_count_delta            # 第81-84行：拼接批次结果            if ress is None:                ress = embeddings            else:                ress = np.concatenate((ress, embeddings), axis=0)  # NumPy拼接        return ress, token_count    def encode_queries(self, text: str):  # 第87-88行：查询编码        return self._model.encode_queries(text)

class OpenAIEmbed(Base):  # 第91行：OpenAI embedding实现    _FACTORY_NAME = "OpenAI"  # 第92行：工厂名称    def __init__(self, key, model_name="text-embedding-ada-002", base_url="https://api.openai.com/v1"):        # 第95-96行：base_url默认为OpenAI官方API        if not base_url:            base_url = "https://api.openai.com/v1"        # 第97-98行：创建OpenAI客户端        self.client = OpenAI(api_key=key, base_url=base_url)        self.model_name = model_name    def encode(self, texts: list):  # 第100-114行：批量文档编码        batch_size = 16  # 第102行：OpenAI batch限制        texts = [truncate(t, 8191) for t in texts]  # 第103行：截断至8191 tokens（ada-002限制）        ress = []        total_tokens = 0        # 第106-113行：分批调用API        for i in range(0, len(texts), batch_size):            res = self.client.embeddings.create(                input=texts[i : i + batch_size],                model=self.model_name,                encoding_format="float",  # 第107行：返回float格式                extra_body={"drop_params": True}  # 第107行：丢弃不支持参数            )            try:                # 第109-110行：提取embedding和token数                ress.extend([d.embedding for d in res.data])                total_tokens += total_token_count_from_response(res)            except Exception as _e:                log_exception(_e, res)                raise Exception(f"Error: {res}")        return np.array(ress), total_tokens    def encode_queries(self, text):  # 第116-122行：查询编码        res = self.client.embeddings.create(            input=[truncate(text, 8191)],            model=self.model_name,            encoding_format="float",            extra_body={"drop_params": True}        )        try:            return np.array(res.data[0].embedding), total_token_count_from_response(res)        except Exception as _e:            log_exception(_e, res)            raise Exception(f"Error: {res}")

class Base(ABC):  # 第28行：抽象基类    def __init__(self, key, model_name, **kwargs):        """        Abstract base class constructor.        Parameters are not stored; initialization is left to subclasses.        """        pass  # 第34行：基类不存储参数    def similarity(self, query: str, texts: list):  # 第36-37行：相似度计算方法        raise NotImplementedError("Please implement encode method!")    @staticmethod    def _normalize_rank(rank: np.ndarray) -> np.ndarray:  # 第39-53行：归一化方法        """        Normalize rank values to the range 0 to 1.        Avoids division by zero if all ranks are identical.        """        min_rank = np.min(rank)  # 第45行：最小值        max_rank = np.max(rank)  # 第46行：最大值        # 第48-51行：如果min和max接近（所有rank相似），返回零数组        if not np.isclose(min_rank, max_rank, atol=1e-3):            rank = (rank - min_rank) / (max_rank - min_rank)  # 归一化到0-1        else:            rank = np.zeros_like(rank)  # 全零        return rank

class JinaRerank(Base):  # 第56行：Jina rerank实现    _FACTORY_NAME = "Jina"  # 第57行：工厂名称    def __init__(self, key, model_name="jina-reranker-v2-base-multilingual", base_url="https://api.jina.ai/v1/rerank"):        self.base_url = "https://api.jina.ai/v1/rerank"  # 第60行：固定API endpoint        self.headers = {"Content-Type": "application/json", "Authorization": f"Bearer {key}"}  # 第61行：认证头        self.model_name = model_name    def similarity(self, query: str, texts: list):  # 第64-74行：相似度计算        texts = [truncate(t, 8196) for t in texts]  # 第65行：截断至8196 tokens        # 第66行：构造请求数据        data = {            "model": self.model_name,            "query": query,            "documents": texts,            "top_n": len(texts)  # 返回所有文本的rank        }        # 第67行：POST请求        res = requests.post(self.base_url, headers=self.headers, json=data).json()        rank = np.zeros(len(texts), dtype=float)  # 第68行：初始化rank数组        try:            # 第70-71行：解析results数组            for d in res["results"]:                rank[d["index"]] = d["relevance_score"]  # 按索引赋值score        except Exception as _e:            log_exception(_e, res)        return rank, total_token_count_from_response(res)  # 第74行：返回rank和token数

{  "results": [    {"index": 0, "relevance_score": 0.95},    {"index": 2, "relevance_score": 0.85},    {"index": 1, "relevance_score": 0.75}  ],  "usage": {"total_tokens": 1024}}

class CoHereRerank(Base):  # 第232行：Cohere rerank实现    _FACTORY_NAME = ["Cohere", "VLLM"]  # 第233行：双工厂名称（Cohere和VLLM共用）    def __init__(self, key, model_name, base_url=None):        from cohere import Client        # 第239-242行：base_url可选，支持自定义endpoint        client_kwargs = {"api_key": key}        if base_url and base_url.strip():            client_kwargs["base_url"] = base_url  # 自定义URL（如VLLM本地部署）        self.client = Client(**client_kwargs)        self.model_name = model_name.split("___")[0]  # 第243行：去除___后缀    def similarity(self, query: str, texts: list):  # 第245-260行：相似度计算        token_count = num_tokens_from_string(query) + sum([num_tokens_from_string(t) for t in texts])  # 第246行：token估算        # 第247-253行：调用Cohere rerank API        res = self.client.rerank(            model=self.model_name,            query=query,            documents=texts,            top_n=len(texts),  # 返回所有文本            return_documents=False  # 不返回文档内容，节省带宽        )        rank = np.zeros(len(texts), dtype=float)  # 第254行：初始化rank数组        try:            # 第256-257行：解析results数组            for d in res.results:                rank[d.index] = d.relevance_score  # Cohere SDK返回对象属性        except Exception as _e:            log_exception(_e, res)        return rank, token_count  # 第260行：返回rank和token数

class HuggingfaceRerank(Base):  # 第388行：本地rerank实现    _FACTORY_NAME = "HuggingFace"  # 第389行：工厂名称    @staticmethod    def post(query: str, texts: list, url="127.0.0.1"):  # 第392-409行：静态POST方法        exc = None        scores = [0 for _ in range(len(texts))]  # 第394行：初始化scores        batch_size = 8  # 第395行：批次大小        for i in range(0, len(texts), batch_size):            try:                # 第398-400行：HTTP POST请求本地服务                res = requests.post(                    f"http://{url}/rerank",                    headers={"Content-Type": "application/json"},                    json={                        "query": query,                        "texts": texts[i : i + batch_size],                        "raw_scores": False,  # 返回归一化score                        "truncate": True  # 自动截断超长文本                    }                )                # 第402-403行：解析响应并赋值                for o in res.json():                    scores[o["index"] + i] = o["score"]  # 注意：batch内index需+offset            except Exception as e:                exc = e  # 记录异常但不立即抛出        if exc:            raise exc  # 第408行：最后抛出异常        return np.array(scores)  # 第409行：返回scores数组    def __init__(self, key, model_name="BAAI/bge-reranker-v2-m3", base_url="http://127.0.0.1"):        self.model_name = model_name.split("___")[0]  # 第412行：去除后缀        self.base_url = base_url  # 第413行：本地服务URL    def similarity(self, query: str, texts: list) -> tuple[np.ndarray, int]:  # 第415-421行        if not texts:            return np.array([]), 0        token_count = 0        for t in texts:            token_count += num_tokens_from_string(t)  # 第419-420行：本地估算token        # 第421行：调用静态post方法        return HuggingfaceRerank.post(query, texts, self.base_url), token_count

{  "components": {    "File": {      "obj": {        "component_name": "File",        "params": {"path": "/data/document.pdf"}      },      "downstream": ["Parser"]    },    "Parser": {      "obj": {        "component_name": "Parser",        "params": {"parser_id": "pdf"}      },      "downstream": ["Embedding"]    },    "Embedding": {      "obj": {        "component_name": "Embedding",        "params": {"model": "BAAI/bge-m3"}      },      "downstream": ["Retrieval"]    },    "Retrieval": {      "obj": {        "component_name": "Retrieval",        "params": {"top_k": 100, "kb_ids": ["kb_001"]}      },      "downstream": ["Rerank"]    },    "Rerank": {      "obj": {        "component_name": "Rerank",        "params": {"model": "BAAI/bge-reranker-v2-m3"}      },      "downstream": ["Generate"]    },    "Generate": {      "obj": {        "component_name": "Generate",        "params": {"llm": "qwen-plus"}      },      "downstream": ["END"]    }  }}

# 在rag/flow/pipeline.py:117添加日志async def run(self, **kwargs):    print(f"[DEBUG] Pipeline.run开始，path={self.path}, components={list(self.components.keys())}")    # 在第143行循环内添加组件调用日志    while idx < len(self.path) and not self.error:        print(f"[DEBUG] 执行组件{self.path[idx]}，上游组件{self.path[idx-1]}输出={last_cpn.output()}")        await cpn_obj.invoke(**last_cpn.output())        print(f"[DEBUG] 组件{self.path[idx]}执行完成，error={cpn_obj.error()}")

# 在rag/llm/embedding_model.py:78添加批次日志for i in range(0, len(texts), batch_size):    print(f"[DEBUG] Embedding批次{i}-{i+batch_size}，texts数量={len(texts[i:i+batch_size])}")    embeddings, token_count_delta = self._model.encode(texts[i : i + batch_size])    print(f"[DEBUG] 批次embedding形状={embeddings.shape}, token_count={token_count_delta}")

# 在rag/llm/rerank_model.py:70添加映射日志for d in res["results"]:    print(f"[DEBUG] Rerank结果：index={d['index']}, relevance_score={d['relevance_score']}")    rank[d["index"]] = d["relevance_score"]print(f"[DEBUG] Rerank最终rank数组={rank}")

# 检查环境变量docker exec ragflow-container printenv COMPOSE_PROFILES# 应输出tei-qwen或tei-bge# 检查settings配置docker exec ragflow-container cat /ragflow/conf/service_conf.yaml | grep EMBEDDING

# 在pipeline.py:51添加异常捕获try:    bin = REDIS_CONN.get(log_key)except Exception as redis_err:    print(f"[ERROR] Redis连接失败：{redis_err}")    # 继续执行，不中断Pipeline

# 在embedding_model.py:103强制截断texts = [truncate(t, 8191) if len(t) > 8191 else t for t in texts]

# 在embedding_model.py:195增加延迟time.sleep(30)  # 从10秒增加到30秒

# 在rerank_model.py:48放宽toleranceif not np.isclose(min_rank, max_rank, atol=1e-6):  # 从1e-3改为1e-6    rank = (rank - min_rank) / (max_rank - min_rank)