12. The Intelligence Layer: Multi-Agent Orchestration with LangGraph

---

🤔 Why Multiple Agents

If you are building a financial AI platform, a single general-purpose agent handling all responsibilities is theoretically possible — but practically fragile. Its context window limits how much domain knowledge it holds simultaneously, its broad permissions are a large blast radius when compromised, and it cannot be updated or scaled independently. This section explains why specialised agents with clean separation of concerns are the right architecture for production agentic systems.

Specialised agents solve these problems:

• 🛡️ The ValidationAgent knows the data quality rules for financial transactions. It is trained on the validation schema, configured with the acceptable value ranges, and responsible for one decision: is this data valid?
• 🔏 The PIIScrubberAgent knows the PII entity types defined in the platform’s GDPR compliance requirements. It makes one class of decision: what in this text is personal data and must be replaced?
• ⚖️ The compliance audit agent knows the regulatory requirements encoded in the OPA Rego policies. It interprets OPA decisions in the context of the regulatory framework they implement.

↑ Back to top · Next: LangGraph: Stateful Agent Graphs →

---

🕸️ LangGraph: Stateful Agent Graphs

LangGraph provides the orchestration framework. Its core abstraction is the StateGraph: a directed graph where nodes represent agent invocations and edges represent control flow between agents.

The state that flows through the graph is typed — it is a Python TypedDict (or Pydantic model) that carries all information an agent might need:

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Optional

class AgentState(TypedDict):
    user_query: str
    transactions: list
    validation_result: Optional[dict]
    compliance_decision: Optional[dict]
    finance_response: Optional[str]
    human_approval: Optional[bool]
    final_response: Optional[str]

Each node in the graph is a function that takes an AgentState and returns an updated AgentState:

def validation_node(state: AgentState) -> AgentState:
    transactions = state.get("transactions", [])
    results = [validation_agent.validate(tx) for tx in transactions]
    state["validation_result"] = {
        "passed": all(r.passed for r in results),
        "issues": [r.issues for r in results if not r.passed],
    }
    return state

def compliance_node(state: AgentState) -> AgentState:
    if not state["validation_result"]["passed"]:
        state["compliance_decision"] = {"approved": False, "reason": "Validation failed"}
        return state
    decision = compliance_agent.evaluate(state["transactions"])
    state["compliance_decision"] = decision
    return state

The graph is assembled by declaring nodes and edges:

graph = StateGraph(AgentState)
graph.add_node("validation", validation_node)
graph.add_node("compliance", compliance_node)
graph.add_node("human_approval", human_approval_node)
graph.add_node("final_response", final_response_node)

graph.add_edge(START, "validation")
graph.add_edge("validation", "compliance")
graph.add_edge("compliance", "human_approval")
graph.add_edge("human_approval", "final_response")
graph.add_edge("final_response", END)

This produces a linear workflow: validate → evaluate compliance → check for human approval → compose final response. LangGraph also supports conditional edges (branching), parallel nodes (fan-out and fan-in), and cycles (loops with termination conditions) — enabling complex multi-step reasoning flows.

↑ Back to top · Next: Dynamic Agent Composition →

---

⚙️ Dynamic Agent Composition

Hardcoding agent names in the graph definition creates a maintenance problem: adding a new agent requires modifying the graph construction code. The platform uses a configuration-driven approach instead.

A YAML composition file defines the agent sequence:

# config/agent_workflow.yml
agents:
  - name: sharia_audit    # compliance audit agent
    config:
      model: mistral
      temperature: 0.1
  - name: personal_finance
    config:
      model: mistral
      temperature: 0.3

The _build_graph() function reads this configuration and instantiates agents dynamically from the registry:

def _build_graph(config_file: Optional[str] = None) -> StateGraph:
    graph = StateGraph(AgentState)
    factory = EnhancedAgentFactory()
    network = create_agent_network(factory)

    agent_entries = get_agent_deployment_config(config_file)
    if not agent_entries:
        raise RuntimeError("No agents configured for orchestrator workflow")

    node_names = []
    for entry in agent_entries:
        agent_name = entry["name"]
        agent_config = AgentConfig(**entry.get("config", {}) or {})

        def make_node(node_name: str, config: AgentConfig):
            def node(state: AgentState) -> AgentState:
                message = AgentMessage(
                    message_id=str(uuid.uuid4()),
                    source="orchestrator",
                    target=node_name,
                    message_type="invoke",
                    payload={"state": state},
                    metadata={"config": config.to_dict()},
                )
                response = network.request(message)
                return response.payload.get("state", state)
            node.__name__ = f"{node_name}_node"
            return node

        graph.add_node(agent_name, make_node(agent_name, agent_config))
        node_names.append(agent_name)

    # Add shared workflow nodes
    graph.add_node("human_approval", human_approval_node)
    graph.add_node("final_response", final_response_node)

    previous = START
    for node_name in node_names:
        graph.add_edge(previous, node_name)
        previous = node_name
    graph.add_edge(previous, "human_approval")
    graph.add_edge("human_approval", "final_response")
    graph.add_edge("final_response", END)

    return graph

Adding a new agent to the workflow is now a YAML change, not a code change. The new agent must be registered in the EnhancedAgentFactory, but the orchestration graph adapts automatically.

↑ Back to top · Next: The A2A Protocol →

---

✉️ The A2A Protocol

The AgentMessage protocol defines how agents communicate with each other:

from dataclasses import dataclass, field
from typing import Any, Dict

@dataclass
class AgentMessage:
    message_id: str
    source: str          # Sending agent or "orchestrator"
    target: str          # Receiving agent name
    message_type: str    # "invoke", "response", "error"
    payload: Dict[str, Any] = field(default_factory=dict)
    metadata: Dict[str, Any] = field(default_factory=dict)

This message structure is transport-agnostic. An AgentMessage can be serialised to JSON and sent over Kafka, or passed as a Python object in an in-process call. The transport layer handles the serialisation; the agents and orchestrator work with AgentMessage objects regardless of transport.

The protocol enforces a key discipline: agents do not call each other directly. They send messages. The network layer routes messages to the correct recipient. This indirection enables the transport to be swapped without changing agent code.

↑ Back to top · Next: LocalAgentNetwork vs KafkaAgentNetwork →

---

🔀 LocalAgentNetwork vs KafkaAgentNetwork

The platform provides two agent network implementations, selected by environment variable:

🖥️ LocalAgentNetwork: Development and Testing

class LocalAgentNetwork(AgentNetwork):
    def __init__(self, factory: EnhancedAgentFactory):
        self.factory = factory

    def request(self, message: AgentMessage, timeout_seconds: float = 30.0) -> AgentMessage:
        managed_agent = self.factory.create_agent(
            message.target,
            singleton=True,
            config=AgentConfig(**message.metadata.get("config", {})),
        )
        if managed_agent is None:
            raise RuntimeError(f"Target agent '{message.target}' not available")

        response_state = managed_agent.process(message.payload.get("state", {}))
        return AgentMessage(
            message_id=str(uuid.uuid4()),
            source=message.target,
            target=message.source,
            message_type="response",
            payload={"state": response_state},
            metadata={"origin": "local"},
        )

LocalAgentNetwork invokes agents as in-process function calls. There is no serialisation, no network overhead, and no broker to configure. It is ideal for development and testing because the entire multi-agent workflow can be run with a single pytest command.

The singleton=True parameter ensures that agent instances are reused across invocations within the same session — saving the cost of model loading and warm-up on every message.

☁️ KafkaAgentNetwork: Distributed Production

class KafkaAgentNetwork(AgentNetwork):
    def __init__(self, bootstrap_servers: str, topic: str):
        self.bootstrap_servers = bootstrap_servers
        self.topic = topic

    def request(self, message: AgentMessage, timeout_seconds: float = 30.0) -> AgentMessage:
        raise NotImplementedError("Full Kafka A2A implementation in progress")

KafkaAgentNetwork is the production transport for deployments where agents run in separate Kubernetes pods or even separate clusters. Messages are serialised and published to a dedicated A2A Kafka topic (agent_a2a). A consumer in the target agent’s pod receives the message, processes it, and publishes the response.

The key advantage of the Kafka-based network is scalability: agent pods can be scaled independently based on their specific load profiles. A compliance audit agent with a slow LLM backend can run on more pods than a validation agent that does fast deterministic checks.

Selecting the transport is a single environment variable:

AGENT_NETWORK_MODE=local        # Uses LocalAgentNetwork (default)
AGENT_NETWORK_MODE=distributed  # Uses KafkaAgentNetwork
AGENT_NETWORK_TRANSPORT=kafka   # Transport protocol for distributed mode
AGENT_NETWORK_TOPIC=agent_a2a   # Kafka topic for A2A messages

↑ Back to top · Next: LLM Provider Abstraction →

---

🤖 LLM Provider Abstraction

Agents that use LLMs are abstracted from the specific LLM provider through the LLMProvider class:

class LLMProvider:
    DEFAULT_PROVIDER = "local"
    DEFAULT_LOCAL_URL = "http://localhost:11434/v1"
    DEFAULT_LOCAL_MODEL = "mistral"

    @staticmethod
    def create_client() -> Optional[openai.OpenAI]:
        provider = LLMProvider.get_provider().lower()
        if provider == "local":
            client = LLMProvider._create_local_client()
            if client:
                return client
            logger.warning("Local LLM unavailable; falling back to cloud provider")
        return LLMProvider._create_cloud_client()

    @staticmethod
    def _create_local_client() -> Optional[openai.OpenAI]:
        base_url = LLMProvider._get_local_base_url()
        client = openai.OpenAI(
            api_key="ollama",
            base_url=base_url,
        )
        if not LLMProvider.check_local_availability():
            return None
        return client

The abstraction uses the OpenAI client library for both local (Ollama) and cloud providers. Ollama exposes an OpenAI-compatible API at /v1, meaning the same client code works for both providers. Switching from local to cloud is a single environment variable change:

LLM_PROVIDER=local    # Uses Ollama at LLM_LOCAL_BASE_URL (default)
LLM_PROVIDER=cloud    # Uses cloud provider at configured endpoint

The automatic fallback (if not LLMProvider.check_local_availability()) means the platform degrades gracefully when local inference is unavailable. In CI/CD, where Ollama is typically not running, the platform automatically falls back to the configured cloud provider.

↑ Back to top · Next: Workflow Checkpointing →

---

💾 Workflow Checkpointing

LangGraph supports persistent checkpointing of workflow state. When a workflow is interrupted — by a human approval gate, a transient service failure, or a pod restart — the orchestrator can resume from the last checkpoint rather than starting over.

The platform uses PostgreSQL for checkpoint storage when available, with in-memory fallback for development:

def _get_checkpointer():
    if PostgresSaver is None or pg is None:
        return None
    try:
        db_config = load_db_config()
        conn = pg.connect(**db_config)
        saver = PostgresSaver(conn)
        saver.setup()  # Creates checkpoint tables if they don't exist
        return saver
    except Exception as error:
        logger.warning("PostgresSaver unavailable: %s", error)
        return None

def create_orchestrator(config_file=None):
    graph = _build_graph(config_file)
    checkpointer = _get_checkpointer()
    if checkpointer is None and MemorySaver is not None:
        checkpointer = MemorySaver()
    compiled = graph.compile(checkpointer=checkpointer) if checkpointer else graph.compile()
    return OrchestratorRunner(compiled, checkpointer=checkpointer)

PostgreSQL checkpointing is essential for the human-approval gate. When the workflow reaches a high-value transaction that requires human review, the workflow pauses at the human_approval node. The human reviewer approves or denies the transaction through the dashboard. The workflow resumes from the checkpoint regardless of whether the original pod is still running.

↑ Back to top · Next: Human-in-the-Loop →

---

👤 Human-in-the-Loop

The human_approval_node demonstrates how agentic workflows maintain human oversight for high-stakes decisions:

def human_approval_node(state: AgentState) -> AgentState:
    query_text = state["user_query"]
    if any(kw in query_text.lower() for kw in HUMAN_APPROVAL_CHECK_KEYWORDS):
        amounts = re.findall(AMOUNT_REGEX, query_text)
        for amount_str in amounts:
            parsed = float(amount_str.replace(",", ""))
            if parsed > HUMAN_APPROVAL_THRESHOLD:
                state["human_approval"] = interrupt(
                    f"Transaction over ${HUMAN_APPROVAL_THRESHOLD:,} requires human approval. Approve? (True/False)"
                )
                break
    return state

The interrupt() call pauses the LangGraph workflow and surfaces the decision to the human reviewer. The HUMAN_APPROVAL_THRESHOLD (configurable via environment variable, default $5,000 for debt restructuring decisions) determines when human oversight is required.

This pattern maps to NIST AI RMF‘s GOVERN function: consequential AI decisions must have defined human oversight procedures. The orchestrator’s human approval gate is not optional — it is hard-coded into the graph topology and cannot be bypassed by prompt injection or manipulation of the agent inputs.

↑ Back to top · Next: Agent Performance Monitoring →

---

📈 Agent Performance Monitoring

Each agent invocation is recorded in Prometheus:

# Counters
agent_invocations_total{agent="validation_agent", status="success"}
agent_invocations_total{agent="validation_agent", status="error"}

# LLM token usage
agent_llm_tokens_total{agent="compliance_agent", type="input"}
agent_llm_tokens_total{agent="compliance_agent", type="output"}

# Human approval rate
agent_human_approvals_total{decision="approved"}
agent_human_approvals_total{decision="denied"}

The human approval rate is an important metric for the NIST AI RMF‘s MEASURE function. If the approval rate drops sharply, it may indicate that the approval threshold needs adjustment — either agents are escalating too many routine transactions (threshold too low) or not enough high-risk ones (threshold too high).

↑ Back to top · Next: Key Takeaways →

---

---

❓ Frequently Asked Questions

Common questions about multi-agent orchestration with LangGraph answered from real-world implementation experience.

What is LangGraph and how does it differ from LangChain for multi-agent systems?

LangGraph is a stateful orchestration framework built on LangChain that models agent workflows as directed graphs rather than linear chains. In LangGraph, each node is an agent function that receives a typed AgentState and returns an updated state; edges define control flow between agents. This enables conditional branching (different agents based on validation outcome), parallel fan-out, cycles with termination conditions, and persistent checkpointing across pod restarts — capabilities that LangChain’s sequential chain model does not natively support.

How does the A2A (Agent-to-Agent) protocol work in a distributed agent network?

The AgentMessage protocol defines a transport-agnostic message envelope: message_id, source, target, message_type (invoke/response/error), payload (carries the AgentState), and metadata. Agents never call each other directly — they send AgentMessage objects to the network layer, which routes them. In LocalAgentNetwork, routing is an in-process function call. In KafkaAgentNetwork, messages are serialised to JSON and published to the agent_a2a topic. The same agent code handles both because it only ever interacts with the message object, never the transport.

How do you implement human-in-the-loop approval gates in LangGraph?

The human approval gate is implemented as a graph node that calls LangGraph’s interrupt() function when the transaction amount exceeds HUMAN_APPROVAL_THRESHOLD. The interrupt() call pauses the workflow and surfaces the decision to the human reviewer via the operational dashboard. The workflow state is persisted to PostgreSQL via PostgresSaver at the checkpoint. When the reviewer approves or denies, the workflow resumes from the checkpoint — regardless of whether the original pod is still running. The gate is hard-coded into the graph topology and cannot be bypassed by prompt injection.

What is the difference between LocalAgentNetwork and KafkaAgentNetwork?

LocalAgentNetwork routes agent messages as in-process Python function calls — no serialisation, no broker, no network overhead. It is ideal for development and testing because the entire multi-agent workflow runs with a single pytest command. KafkaAgentNetwork routes messages through a dedicated Kafka topic (agent_a2a), enabling agents to run in separate Kubernetes pods or clusters, with each agent type independently scalable based on its load profile. Both implement the same AgentNetwork interface; switching between them is a single AGENT_NETWORK_MODE environment variable change.

↑ Back to top · Next: Key Takeaways →

---

---