Architecting Enterprise Multi-Agent Systems: Solving Circular Imports in LangGraph RAG

Introduction

In large-scale Python codebases, few errors are as frustrating as the ImportError: cannot import name.... While easily dismissed in small scripts, circular imports become a critical architectural failure in enterprise AI systems. When building Multi-Agent LangGraph RAG systems with shared state and memory, the risk of circular dependencies skyrockets. Agents need tools, tools need to read/write state, state needs to trigger agent callbacks, and the orchestrator needs to know about all of them.

This article explores the architectural patterns required to eliminate circular imports, culminating in an end-to-end implementation of an Enterprise Financial RAG system using LangGraph.

Part 1: Why Agent Systems Trigger Circular Imports

In a naive multi-agent design, the dependency graph looks like a tangled web:

Agent A imports Tool B to perform actions.
Tool B imports StateManager C to save retrieved context.
StateManager C imports Agent A to notify it when context is updated.

Result: Circular Import. The application crashes on startup.

Part 2: Architectural Patterns to Break the Cycle

To build enterprise-grade agent systems, we must enforce strict boundaries using the following patterns:

1. Dependency Inversion Principle (DIP) via `typing.Protocol`

High-level modules (Agents) should not depend on low-level modules (Concrete Tools). Both should depend on abstractions (Interfaces). In Python, we use typing.Protocol for structural subtyping.

2. Dependency Injection (DI) & The Composition Root

Never instantiate dependencies inside a class. Pass them through the constructor. The actual wiring of concrete classes happens in a single file called the Composition Root (usually main.py).

3. Strict Layered Architecture (Hexagonal/Clean)

Core/Domain: Interfaces, State definitions, Exceptions. (Imports nothing external).
Infrastructure: Concrete Tools, LLM wrappers, Vector DB clients. (Imports Core).
Application: Agents. (Imports Core and Infrastructure interfaces).
Orchestration: LangGraph workflows. (Imports Application and Infrastructure).

4. State Decoupling (The "Dumb Tool" Pattern)

Tools should never import the Orchestrator or the Agents. Tools should only accept input, perform a task, and return a result. The Orchestrator (LangGraph) is solely responsible for updating the State based on the Tool's result.

Part 3: Real-World Use Case

Scenario: An Enterprise Financial Analysis Assistant.

Goal: Process a user's query about a company's 10-K filing.

Architecture:

Research Agent: Uses a RAG tool to retrieve relevant document chunks.
Analysis Agent: Uses a Calculator tool to compute financial metrics from the retrieved text.
State: Shared LangGraph state holding the conversation history, retrieved documents, and calculated metrics.
Memory: A persistent memory store to remember user preferences across sessions.

Part 4: End-to-End Code Implementation

Prerequisites:

pip install langgraph langchain-openai langchain-core pydantic

Step 1: The Core (Interfaces & State)

Notice that core imports absolutely nothing from our application. This is the foundation that prevents circular imports.

# core/interfaces.py
from typing import Protocol, Any, Dict, List
from langchain_core.messages import BaseMessage

class ITool(Protocol):
    """Abstraction for any tool. Agents depend on this, not concrete tools."""
    name: str
    description: str
    
    async def ainvoke(self, input: Dict[str, Any]) -> str:
        pass

class IMemoryStore(Protocol):
    """Abstraction for persistent memory."""
    async def save_preference(self, user_id: str, key: str, value: Any) -> None: ...
    async def get_preference(self, user_id: str, key: str) -> Any: ...

# core/state.py
from typing import TypedDict, Annotated
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    """
    The shared state of the LangGraph. 
    Notice it contains NO references to Agents or Tools.
    """
    messages: Annotated[List[BaseMessage], add_messages]
    retrieved_context: str
    calculated_metrics: Dict[str, float]
    user_id: str

Step 2: Infrastructure (Concrete Tools & Memory)

Tools implement the ITool protocol. They do NOT import agents or the graph.

# infrastructure/tools.py
from typing import Any, Dict
from core.interfaces import ITool

class VectorRAGTool:
    name = "vector_search"
    description = "Searches the vector database for financial document chunks."

    async def ainvoke(self, input: Dict[str, Any]) -> str:
        query = input.get("query", "")
        # Mocking vector search
        return f"[RAG Context]: Revenue for Q3 was $5.2B, up 12% YoY. Operating margin is 22%."

class FinancialCalculatorTool:
    name = "financial_calculator"
    description = "Calculates financial metrics based on provided text."

    async def ainvoke(self, input: Dict[str, Any]) -> str:
        text = input.get("text", "")
        # Mocking calculation
        return "Calculated Metrics: {'revenue_growth': 0.12, 'operating_margin': 0.22}"

# infrastructure/memory.py
from typing import Any, Dict
from core.interfaces import IMemoryStore

class InMemoryStore:
    """Mock implementation of persistent memory."""
    def __init__(self):
        self._store: Dict[str, Dict[str, Any]] = {}

    async def save_preference(self, user_id: str, key: str, value: Any) -> None:
        self._store.setdefault(user_id, {})[key] = value

    async def get_preference(self, user_id: str, key: str) -> Any:
        return self._store.get(user_id, {}).get(key)

Step 3: Application (The Agents)

Agents depend on the ITool interface, not the concrete VectorRAGTool. This is Dependency Inversion in action.

# application/agents.py
from typing import List
from langchain_core.messages import AIMessage, SystemMessage
from langchain_openai import ChatOpenAI
from core.interfaces import ITool
from core.state import AgentState

class ResearchAgent:
    def __init__(self, tools: List[ITool], llm: ChatOpenAI):
        self.tools = {t.name: t for t in tools}
        self.llm = llm

    async def __call__(self, state: AgentState) -> dict:
        """LangGraph node function."""
        user_query = state["messages"][-1].content
        
        # Use the injected tool (Dependency Injection)
        rag_tool = self.tools["vector_search"]
        context = await rag_tool.ainvoke({"query": user_query})
        
        response = await self.llm.ainvoke([
            SystemMessage(content="You are a financial researcher. Use the context to answer."),
            *state["messages"],
            AIMessage(content=f"Context retrieved: {context}")
        ])
        
        # Update state. Notice we return a dict, we don't mutate state directly.
        return {
            "messages": [response],
            "retrieved_context": context
        }

class AnalysisAgent:
    def __init__(self, tools: List[ITool], llm: ChatOpenAI):
        self.tools = {t.name: t for t in tools}
        self.llm = llm

    async def __call__(self, state: AgentState) -> dict:
        context = state.get("retrieved_context", "")
        
        calc_tool = self.tools["financial_calculator"]
        metrics_text = await calc_tool.ainvoke({"text": context})
        
        response = await self.llm.ainvoke([
            SystemMessage(content="You are a financial analyst. Analyze the metrics."),
            *state["messages"],
            AIMessage(content=f"Metrics data: {metrics_text}")
        ])
        
        return {
            "messages": [response],
            "calculated_metrics": {"revenue_growth": 0.12, "operating_margin": 0.22}
        }

Step 4: Orchestration (LangGraph Workflow)

The graph wires the agents together. It imports the Application layer, but the Application layer does not import the Graph.

# orchestration/graph.py
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from core.state import AgentState
from application.agents import ResearchAgent, AnalysisAgent

def should_continue_analysis(state: AgentState) -> str:
    """Conditional edge to route between agents."""
    last_message = state["messages"][-1].content.lower()
    if "calculate" in last_message or "metrics" in last_message:
        return "analyst"
    return END

def build_graph(researcher: ResearchAgent, analyst: AnalysisAgent):
    workflow = StateGraph(AgentState)

    # Add nodes
    workflow.add_node("researcher", researcher)
    workflow.add_node("analyst", analyst)

    # Add edges
    workflow.add_edge(START, "researcher")
    workflow.add_conditional_edges(
        "researcher",
        should_continue_analysis,
        {"analyst": "analyst", END: END}
    )
    workflow.add_edge("analyst", END)

    # Compile with checkpointer for short-term memory (thread persistence)
    memory = MemorySaver()
    return workflow.compile(checkpointer=memory)

Step 5: The Composition Root (Dependency Injection)

This is the ONLY file that imports concrete implementations. This breaks all potential circular dependencies.

# main.py
import asyncio
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI

# Import Interfaces (Core)
from core.interfaces import ITool, IMemoryStore

# Import Concrete Implementations (Infrastructure & Application)
from infrastructure.tools import VectorRAGTool, FinancialCalculatorTool
from infrastructure.memory import InMemoryStore
from application.agents import ResearchAgent, AnalysisAgent

# Import Orchestrator
from orchestration.graph import build_graph

async def main():
    # 1. Instantiate Infrastructure
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    tools: list[ITool] = [VectorRAGTool(), FinancialCalculatorTool()]
    memory_store: IMemoryStore = InMemoryStore()

    # 2. Save some user preference in persistent memory
    await memory_store.save_preference("user_123", "report_format", "markdown")

    # 3. Instantiate Application (Injecting dependencies)
    researcher = ResearchAgent(tools=tools, llm=llm)
    analyst = AnalysisAgent(tools=tools, llm=llm)

    # 4. Build the Graph
    app = build_graph(researcher=researcher, analyst=analyst)

    # 5. Execute
    config = {"configurable": {"thread_id": "session_1"}}
    
    initial_state = {
        "messages": [HumanMessage(content="What is the revenue growth and operating margin for Q3? Calculate the metrics.")],
        "retrieved_context": "",
        "calculated_metrics": {},
        "user_id": "user_123"
    }

    print("Starting Enterprise RAG Workflow...")
    async for event in app.astream(initial_state, config=config):
        for node_name, output in event.items():
            print(f"\n--- Node: {node_name} ---")
            if "messages" in output:
                print(output["messages"][-1].content)
            if "calculated_metrics" in output and output["calculated_metrics"]:
                print(f"Metrics Updated: {output['calculated_metrics']}")

if __name__ == "__main__":
    asyncio.run(main())

Part 5: Why This Architecture Eliminates Circular Imports

Let's trace the import graph of the code above to prove it is acyclic:

core/ imports nothing from our project. It is the leaf node.
infrastructure/tools.py imports core.interfaces. (Acyclic).
application/agents.py imports core.interfaces and core.state. It does not import infrastructure/tools.py. It receives tools via the __init__ constructor. (Acyclic).
orchestration/graph.py imports core.state and application.agents. It does not import tools. (Acyclic).
main.py imports everything. Because it is the entry point, it is allowed to know about all concrete classes. It acts as the Composition Root.

The "Secret Weapon": `typing.TYPE_CHECKING`

In enterprise codebases, you sometimes need type hinting for a class without actually importing it at runtime. Python provides TYPE_CHECKING for this:

# application/agents.py
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    # This import only runs for static type checkers (mypy, IDEs)
    # It is ignored at runtime, preventing circular imports!
    from orchestration.graph import AgentGraph 

class ResearchAgent:
    def set_graph_reference(self, graph: "AgentGraph"):
        pass

Conclusion

Handling circular imports in large-scale Python AI codebases isn't about using "hacks" like importing inside functions. It is about strict architectural discipline.

By utilizing Dependency Inversion (Protocols), Dependency Injection, and a Strict Layered Architecture, we can build highly complex, multi-agent LangGraph systems with shared state and memory that are robust, testable, and completely free of circular dependencies.

When your tools don't know about your agents, and your agents don't know about your graph, your system becomes infinitely scalable.