RAGFlow源码解析-7、Agent画布系统与组件编排深度解析(第五周)

classGraph:    """        dsl = {            "components": {                "begin": {                    "obj":{                        "component_name": "Begin",                        "params": {},                    },                    "downstream": ["answer_0"],                    "upstream": [],                },                "retrieval_0": {                    "obj": {                        "component_name": "Retrieval",                        "params": {}                    },                    "downstream": ["generate_0"],                    "upstream": ["answer_0"],                },                "generate_0": {                    "obj": {                        "component_name": "Generate",                        "params": {}                    },                    "downstream": ["answer_0"],                    "upstream": ["retrieval_0"],                }            },            "history": [],            "path": ["begin"],            "retrieval": {"chunks": [], "doc_aggs": []},            "globals": {                "sys.query": "",                "sys.files": []            }        }        """

def __init__(self, dsl: str, tenant_id=None, task_id=None, custom_header=None):    self.path = []  # 第80行：执行路径初始化    self.components = {}  # 第81行：组件字典初始化    self.error = ""  # 第82行：错误信息    self.dsl = json.loads(dsl)  # 第83行：解析DSL JSON字符串    self._tenant_id = tenant_id  # 第84行：租户ID    self.task_id = task_id if task_id else get_uuid()  # 第85行：任务ID（自动生成UUID）    self.custom_header = custom_header  # 第86行：自定义HTTP头    self._thread_pool = ThreadPoolExecutor(max_workers=5)  # 第87行：线程池（最大5个工作线程）    self.load()  # 第88行：加载组件def load(self):    self.components = self.dsl["components"]  # 第91行：获取组件字典    cpn_nms = set([])  # 第92行：组件名称集合    # 第93-103行：遍历组件并实例化    for k, cpn in self.components.items():        cpn_nms.add(cpn["obj"]["component_name"])  # 第94行：记录组件名称        # 第95-97行：创建参数对象并更新        param = component_class(cpn["obj"]["component_name"] + "Param")()  # 第95行：动态创建参数类        cpn["obj"]["params"]["custom_header"] = self.custom_header        param.update(cpn["obj"]["params"])  # 第97行：更新参数        try:            param.check()  # 第99行：参数校验        except Exception as e:            raise ValueError(self.get_component_name(k) + f": {e}")  # 第101行：抛出校验错误        # 第103行：实例化组件对象        cpn["obj"] = component_class(cpn["obj"]["component_name"])(self, k, param)    self.path = self.dsl["path"]  # 第105行：恢复执行路径

def get_value_with_variable(self, value: str) -> Any:    # 第163行：正则匹配变量表达式 {{变量名}}    pat = re.compile(r"\{* *\{([a-zA-Z:0-9]+@[A-Za-z0-9_.-]+|sys\.[A-Za-z0-9_.]+|env\.[A-Za-z0-9_.]+)\} *\}*")    out_parts = []    last = 0    # 第167-184行：遍历所有匹配项    for m in pat.finditer(value):        out_parts.append(value[last:m.start()])  # 第168行：添加匹配前的文本        key = m.group(1)  # 第169行：提取变量名        v = self.get_variable_value(key)  # 第170行：获取变量值        if v is None:            rep = ""        elif isinstance(v, partial):            # 第173-177行：处理partial函数（延迟执行）            buf = []            for chunk in v():                buf.append(chunk)            rep = "".join(buf)        elif isinstance(v, str):            rep = v        else:            rep = json.dumps(v, ensure_ascii=False)  # 第181行：JSON序列化        out_parts.append(rep)  # 第183行：添加替换值        last = m.end()    out_parts.append(value[last:])  # 第186行：添加剩余文本    return("".join(out_parts))def get_variable_value(self, exp: str) -> Any:    exp = exp.strip("{").strip("}").strip(" ").strip("{").strip("}")  # 第190行：清理花括号    # 第191-192行：全局变量（无@符号）    if exp.find("@") < 0:        return self.globals[exp]    # 第193-204行：组件变量（cpn_id@var_nm格式）    cpn_id, var_nm = exp.split("@")  # 第193行：分割组件ID和变量名    cpn = self.get_component(cpn_id)    if not cpn:        raise Exception(f"Can't find variable: '{cpn_id}@{var_nm}'")    parts = var_nm.split(".", 1)  # 第197行：分割根键和路径    root_key = parts[0]    rest = parts[1] if len(parts) > 1 else ""    root_val = cpn["obj"].output(root_key)  # 第200行：获取组件输出    if not rest:        return root_val    return self.get_variable_param_value(root_val, rest)  # 第204行：递归获取嵌套值def get_variable_param_value(self, obj: Any, path: str) -> Any:    cur = obj    if not path:        return cur    # 第210-232行：遍历路径键    for key in path.split('.'):        if cur is None:            return None        # 第214-218行：字符串尝试JSON解析        if isinstance(cur, str):            try:                cur = json.loads(cur)            except Exception:                return None        # 第220-222行：字典类型        if isinstance(cur, dict):            cur = cur.get(key)            continue        # 第224-230行：列表/元组类型（支持索引访问）        if isinstance(cur, (list, tuple)):            try:                idx = int(key)                cur = cur[idx]            except Exception:                return None            continue        cur = getattr(cur, key, None)  # 第232行：对象属性访问    return cur

def get_component(self, cpn_id) -> Union[None, dict[str, Any]]:    return self.components.get(cpn_id)  # 第148行：获取组件字典def get_component_obj(self, cpn_id) -> ComponentBase:    return self.components.get(cpn_id)["obj"]  # 第151行：获取组件实例对象def get_component_type(self, cpn_id) -> str:    return self.components.get(cpn_id)["obj"].component_name  # 第154行：获取组件类型名称def get_component_input_form(self, cpn_id) -> dict:    return self.components.get(cpn_id)["obj"].get_input_form()  # 第157行：获取组件输入表单def get_tenant_id(self):    return self._tenant_id  # 第160行：获取租户ID

class Canvas(Graph):  # 第279行：继承Graph基类    def __init__(self, dsl: str, tenant_id=None, task_id=None, canvas_id=None, custom_header=None):        # 第282-285行：初始化全局变量        self.globals = {            "sys.query": "",  # 用户查询            "sys.files": []   # 文件列表        }        self.variables = {}  # 第286行：局部变量字典        # 第287行：调用父类初始化        super().__init__(dsl, tenant_id, task_id, custom_header=custom_header)        self._id = canvas_id  # 第288行：画布ID    def load(self):        super().load()  # 第291行：调用父类加载        # 第292-299行：加载历史和全局变量        self.history = self.dsl["history"]        if "globals" in self.dsl:            self.globals = self.dsl["globals"]        else:            self.globals = {                "sys.query": "",                "sys.files": []            }        # 第300-305行：加载变量和检索结果        if "variables" in self.dsl:            self.variables = self.dsl["variables"]        else:            self.variables = {}        self.retrieval = self.dsl["retrieval"]  # 第307行：检索结果        self.memory = self.dsl.get("memory", [])  # 第308行：记忆列表

# Graph.reset (第128-136行)def reset(self):    self.path = []  # 第129行：清空执行路径    for k, cpn in self.components.items():        self.components[k]["obj"].reset()  # 第131行：重置所有组件    try:        REDIS_CONN.delete(f"{self.task_id}-logs")  # 第133行：删除Redis日志        REDIS_CONN.delete(f"{self.task_id}-cancel")  # 第134行：删除取消标志    except Exception as e:        logging.exception(e)# Canvas.reset (推测实现)def reset(self, mem=False):    super().reset()  # 调用父类重置    if not mem:  # 如果不保留记忆        self.history = []  # 清空对话历史        self.retrieval = []  # 清空检索结果        self.memory = []  # 清空记忆    print(self.variables)  # 打印变量（调试用）    # 重置全局系统变量    for k in self.globals.keys():        if k.startswith("sys."):            if isinstance(self.globals[k], str):                self.globals[k] = ""  # 字符串类型清空            elif isinstance(self.globals[k], int):                self.globals[k] = 0  # 整数类型归零            elif isinstance(self.globals[k], list):                self.globals[k] = []  # 列表类型清空

class ComponentBase:    component_name = "Base"  # 组件名称（类属性）    def __init__(self, canvas, id, param):        self._canvas = canvas  # 画布引用        self._id = id  # 组件ID        self._param = param  # 参数对象        self._output = {}  # 输出字典        self._error = ""  # 错误信息    def reset(self):        """重置组件状态"""        self._output = {}        self._error = ""    def output(self, key=None):        """获取输出"""        if key is None:            return self._output        return self._output.get(key)    def set_output(self, key, value):        """设置输出"""        self._output[key] = value    def error(self):        """获取错误信息"""        return self._error    def set_error(self, msg):        """设置错误信息"""        self._error = msg    async def invoke(self, **kwargs):        """执行组件（子类必须实现）"""        raise NotImplementedError("Subclasses must implement invoke method")    def get_downstream(self):        """获取下游组件列表"""        cpn = self._canvas.get_component(self._id)        return cpn.get("downstream", [])    def get_upstream(self):        """获取上游组件列表"""        cpn = self._canvas.get_component(self._id)        return cpn.get("upstream", [])    def get_input_form(self):        """获取输入表单定义"""        return {}    def __str__(self):        """序列化为JSON字符串"""        return json.dumps({            "component_name": self.component_name,            "params": self._param.to_dict() if hasattr(self._param, "to_dict") else {}        }, ensure_ascii=False)

class ComponentParamBase:    def __init__(self):        self.custom_header = None  # 自定义HTTP头    def update(self, params):        """更新参数"""        for k, v in params.items():            if hasattr(self, k):                setattr(self, k, v)    def check(self):        """参数校验（子类可重写）"""        pass    def to_dict(self):        """转换为字典"""        return {k: v for k, v in self.__dict__.items() if not k.startswith('_')}

class LLMParam(ComponentParamBase):    def __init__(self):        super().__init__()        self.llm_id = ""  # LLM模型ID        self.prompt = ""  # 提示词        self.temperature = 0.1  # 温度参数        self.max_tokens = 2048  # 最大token数        self.max_retries = 5  # 最大重试次数        self.delay_after_error = 2.0  # 错误后延迟        self.max_rounds = 5  # 最大轮数        self.tools = []  # 工具列表        self.mcp = []  # MCP服务器列表    def check(self):        """校验参数"""        if not self.llm_id:            raise ValueError("llm_id cannot be empty")        if self.temperature < 0 or self.temperature > 2:            raise ValueError("temperature must be between 0 and 2")

class LLM(ComponentBase):    component_name = "LLM"    def __init__(self, canvas, id, param: LLMParam):        super().__init__(canvas, id, param)        # 创建LLM Bundle        self.chat_mdl = LLMBundle(            canvas.get_tenant_id(),            TenantLLMService.llm_id2llm_type(param.llm_id),            param.llm_id,            max_retries=param.max_retries,            retry_interval=param.delay_after_error,            max_rounds=param.max_rounds        )    async def invoke(self, **kwargs):        """执行LLM调用"""        # 获取上游输出        upstream_output = kwargs.get("input", {})        # 构建提示词（替换变量）        prompt = self._canvas.get_value_with_variable(self._param.prompt)        # 构建消息历史        history = self._canvas.history.copy()        history.append({"role": "user", "content": prompt})        # 调用LLM        gen_conf = {            "temperature": self._param.temperature,            "max_tokens": self._param.max_tokens        }        ans = ""        async for delta in self.chat_mdl.async_chat_streamly("", history, gen_conf):            if isinstance(delta, str):                ans += delta                yield delta  # 流式输出        # 设置输出        self.set_output("text", ans)        self._canvas.history.append({"role": "assistant", "content": ans})

class Agent(LLM, ToolBase):    component_name = "Agent"    def __init__(self, canvas, id, param: LLMParam):        LLM.__init__(self, canvas, id, param)        # 第89-93行：加载工具        self.tools = {}        for idx, cpn in enumerate(self._param.tools):            cpn = self._load_tool_obj(cpn)            original_name = cpn.get_meta()["function"]["name"]            indexed_name = f"{original_name}_{idx}"  # 工具名添加索引后缀            self.tools[indexed_name] = cpn        # 第96-102行：创建LLM Bundle        self.chat_mdl = LLMBundle(            self._canvas.get_tenant_id(),            TenantLLMService.llm_id2llm_type(self._param.llm_id),            self._param.llm_id,            max_retries=self._param.max_retries,            retry_interval=self._param.delay_after_error,            max_rounds=self._param.max_rounds,            verbose_tool_use=True  # 详细工具使用日志        )        # 第103-107行：构建工具元数据        self.tool_meta = []        for indexed_name, tool_obj in self.tools.items():            original_meta = tool_obj.get_meta()            indexed_meta = deepcopy(original_meta)            indexed_meta["function"]["name"] = indexed_name            self.tool_meta.append(indexed_meta)        # 第109-117行：加载MCP工具        for mcp in self._param.mcp:            _, mcp_server = MCPServerService.get_by_id(mcp["mcp_id"])            custom_header = self._param.custom_header            tool_call_session = MCPToolCallSession(mcp_server, mcp_server.variables, custom_header)            for tnm, meta in mcp["tools"].items():                # 添加MCP工具到tool_meta                # ...

class Retrieval(ComponentBase):    component_name = "Retrieval"    def __init__(self, canvas, id, param: RetrievalParam):        super().__init__(canvas, id, param)        self.retrieval_mdl = LLMBundle(            canvas.get_tenant_id(),            LLMType.EMBEDDING,            param.embd_id        )    async def invoke(self, **kwargs):        """执行检索"""        # 获取查询（替换变量）        query = self._canvas.get_value_with_variable(self._param.query)        # 调用检索器        from rag.nlp import search        dealer = search.Dealer(settings.docStoreConn)        ranks = await dealer.retrieval(            question=query,            embd_mdl=self.retrieval_mdl,            tenant_ids=[self._canvas.get_tenant_id()],            kb_ids=self._param.kb_ids,            page=1,            page_size=self._param.top_n,            similarity_threshold=self._param.similarity_threshold,            vector_similarity_weight=self._param.vector_similarity_weight        )        # 设置输出        self.set_output("chunks", ranks["chunks"])        self.set_output("doc_aggs", ranks["doc_aggs"])        # 缓存到canvas        self._canvas.retrieval = ranks

class Categorize(ComponentBase):    component_name = "Categorize"    def __init__(self, canvas, id, param: CategorizeParam):        super().__init__(canvas, id, param)        self.chat_mdl = LLMBundle(            canvas.get_tenant_id(),            LLMType.CHAT,            param.llm_id        )    async def invoke(self, **kwargs):        """执行分类"""        # 获取输入文本        input_text = self._canvas.get_value_with_variable(self._param.input)        # 构建分类提示词        categories = self._param.categories  # 分类列表        prompt = f"请将以下文本分类到以下类别之一：{', '.join(categories)}\n\n文本：{input_text}\n\n类别："        # 调用LLM        history = [{"role": "user", "content": prompt}]        ans = ""        async for delta in self.chat_mdl.async_chat_streamly("", history, {}):            if isinstance(delta, str):                ans += delta        # 解析分类结果        category = self._parse_category(ans, categories)        # 设置输出        self.set_output("category", category)        # 动态设置下游组件（根据分类结果）        downstream_map = self._param.category_to_downstream        downstream = downstream_map.get(category, self.get_downstream())        # 更新组件的downstream        cpn = self._canvas.get_component(self._id)        cpn["downstream"] = [downstream]

class Switch(ComponentBase):    component_name = "Switch"    def __init__(self, canvas, id, param: SwitchParam):        super().__init__(canvas, id, param)    async def invoke(self, **kwargs):        """执行分支判断"""        # 获取条件变量        condition_var = self._canvas.get_value_with_variable(self._param.condition)        # 遍历条件分支        for case in self._param.cases:            # 评估条件            if self._evaluate_condition(condition_var, case["condition"]):                # 匹配成功，设置下游组件                cpn = self._canvas.get_component(self._id)                cpn["downstream"] = [case["downstream"]]                self.set_output("matched", case["condition"])                return        # 无匹配，使用默认分支        if self._param.default_downstream:            cpn = self._canvas.get_component(self._id)            cpn["downstream"] = [self._param.default_downstream]            self.set_output("matched", "default")    def _evaluate_condition(self, value, condition):        """评估条件"""        # 支持多种条件类型：等于、包含、正则匹配等        if condition.startswith("=="):            return value == condition[2:].strip()        elif condition.startswith("contains:"):            return condition[9:].strip() in str(value)        elif condition.startswith("regex:"):            return bool(re.match(condition[6:].strip(), str(value)))        else:            return value == condition

class Loop(ComponentBase):    component_name = "Loop"    def __init__(self, canvas, id, param: LoopParam):        super().__init__(canvas, id, param)    async def invoke(self, **kwargs):        """执行循环"""        # 获取循环变量        items = self._canvas.get_value_with_variable(self._param.items)        if not isinstance(items, (list, tuple)):            items = [items]        # 设置循环上下文        self._canvas.variables[f"{self._id}_index"] = 0        self._canvas.variables[f"{self._id}_items"] = items        # 设置下游为循环体第一个组件        cpn = self._canvas.get_component(self._id)        cpn["downstream"] = [self._param.loop_body]        self.set_output("count", len(items))

class ToolBase:    def __init__(self, canvas, id, param):        self._canvas = canvas        self._id = id        self._param = param    def get_meta(self) -> dict:        """获取工具元数据（OpenAI function calling格式）"""        raise NotImplementedError("Subclasses must implement get_meta")    async def invoke(self, arguments: dict):        """执行工具"""        raise NotImplementedError("Subclasses must implement invoke")    def _load_tool_obj(self, tool_config):        """加载工具对象"""        tool_name = tool_config["component_name"]        tool_param = component_class(tool_name + "Param")()        tool_param.update(tool_config["params"])        return component_class(tool_name)(self._canvas, self._id, tool_param)

{  "type": "function",  "function": {    "name": "search_web",    "description": "搜索网络获取信息",    "parameters": {      "type": "object",      "properties": {        "query": {          "type": "string",          "description": "搜索查询词"        },        "top_k": {          "type": "integer",          "description": "返回结果数量",          "default": 5        }      },      "required": ["query"]    }  }}

class WebSearch(ToolBase):    def get_meta(self) -> dict:        return {            "type": "function",            "function": {                "name": "web_search",                "description": "搜索网络获取实时信息",                "parameters": {                    "type": "object",                    "properties": {                        "query": {"type": "string", "description": "搜索查询"},                        "top_k": {"type": "integer", "default": 5}                    },                    "required": ["query"]                }            }        }    async def invoke(self, arguments: dict):        query = arguments["query"]        top_k = arguments.get("top_k", 5)        # 调用搜索API        results = await self._search(query, top_k)        return {            "results": results,            "count": len(results)        }

async def run(self, **kwargs):    """执行画布"""    # 初始化执行路径    if not self.path:        self.path.append("begin")    # 执行第一个组件    cpn_obj = self.get_component_obj(self.path[0])    async for _ in cpn_obj.invoke(**kwargs):        pass  # 消费流式输出    if cpn_obj.error():        self.error = "[ERROR]" + cpn_obj.error()        return    # 循环执行下游组件    idx = len(self.path) - 1    cpn_obj = self.get_component_obj(self.path[idx])    idx += 1    self.path.extend(cpn_obj.get_downstream())    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])        # 执行组件（传递上游输出）        async for _ in cpn_obj.invoke(**last_cpn.output()):            pass        if cpn_obj.error():            self.error = "[ERROR]" + cpn_obj.error()            break        idx += 1        self.path.extend(cpn_obj.get_downstream())    # 返回最终输出    if not self.error:        return self.get_component_obj(self.path[-1]).output()    return {}

# 在agent/canvas.py:170添加日志v = self.get_variable_value(key)print(f"[DEBUG] 变量解析：key={key}, value={v}, type={type(v)}")

# 在组件invoke方法开始添加日志async def invoke(self, **kwargs):    print(f"[DEBUG] 组件执行：id={self._id}, name={self.component_name}, kwargs={kwargs}")    # ...

# 在agent/component/agent_with_tools.py工具调用处添加日志tool_result = await tool.invoke(arguments)print(f"[DEBUG] 工具调用：tool={tool_name}, args={arguments}, result={tool_result}")

# 检查变量表达式格式# 正确：{{retrieval_0@chunks}}# 错误：{{retrieval_0@chunk}}（变量名错误）# 在canvas.py:196添加详细错误信息if not cpn:    available_cpn_ids = list(self.components.keys())    raise Exception(f"Can't find variable: '{cpn_id}@{var_nm}'. Available components: {available_cpn_ids}")

# 在canvas.py:232添加路径解析日志cur = getattr(cur, key, None)if cur is None:    logging.warning(f"路径解析失败：path={path}, current_key={key}, available_keys={dir(obj) ifhasattr(obj, '__dict__') elselist(obj.keys()) ifisinstance(obj, dict) else []}")

# 在执行循环中添加路径检测visited = set()while idx < len(self.path) and not self.error:    cpn_id = self.path[idx]    if cpn_id in visited:        raise Exception(f"检测到循环依赖：组件{cpn_id}已执行过")    visited.add(cpn_id)    # ...

# 确保使用async for消费流式输出async for _ in cpn_obj.invoke(**kwargs):    pass  # 必须消费，否则流式输出丢失

# 已实现：工具名索引化（agent_with_tools.py:92）indexed_name = f"{original_name}_{idx}"# 例如：search_web_0, search_web_1

# 检查MCP服务器状态docker exec ragflow-container curl http://mcp-server:port/health# 检查MCP配置docker exec ragflow-container cat /ragflow/conf/mcp_config.yaml