Architecting Cognitive Separation in Enterprise LangGraph Agents

A Blueprint for Planning, Retrieval, Reasoning, and Execution in Multi-Agent RAG

In enterprise AI, monolithic "do-it-all" agents fail. They hallucinate tool parameters, retrieve irrelevant context because they didn't plan first, and lose track of long-term goals. The solution is Cognitive Architecture: explicitly separating thinking from doing.

This guide demonstrates how to decouple these four cognitive functions into distinct nodes within a LangGraph state machine, orchestrated by a Supervisor agent with persistent memory.

1. The Architecture: Four Pillars of Cognition

We do not build one giant LLM call. We build a graph where each node owns a single cognitive responsibility:

Cognitive Function	Responsibility	LangGraph Node Type	Key Characteristic
Planning	Decomposes user intent into a structured DAG of sub-tasks. No tools, no retrieval. Pure strategy.	planner_node	Outputs JSON schema; deterministic routing.
Retrieval	Executes search only based on planner directives. Re-ranks and filters. Never reasons about answers.	retriever_node	Tool-bound; returns structured documents.
Reasoning	Synthesizes retrieved context + task goal. Detects gaps. Decides if more retrieval or action is needed.	reasoner_node	Conditional edges; self-reflection loop.
Action Execution	Calls external APIs/DBs with validated parameters. Returns raw results only.	executor_node	Sandboxed; idempotent; error-handled.

Why Separate Them?

• Debuggability: When an answer is wrong, you can inspect whether the plan was bad, the retrieval missed docs, or the reasoner misinterpreted them.

• Cost Control: Use cheap/fast models for planning and routing; reserve expensive reasoning models for synthesis.

• Safety: The executor never decides what to do; it only executes approved plans. This enables human-in-the-loop approval gates between reasoning and execution.

2. Real-Time Enterprise Use Case: Automated Compliance Audit

Scenario: A financial services firm needs to audit vendor contracts against new SEC regulations. The query: "Audit Acme Corp's Q3 data processing agreement against SEC Rule 206(4)-7 and flag any missing breach notification clauses."

This requires:

1. Planning: Identify regulation → locate contract → extract clauses → compare.

2. Retrieval: Fetch SEC rule text AND internal contract from SharePoint.

3. Reasoning: Map contract language to regulatory requirements; identify gaps.

4. Action: Write findings to Jira and notify the compliance officer via Slack.

A single agent would likely retrieve the contract but miss the specific SEC subsection, or hallucinate clause mappings. Our separated architecture prevents this.

3. Implementation: Enterprise LangGraph with Memory & State

Prerequisites

pip install langgraph langchain-openai langchain-community langgraph-checkpoint-postgres

Step 1: Define the Shared State Schema

State is the contract between cognitive modules. It must be typed and versioned.

from typing import Annotated, Literal, TypedDict, Sequence
from langgraph.graph.message import add_messages
from langchain_core.documents import Document

class AgentState(TypedDict):
    # Message history for conversational memory
    messages: Annotated[Sequence, add_messages]
    
    # Planning artifacts
    current_plan: list[dict]          # Structured task decomposition
    active_task_index: int            # Pointer to current sub-task
    
    # Retrieval artifacts  
    retrieved_docs: list[Document]    # Filtered, re-ranked context
    retrieval_query: str              # Explicit query from planner
    
    # Reasoning artifacts
    reasoning_trace: str              # Chain-of-thought for auditability
    gap_detected: bool                # Triggers re-retrieval loop
    
    # Execution artifacts
    action_results: dict              # Raw API/tool outputs
    final_report: str                 # Synthesized deliverable
    
    # Memory metadata
    thread_id: str                    # For checkpoint persistence
    user_id: str                      # For RBAC-scoped retrieval

Step 2: Implement Cognitive Nodes (Separation Enforced)

Planner Node — Strategy Only

import json
from langchain_openai import ChatOpenAI

planner_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)  # Fast, cheap

PLAN_PROMPT = """You are a compliance audit planner. Given the user request, 
output ONLY a JSON array of sub-tasks. Each task has: 
- 'id': integer
- 'description': what to accomplish  
- 'requires_retrieval': bool
- 'requires_action': bool
Do NOT retrieve documents or call tools. Only plan."""

async def planner_node(state: AgentState) -> dict:
    response = await planner_llm.ainvoke([
        {"role": "system", "content": PLAN_PROMPT},
        *state["messages"]
    ])
    plan = json.loads(response.content)
    return {
        "current_plan": plan,
        "active_task_index": 0,
        "messages": [{"role": "assistant", "content": f"Plan created: {len(plan)} tasks"}]
    }

Retriever Node — Search Only

from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document

# Enterprise: scoped retriever respects user permissions
def get_scoped_retriever(user_id: str):
    # In production: filter by user_id metadata in vector DB
    return FAISS.load_local(f"./compliance_kb_{user_id}", ...)

async def retriever_node(state: AgentState) -> dict:
    task = state["current_plan"][state["active_task_index"]]
    if not task.get("requires_retrieval"):
        return {"retrieved_docs": []}
    
    retriever = get_scoped_retriever(state["user_id"])
    docs = await retriever.asimilarity_search(
        state["retrieval_query"], 
        k=5,
        filter={"doc_type": "regulation" if "SEC" in task["description"] else "contract"}
    )
    return {"retrieved_docs": docs}

Reasoner Node — Synthesis & Gap Detection

REASONER_PROMPT = """You are a compliance reasoner. Given the task goal and retrieved docs:
1. Determine if the docs sufficiently address the task.
2. If YES: synthesize findings into 'analysis'.
3. If NO: set gap_detected=true and suggest refined retrieval_query.
Output JSON: {"analysis": "...", "gap_detected": bool, "refined_query": "..."}"""

reasoner_llm = ChatOpenAI(model="gpt-4o", temperature=0)  # Strong reasoning model

async def reasoner_node(state: AgentState) -> dict:
    task = state["current_plan"][state["active_task_index"]]
    docs_text = "\n".join(d.page_content for d in state["retrieved_docs"])
    
    response = await reasoner_llm.ainvoke([
        {"role": "system", "content": REASONER_PROMPT},
        {"role": "user", "content": f"Task: {task['description']}\nDocs: {docs_text}"}
    ])
    result = json.loads(response.content)
    
    return {
        "reasoning_trace": result["analysis"],
        "gap_detected": result["gap_detected"],
        "retrieval_query": result.get("refined_query", ""),
        "messages": [{"role": "assistant", "content": result["analysis"]}]
    }

Executor Node — Action Only

TOOLS = {...}  # Jira, Slack, DB tools registered separately

async def executor_node(state: AgentState) -> dict:
    task = state["current_plan"][state["active_task_index"]]
    if not task.get("requires_action"):
        return {}
    
    # Executor receives EXPLICIT instructions from reasoner, not raw user query
    tool_call = parse_tool_call_from_reasoning(state["reasoning_trace"])
    result = await TOOLS[tool_call.name].ainvoke(tool_call.args)
    
    return {"action_results": {task["id"]: result}}

Step 3: Wire the Graph with Conditional Routing & Memory

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

def route_after_reasoner(state: AgentState) -> Literal["retriever_node", "executor_node", "advance_task"]:
    if state["gap_detected"]:
        return "retriever_node"      # Self-correcting retrieval loop
    task = state["current_plan"][state["active_task_index"]]
    if task.get("requires_action"):
        return "executor_node"
    return "advance_task"           # Move to next planned sub-task

def advance_task_node(state: AgentState) -> dict:
    next_idx = state["active_task_index"] + 1
    if next_idx >= len(state["current_plan"]):
        return {"active_task_index": next_idx}  # Will trigger END
    return {"active_task_index": next_idx, "retrieval_query": ""}

# Build graph
graph = StateGraph(AgentState)
graph.add_node("planner", planner_node)
graph.add_node("retriever", retriever_node)
graph.add_node("reasoner", reasoner_node)
graph.add_node("executor", executor_node)
graph.add_node("advance", advance_task_node)

graph.set_entry_point("planner")
graph.add_edge("planner", "retriever")
graph.add_edge("retriever", "reasoner")
graph.add_conditional_edges("reasoner", route_after_reasoner, {
    "retriever_node": "retriever",
    "executor_node": "executor",
    "advance_task": "advance"
})
graph.add_edge("executor", "advance")
graph.add_conditional_edges("advance", lambda s: "END" if s["active_task_index"] >= len(s["current_plan"]) else "retriever")

# Enterprise memory: Postgres-backed checkpoints for crash recovery & audit
checkpointer = AsyncPostgresSaver.from_conn_string("postgresql://...")
app = graph.compile(checkpointer=checkpointer)

Step 4: Run with Persistent Thread Memory

import asyncio

async def run_audit():
    config = {
        "configurable": {
            "thread_id": "audit-acme-q3-2026",
            "user_id": "compliance_officer_42"
        }
    }
    
    result = await app.ainvoke({
        "messages": [{"role": "user", "content": 
            "Audit Acme Corp Q3 data processing agreement against SEC 206(4)-7"}],
        "thread_id": "audit-acme-q3-2026",
        "user_id": "compliance_officer_42"
    }, config=config)
    
    print(result["final_report"])

asyncio.run(run_audit())

4. Enterprise Production Considerations

Observability per Cognitive Layer

Instrument each node independently with OpenTelemetry:

• Planner: Track plan length, replan frequency, token cost

• Retriever: Track recall@k, latency, filter hit rate

• Reasoner: Track gap detection rate, synthesis token usage

• Executor: Track tool success/failure rate, P99 latency

Human-in-the-Loop Between Reasoning and Execution

Add an interrupt before the executor for high-risk actions:

from langgraph.types import interrupt, Command

async def executor_node(state: AgentState):
    proposed_action = parse_tool_call_from_reasoning(state["reasoning_trace"])
    # Blocks graph until human approves/rejects
    approval = interrupt({"action": proposed_action, "reasoning": state["reasoning_trace"]})
    if approval["decision"] != "approve":
        return {"action_results": {"blocked": True, "feedback": approval["feedback"]}}
    # ... execute

Memory Hierarchy

• Short-term: messages in state (within-thread conversation)

• Working memory: current_plan, retrieved_docs (ephemeral per invocation)

• Long-term: Postgres checkpoint (cross-session resume, audit trail)

• Semantic memory: Vector store (enterprise knowledge base)

5. Key Takeaways

1. Separation is architectural, not prompt-based. Each cognitive function is a distinct graph node with its own LLM, prompt, and output schema.

2. State is the interface. Nodes communicate only through typed state fields, never through implicit context.

3. Retrieval follows planning. Never let the LLM decide what to search for mid-reasoning; the planner emits explicit retrieval queries.

4. Reasoning includes self-correction. The gap-detection loop is what makes RAG reliable in regulated domains.

5. Memory is multi-layered. Checkpoints provide durability; vector stores provide knowledge; state provides working context.

This architecture scales from compliance audits to customer support triage to code review pipelines. The cognitive separation pattern is model-agnostic and survives LLM upgrades—because your business logic lives in the graph topology, not in fragile mega-prompts.