# How Langflow Implements Multi-Agent Orchestration and Conversation Management

> Discover how Langflow uses a three-layer architecture for multi-agent orchestration and conversation management, integrating a graph runner, agent class, and custom tool pipelines.

- Repository: [Langflow/langflow](https://github.com/langflow-ai/langflow)
- Tags: internals
- Published: 2026-02-24

---

**Langflow implements multi-agent orchestration through a three-layer architecture that combines a graph-based flow runner, a conversation-aware base agent class, and concrete agent components that wrap LangChain executors with custom tool pipelines.**

Langflow (langflow-ai/langflow) treats every agent as a reusable graph vertex capable of participating in cyclic flows. The platform manages complex **multi-agent conversations** by reconstructing chat context at each node while a central runner coordinates execution order and session persistence. This analysis examines the specific source files and methods that enable agents to hand off context and tools seamlessly within a unified flow.

## Graph Execution Layer

The orchestration entry point is `LangflowRunnerExperimental` in [`src/backend/base/langflow/services/flow/flow_runner.py`](https://github.com/langflow-ai/langflow/blob/main/src/backend/base/langflow/services/flow/flow_runner.py). When a flow triggers, the runner loads the JSON definition and constructs a **`Graph`** object via `create_graph_from_flow`, mapping each vertex to a component including individual agents.

The runner executes vertices in topological order, but **cycles are explicitly supported** (e.g., loops back to previous agents). During `run_graph`, the runner injects the current `session_id`, `input_value`, and `chat_history` into each component’s execution context:

```python
graph = await self.create_graph_from_flow(session_id, flow_dict, user_id=user_id)
result = await self.run_graph(
        input_value=input_value,
        input_type=input_type,
        output_type=output_type,
        session_id=session_id,
        graph=graph,
        stream=stream,
)

```

The `process_agent_events` helper manages streaming partial outputs while preserving execution order, ensuring that downstream agents receive complete state updates before activation.

## Base Agent Component Architecture

All Langflow agents inherit from `ALTKBaseAgentComponent` located in [`src/lfx/src/lfx/base/agents/altk_base_agent.py`](https://github.com/langflow-ai/langflow/blob/main/src/lfx/src/lfx/base/agents/altk_base_agent.py). This base class extends the generic LangChain `AgentComponent` and provides three critical responsibilities for **conversation management**:

**Context Construction** – The `build_conversation_context()` method gathers stored `chat_history` (either `Data` objects or `Message` lists) and appends the current `input_value` as a `HumanMessage`. It normalizes content via `normalize_message_content()` to handle both string and list-formatted message payloads from `Data.to_lc_message()`.

**Query Extraction** – `get_user_query()` extracts raw text from the latest message, handling both plain strings and `Message` objects through attribute detection:

```python
def get_user_query(self) -> str:
    if hasattr(self.input_value, "get_text"):
        return self.input_value.get_text()
    return str(self.input_value)

```

**Agent Execution Loop** – The `run_agent` method initializes an `AgentExecutor`, calls `update_runnable_instance` to inject processed tools, normalizes image payloads, and streams events through `process_agent_events`. This loop ensures each agent rebuilds its conversation context independently while maintaining access to the shared session history.

## Tool Pipeline and Concrete Agents

The concrete implementation in [`src/lfx/src/lfx/components/altk/altk_agent.py`](https://github.com/langflow-ai/langflow/blob/main/src/lfx/src/lfx/components/altk/altk_agent.py) demonstrates how **ALTKAgentComponent** configures complex tool pipelines. The `configure_tool_pipeline()` method conditionally wraps tools with validation and post-processing layers:

- **`PreToolValidationWrapper`** – Applies SPARC validation as the outermost wrapper
- **`PostToolProcessingWrapper`** – Handles JSON post-processing as the innermost layer

```python
def configure_tool_pipeline(self) -> None:
    wrappers = []
    if self.enable_post_tool_reflection:
        wrappers.append(PostToolProcessingWrapper(...))
    if self.enable_tool_validation:
        wrappers.append(PreToolValidationWrapper())
    self.pipeline_manager.configure_wrappers(wrappers)

```

Before execution, `update_runnable_instance` converts LangChain tools to SPARC-compatible specs via `convert_langchain_tools_to_sparc_tool_specs_format`, processes the tool list through the configured pipeline, and assigns the wrapped tools to the runnable:

```python
def update_runnable_instance(self, agent, runnable, tools):
    # Build conversation context, initialise pipeline, inject specs, process tools

    ...
    runnable.tools = processed_tools
    return runnable

```

## Multi-Agent Conversation Flow

When multiple **agent components** exist in a single flow (e.g., Agent A → Agent B → LLM), the graph orchestrator ensures the output of one agent becomes the `input_value` for the next. Because `ALTKBaseAgentComponent` rebuilds the conversation context at each node using the session's `chat_history`, downstream agents automatically receive the full transcript including previous agent outputs.

This architecture enables **seamless multi-agent dialogues** where each participant maintains its own tool configuration and LLM provider while sharing a unified conversation state through the session ID. The runner persists intermediate outputs to the session store, allowing cyclic flows where agents iteratively refine responses based on prior agent outputs.

```python
from langflow.services.flow.flow_runner import LangflowRunnerExperimental
from uuid import uuid4
import asyncio

async def demo_multi_agent():
    # Load a flow containing two ALTK agents wired together

    flow_path = "examples/multi_agent_flow.json"
    runner = LangflowRunnerExperimental()
    
    # Execute with shared session state

    result = await runner.run(
        session_id=str(uuid4()),
        flow=flow_path,
        input_value="Book a flight and summarize the itinerary",
        input_type="chat",
        output_type="all",
    )
    print(result)  # Contains messages from both agents

asyncio.run(demo_multi_agent())

```

## Summary

- **LangflowRunnerExperimental** in [`flow_runner.py`](https://github.com/langflow-ai/langflow/blob/main/flow_runner.py) executes cyclic graphs while managing session persistence and streaming via `process_agent_events`.
- **ALTKBaseAgentComponent** in [`altk_base_agent.py`](https://github.com/langflow-ai/langflow/blob/main/altk_base_agent.py) normalizes conversation context through `build_conversation_context()` and extracts user queries via `get_user_query()`.
- **ToolPipelineManager** supports configurable wrappers including `PreToolValidationWrapper` and `PostToolProcessingWrapper` defined in [`altk_tool_wrappers.py`](https://github.com/langflow-ai/langflow/blob/main/altk_tool_wrappers.py).
- **ALTKAgentComponent** in [`altk_agent.py`](https://github.com/langflow-ai/langflow/blob/main/altk_agent.py) converts tools to SPARC specs and injects them into LangChain `AgentExecutor` instances through `update_runnable_instance`.
- Multi-agent flows share state through the `session_id` and `chat_history` parameters, with each agent reconstructing context independently while the graph runner manages execution order.

## Frequently Asked Questions

### How does Langflow maintain conversation history across multiple agents?

Langflow stores conversation history in a session-specific `chat_history` object that passes between graph vertices. Each `ALTKBaseAgentComponent` rebuilds its local conversation context by calling `build_conversation_context()`, which appends the current `input_value` to the stored history. This ensures downstream agents receive complete transcripts regardless of how many agents processed the message previously.

### What is the purpose of tool wrappers in Langflow agents?

Tool wrappers provide pre-execution validation and post-execution processing. The `PreToolValidationWrapper` enforces SPARC-compatible specifications before tool invocation, while `PostToolProcessingWrapper` handles JSON normalization afterward. These wrappers are configured via `configure_tool_pipeline()` in `ALTKAgentComponent` and execute in a nested fashion around the base tool call.

### Can Langflow flows contain cycles between agents?

Yes. The `LangflowRunnerExperimental` explicitly supports cyclic graphs, allowing flows where Agent A outputs to Agent B, which may route back to Agent A for refinement. The runner processes vertices in topological order where possible but permits backward edges, with session state ensuring that cyclic references receive updated conversation context on each iteration.

### How does Langflow differ from standard LangChain agent implementations?

Langflow extends standard LangChain patterns by introducing a graph-based orchestration layer and reusable component architecture. While standard LangChain agents typically run in isolation, Langflow's `ALTKBaseAgentComponent` integrates with the `Graph` system to enable multi-agent workflows where `update_runnable_instance` dynamically injects tool sets and conversation context specific to each node's position in the flow.