Mastering Stateful Multi-Agent Orchestration

1. The Evolution of Agentic Architectures

Early LLM pipelines relied heavily on Directed Acyclic Graphs (DAGs)—linear sequence models where step A feeds step B, which feeds step C. While suitable for simple task automation, DAGs fail to handle real-world software engineering or legal reasoning workflows that require:

• Verification Loops: Running tests, checking outputs, and looping back to a previous node on failure.
• Human-in-the-Loop (HITL) Interruption: Pausing state graph execution to gather user feedback or authorization before moving forward.
• Parallel Consensus: Spawning multiple subagents, running tasks concurrently, and merging state updates via a custom reducer.

2. Key Graph Components in LangGraph

Unlike naive chaining systems, stateful orchestration represents agents as a single unified state machine:

3. Running Local Models: Ollama with DeepSeek-R1 and Qwen

For enterprise environments, keeping data local is a necessity due to privacy, compliance, and custom token constraints. We run:

• DeepSeek-R1: Excellent for logic-heavy, multi-turn reasoning steps (planning, code audit, legal verification) where its chain-of-thought (CoT) token outputs solve complex constraints.
• Qwen-2.5-Coder: A highly optimized, fast local coding model capable of precise tool call structure generation and high-speed iterative writing.

# Standard Ollama local client integration snippet
from langchain_community.chat_models import ChatOllama

# Setup Qwen for fast, precise structural JSON output (Tool Calling)
coder_llm = ChatOllama(model="qwen2.5-coder:7b", format="json", temperature=0.0)

# Setup DeepSeek-R1 for complex planning nodes
planner_llm = ChatOllama(model="deepseek-r1:14b", temperature=0.6)

4. Extending Capabilities with Model Context Protocol (MCP)

The Model Context Protocol (MCP), pioneered by Anthropic, establishes an open, standard protocol for connecting local and remote data sources, environments, and specialized tools to LLMs.

5. Implementation Example: The Self-Correcting Coder

Below is a complete, minimal implementation of a stateful, self-correcting coding agent loop that writes, runs, and auto-corrects code using LangGraph.

import sys
from typing import TypedDict, List, Literal
from langgraph.graph import StateGraph, END

# 1. Define the Shared State
class AgentState(TypedDict):
    prompt: str
    code: str
    error_message: str
    attempts: int
    max_attempts: int

# 2. Define Node Behaviors
def code_generator_node(state: AgentState) -> dict:
    prompt = state["prompt"]
    feedback = state.get("error_message", "")
    attempts = state.get("attempts", 0) + 1
    
    # Construct prompt with self-correction context if errors exist
    system_prompt = "Write clean Python code to solve the prompt."
    if feedback:
        system_prompt += f"\nPrevious attempt failed with error: {feedback}. Correct the bugs."
        
    # [Local LLM Call would execute here]
    mock_code = "def add(a, b): return a + b" if attempts > 1 else "def add(a, b): return a - b"
    return {"code": mock_code, "attempts": attempts}

def test_runner_node(state: AgentState) -> dict:
    code = state["code"]
    error = ""
    try:
        # Run custom unit test in sandboxed execution environment
        exec_scope = {}
        exec(code, exec_scope)
        assert exec_scope["add"](2, 3) == 5, "add(2,3) must equal 5"
    except Exception as e:
        error = str(e)
    return {"error_message": error}

# 3. Define Conditional Routing
def route_after_testing(state: AgentState) -> Literal["coder", "done"]:
    if not state["error_message"]:
        return "done"
    if state["attempts"] >= state["max_attempts"]:
        return "done"
    return "coder"

# 4. Compile the Graph
workflow = StateGraph(AgentState)

# Add Nodes
workflow.add_node("coder", code_generator_node)
workflow.add_node("tester", test_runner_node)

# Set Entry and Edges
workflow.set_entry_point("coder")
workflow.add_edge("coder", "tester")

# Set Conditional Path
workflow.add_conditional_edges(
    "tester",
    route_after_testing,
    {
        "coder": "coder",
        "done": END
    }
)

app = workflow.compile()

# 5. Run the Stateful Agent
initial_state = {"prompt": "Write a function add(a,b) that adds two numbers.", "max_attempts": 3, "attempts": 0}
result = app.invoke(initial_state)

print(f"Final Code: {result['code']}")
print(f"Total Attempts: {result['attempts']}")
print(f"Status: {'Success' if not result['error_message'] else 'Failed'}")

6. Conclusion and Next Steps

Stateful agentic graphs remove the unpredictability of one-shot LLM inferences. By creating resilient, closed-loop execution patterns, we turn fragile prompts into durable, industrial-grade software engineering engines.