RAGFlow/Agent Component System — LLM Component Deep-Dive

Overview: LLM Component Architecture

The LLM component is a bridge between RAGFlow workflows and large language models. It handles chat message formatting, API calls, response parsing, and structured output generation. The component supports OpenAI, Claude, Gemini, and local models via a unified interface.

Message Pipeline: Building API Requests

Before sending a chat completion request to the API, the LLM component assembles a complete conversation context from multiple sources.

Pipeline Stages

1. System Prompt: Always first message. Instructs the model on role, constraints, output format. Example: "You are a helpful assistant. Output JSON responses."
2. Conversation History: Previous turns (user input, assistant response, tool results). Loaded from Canvas history or component state.
3. Context Window: Relevant documents, search results, or code snippets. Inserted as human message or in system prompt using RAG context.
4. Current User Input: The immediate query or instruction for this turn.
5. Tool Definitions (optional): JSON schema of available tools. Included if component.tools config is set. Model uses this to decide when to call tools.

Example Pipeline Build

def build_messages(self):
    messages = []
    
    # 1. System prompt (always first)
    messages.append({
        "role": "system",
        "content": "You are a code review assistant. Provide detailed feedback on code quality, security, and performance."
    })
    
    # 2. History from previous turns
    for turn in self.conversation_history:
        messages.append({"role": turn["role"], "content": turn["content"]})
    
    # 3. RAG context (optional)
    if self.context:
        messages.append({
            "role": "system",
            "content": f"Reference material:
{self.context}"
        })
    
    # 4. Current input
    messages.append({
        "role": "user",
        "content": self.current_input
    })
    
    # 5. Tools (optional)
    tools = self._build_tool_schemas()  # JSON schema array
    
    return messages, tools

Tool Calling: LLM-Invoked Components

When the model decides to call a tool, it returns a structured tool call response. The LLM component parses this and invokes the corresponding Tool node.

Tool Calling Flow

1. LLM Models Tool Need: Model sees tool definitions in system prompt or tools parameter. When appropriate, returns {"type": "tool_call", "tool_name": "...", "arguments": {...}}.
2. Parse Response: Component extracts tool_name and arguments from response.
3. Validate Tool: Check that tool_name exists in Agent Canvas Canvas & Execution — Canvas & Execution" class="xref">Canvas._tools registry. Validate arguments against tool schema.
4. Invoke Tool: Call Tool node's run() method. Pass arguments as kwargs.
5. Collect Result: Tool returns output (JSON). Component appends tool result message to history.
6. Loop Back: Continue conversation with tool result injected. Model may make more tool calls or respond to user.

Tool Call Parsing Example

def parse_tool_calls(response):
    """Extract tool_call objects from LLM response."""
    tool_calls = []
    
    if hasattr(response, 'tool_calls'):
        # OpenAI API response object
        for tc in response.tool_calls:
            tool_calls.append({
                'tool_name': tc.function.name,
                'arguments': json.loads(tc.function.arguments)
            })
    elif isinstance(response, str):
        # Raw JSON from Claude or local model
        try:
            data = json.loads(response)
            if 'tool_calls' in data:
                tool_calls = data['tool_calls']
        except json.JSONDecodeError:
            # Parse failure; fallback to no tools
            pass
    
    return tool_calls

def invoke_tool(tool_name, arguments):
    """Call Tool node and collect result."""
    tool_node = Canvas._tools.get(tool_name)
    if not tool_node:
        raise ValueError(f"Unknown tool: {tool_name}")
    
    result = tool_node.run(**arguments)
    
    # Append to conversation
    self.messages.append({
        "role": "tool",
        "tool_name": tool_name,
        "content": json.dumps(result)
    })
    
    return result

Structured Output Parsing

To ensure LLM outputs conform to expected schemas, the component uses JSON schema validation and re-prompting on parse failure.

Output Parsing Pipeline

1. Define Schema: Specify expected output format as JSON schema (e.g., {"type": "object", "properties": {"summary": {"type": "string"}, "score": {"type": "number"}}}).
2. Instruct Model: Prompt includes schema and example output. Increases likelihood of valid JSON.
3. Parse Response: Attempt to parse LLM output as JSON. Validate against schema.
4. Success Path: If valid, return structured output to caller.
5. Failure Path: If invalid, optionally re-prompt with error message. "Your output was invalid JSON: ... Please fix and retry."

Schema Validation Example

import json
from jsonschema import validate, ValidationError

def parse_structured_output(raw_response, schema):
    """Parse and validate LLM output against schema."""
    try:
        # Extract JSON from response (may be wrapped in text)
        data = json.loads(raw_response)
        
        # Validate against schema
        validate(instance=data, schema=schema)
        
        return data
    
    except json.JSONDecodeError as e:
        raise ValueError(f"Invalid JSON: {e}")
    
    except ValidationError as e:
        raise ValueError(f"Schema validation failed: {e.message}")

def run_with_retry(self, input_data):
    """Execute with re-prompting on parse failure."""
    max_retries = 3
    for attempt in range(max_retries):
        response = self.api_call(...)
        
        try:
            output = parse_structured_output(response, self.output_schema)
            return output
        except ValueError as e:
            if attempt < max_retries - 1:
                # Re-prompt with error
                self.messages.append({
                    "role": "assistant",
                    "content": response
                })
                self.messages.append({
                    "role": "user",
                    "content": f"Your output was invalid. Error: {e}

Please provide valid JSON output."
                })
            else:
                raise

Streaming Support

For real-time token generation, the LLM component buffers streaming responses and incrementally parses them.

Streaming Pipeline

1. Open Stream: Call API with stream=true. Receive iterator of token chunks.
2. Buffer Tokens: Accumulate chunks into a full response string.
3. Emit Tokens: Yield each chunk to caller immediately (for UI update, logging).
4. Parse Complete Response: Once stream ends, parse complete buffer (JSON schema, tool calls, etc.).

Streaming Code Example

async def stream_response(self, messages):
    """Stream tokens from LLM API."""
    buffer = ""
    
    async with self.client.chat.completions.create(
        model=self.model_name,
        messages=messages,
        stream=True
    ) as response:
        async for chunk in response:
            if chunk.choices[0].delta.content:
                token = chunk.choices[0].delta.content
                buffer += token
                
                # Emit token immediately for UI
                yield {"token": token, "partial": buffer}
    
    # Parse complete buffer after streaming
    output = parse_structured_output(buffer, self.output_schema)
    yield {"complete": output}

Context Window Management

To prevent token overflow, the LLM component implements a sliding window of conversation history.

Window Strategy

Window Implementation

def manage_context_window(self, messages, max_tokens=4096, reserve=512):
    """Slide conversation window to fit within token limit."""
    # System prompt is always kept
    system_msgs = [m for m in messages if m.get('role') == 'system']
    history = [m for m in messages if m.get('role') != 'system']
    
    # Count tokens
    system_tokens = sum(self.count_tokens(m['content']) for m in system_msgs)
    available = max_tokens - reserve - system_tokens
    
    # Slide window: remove oldest messages until we fit
    while history:
        total = sum(self.count_tokens(m['content']) for m in history)
        if total <= available:
            break
        history.pop(0)  # Remove oldest
    
    # Reconstruct
    return system_msgs + history

Retry & Fallback Strategy

Transient failures (timeout, rate limit) trigger exponential backoff retries. Persistent failures fall back to simpler formats.

Multi-Turn Conversation State

The LLM component maintains conversation history across multiple workflow executions. This enables context-aware dialogues.

State Management

• Local State: self.conversation_history = [] list of {"role", "content"} dicts. Preserved in memory during session.
• Canvas History: Canvas.history records all execution steps, including LLM inputs/outputs. Immutable audit trail.
• Persistence: On workflow save, serialize conversation_history to JSON. On load, deserialize and restore.

State Example

class LLM(ComponentBase):
    def __init__(self):
        self.conversation_history = []
    
    def run(self, user_input):
        # Add user input to history
        self.conversation_history.append({
            "role": "user",
            "content": user_input
        })
        
        # Build messages (system + history + current input)
        messages = self.build_messages()
        
        # Call API
        response = self.client.chat.completions.create(
            model=self.model_name,
            messages=messages
        )
        
        # Add assistant response to history
        self.conversation_history.append({
            "role": "assistant",
            "content": response.choices[0].message.content
        })
        
        return {"response": response.choices[0].message.content}
    
    def to_dict(self):
        """Serialize state for persistence."""
        return {
            "conversation_history": self.conversation_history,
            ...
        }
    
    @classmethod
    def from_dict(cls, data):
        """Restore state from persistence."""
        instance = cls()
        instance.conversation_history = data.get("conversation_history", [])
        return instance

Error Classification & Recovery

Related Topics

• Component System Overview — Parent L2 module overview
• LLM Component Deep-Dive — Tool calling, structured output, streaming (this page)
• Agent Component — How Agent orchestrates LLM + Tool calls (Component System module)
• Tool Component — How tools are invoked and results cached (Tools & Integration module)