How RAGFlow’s Agent Engine Turns Retrieval into a Problem‑Solving AI

RAGFlow Agent Engine Overview

RAGFlow’s agent is likened to a senior research team lead that not only retrieves relevant documents but also plans, delegates, and integrates results to answer complex queries, representing a shift from a passive "question‑answer" engine to an active "problem‑solving" engine.

Core Technical Features

ReAct Reasoning Framework : Implements a loop of Reasoning → Action → Observation so the LLM can think like a human expert.

Visual Canvas Workflow : A zero‑code, drag‑and‑drop DSL built with React and ReactFlow that lets non‑programmers design agent workflows.

Modular Component & Tool System : Rich built‑in components (Begin, LLM, Retrieval, Categorize, etc.) and tools (knowledge retrieval, code execution, web search, database query) that can be extended.

7.1 ReAct Framework – Reasoning and Acting

The ReAct loop drives the agent’s decision process. Each round consists of:

Reasoning : The LLM analyses the current task and decides the next step.

Action : The LLM selects a tool and supplies parameters.

Observation : The tool’s result is fed back as new information for further reasoning.

The implementation lives in agent/component/agent_with_tools.py:

class Agent(LLM, ToolBase):
    """RAGFlow agent implementation – combines ReAct framework and tool calls"""
    component_name = "Agent"
    def _react_with_tools_streamly(self, prompt, history: list[dict], use_tools, user_defined_prompt={}):
        """Streamed ReAct loop – each iteration performs analysis → decision → execution → reflection"""
        task_desc = analyze_task(self.chat_mdl, prompt, user_request, tool_metas, user_defined_prompt)
        for round_num in range(self._param.max_rounds + 1):
            # Reasoning stage
            response, tk = next_step(self.chat_mdl, hist, tool_metas, task_desc, user_defined_prompt)
            hist.append({"role": "assistant", "content": response})
            # Parse and execute tool calls
            functions = json_repair.loads(re.sub(r"```.*", "", response))
            for func in functions:
                name = func["name"]
                args = func["arguments"]
                if name == COMPLETE_TASK:
                    for txt, tkcnt in complete():
                        yield txt, tkcnt
                    return
                else:
                    tool_response = self.toolcall_session.tool_call(name, args)
                    use_tools.append({"name": name, "arguments": args, "results": tool_response})
            # Reflection stage
            reflection = reflect(self.chat_mdl, hist, tool_results, user_defined_prompt)
            hist.append({"role": "user", "content": reflection})

7.2 Visual Canvas System – Zero‑Code Workflow Builder

The canvas provides a domain‑specific language (DSL) that describes a directed acyclic graph (DAG) of components. The front‑end resides in web/src/pages/agent/canvas/ (React + ReactFlow). The back‑end parses the JSON DSL and executes the graph.

Key back‑end class agent/canvas.py:

class Canvas(Graph):
    def run(self, **kwargs):
        """Execute the workflow – turn visual design into real actions"""
        for k in kwargs.keys():
            if k in ["query", "user_id", "files"] and kwargs[k]:
                self.globals[f"sys.{k}"] = kwargs[k]
        idx = len(self.path) - 1
        while idx < len(self.path):
            self._run_batch(idx, len(self.path))
            for i in range(idx, len(self.path)):
                cpn_obj = self.get_component_obj(self.path[i])
                if cpn_obj.component_name.lower() in ["categorize", "switch"]:
                    self.path.extend(cpn_obj.output("_next"))
                elif cpn_obj.component_name.lower() == "iteration":
                    self.path.append(cpn_obj.get_start())
                else:
                    self.path.extend(self.get_component(self.path[i])["downstream"])
            idx = len(self.path)
        return self.get_component_obj(self.path[-1]).output()

    def __init__(self, dsl: dict, tenant_id: str = None, task_id: str = None):
        self.dsl = dsl
        self.tenant_id = tenant_id
        self.task_id = task_id
        self.graph = nx.DiGraph()
        self.component_instances = {}
        self.execution_context = ExecutionContext()
        self._build_graph(dsl.get("nodes", []), dsl.get("edges", []))
        self._validate_graph()

    def _build_graph(self, nodes: List[dict], edges: List[dict]):
        for node_data in nodes:
            node_id = node_data["id"]
            component_type = node_data["type"]
            component_params = node_data.get("params", {})
            component_class = ComponentFactory.get_component(component_type)
            component_instance = component_class(**component_params)
            self.component_instances[node_id] = component_instance
            self.graph.add_node(node_id, component=component_instance, type=component_type, params=component_params)
        for edge_data in edges:
            source_id = edge_data["source"]
            target_id = edge_data["target"]
            if source_id not in self.graph or target_id not in self.graph:
                raise ValueError(f"Edge connects non‑existent nodes: {source_id} -> {target_id}")
            self.graph.add_edge(source_id, target_id, source_port=edge_data.get("source_port"), target_port=edge_data.get("target_port"))

    def _validate_graph(self):
        if not nx.is_directed_acyclic_graph(self.graph):
            raise ValueError("Workflow contains cycles, which is not allowed")
        begin_nodes = [n for n in self.graph.nodes() if self.graph.nodes[n]["type"] == "Begin"]
        if len(begin_nodes) != 1:
            raise ValueError(f"Workflow must have exactly one Begin node, found {len(begin_nodes)}")
        self.start_node = begin_nodes[0]

7.3 Component & Tool Ecosystem

All components inherit from ComponentBase. Core components include:

Begin : Workflow entry point.

LLM : Text generation.

Retrieval : Knowledge base lookup.

Categorize : Conditional branching.

Tools located in agent/tools/ provide capabilities such as:

Knowledge retrieval from vector stores.

Sandboxed Python code execution.

Real‑time web search.

SQL database queries.

Custom tools can be added by subclassing ToolBase. Example – a weather‑fetching tool:

class WeatherTool(ToolBase):
    component_name = "Weather"
    def _invoke(self, **kwargs):
        city = kwargs.get("city")
        weather_data = self._get_weather_data(city)
        return weather_data

Chapter Summary

The chapter demonstrates how RAGFlow’s agent engine transforms a conventional RAG system into an autonomous problem‑solving platform. By combining the ReAct reasoning loop, a visual canvas that lowers the barrier to workflow creation, and a highly modular component‑tool architecture, developers can build sophisticated AI applications that plan, act, and iterate without manual coding of each step.

Next, the series will explore tracing and diagnosing complex agent workflows.