How to Integrate LangGraph for investment banking with Kubernetes for AI agents

LangGraph gives you the orchestration layer for multi-step banking workflows: retrieval, policy checks, human approvals, and deterministic routing. Kubernetes gives you the runtime layer to run those agents as isolated, scalable services with proper resource controls, retries, and rollout discipline.

If you’re building AI agents for investment banking, the combination is practical: one graph can handle deal intake, compliance screening, and analyst handoff, while Kubernetes keeps the whole thing deployable across environments without turning your agent into a snowflake.

Prerequisites

• •Python 3.10+
• •A Kubernetes cluster:
  • •local: kind or minikube
  • •cloud: EKS, GKE, or AKS
• •kubectl configured and pointing at your cluster
• •Docker installed for building agent images
• •LangGraph installed:
  • •pip install langgraph langchain-core
• •Kubernetes Python client installed:
  • •pip install kubernetes
• •Access to an LLM provider or internal model endpoint
• •A namespace for your banking workloads
• •RBAC permissions to create:
  • •Deployments
  • •Services
  • •ConfigMaps
  • •Secrets

Integration Steps

1. •Build the LangGraph workflow for your banking use case

Start with a graph that does one thing well: route an incoming banking request through enrichment, policy checks, and approval gating.

from typing import TypedDict, Literal

from langgraph.graph import StateGraph, END

class DealState(TypedDict):
    request_id: str
    counterparty: str
    risk_score: int
    decision: Literal["approve", "reject", "review"]

def enrich_deal(state: DealState) -> DealState:
    # Replace with real enrichment from CRM / market data / KYC systems
    state["risk_score"] = 72 if state["counterparty"] == "high_risk_client" else 20
    return state

def compliance_check(state: DealState) -> DealState:
    state["decision"] = "review" if state["risk_score"] > 50 else "approve"
    return state

graph = StateGraph(DealState)
graph.add_node("enrich_deal", enrich_deal)
graph.add_node("compliance_check", compliance_check)

graph.set_entry_point("enrich_deal")
graph.add_edge("enrich_deal", "compliance_check")
graph.add_edge("compliance_check", END)

app = graph.compile()

This is the core pattern you want in banking: explicit state, deterministic transitions, and no hidden control flow.

2. •Wrap the graph in a service that Kubernetes can run

Kubernetes should not run your graph directly. Put it behind a small Python API so it can be scaled and probed like any other service.

from fastapi import FastAPI
from pydantic import BaseModel

app_api = FastAPI()

class DealRequest(BaseModel):
    request_id: str
    counterparty: str

@app_api.post("/analyze")
def analyze(req: DealRequest):
    initial_state = {
        "request_id": req.request_id,
        "counterparty": req.counterparty,
        "risk_score": 0,
        "decision": "review",
    }
    result = app.invoke(initial_state)
    return result

In production, this service becomes your agent runtime. Kubernetes handles replicas; LangGraph handles workflow logic.

3. •Package the agent as a container image

Use a minimal image and keep dependencies pinned. For regulated environments, reproducibility matters more than convenience.

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "agent_service:app_api", "--host", "0.0.0.0", "--port", "8000"]

Example requirements.txt:

fastapi==0.115.0
uvicorn==0.30.6
langgraph==0.2.35
langchain-core==0.3.13
pydantic==2.9.2

Build and push:

docker build -t registry.example.com/banking-agent:1.0.0 .
docker push registry.example.com/banking-agent:1.0.0

4. •Deploy the agent to Kubernetes

Use a Deployment for replicas and a Service for internal routing between banking microservices.

from kubernetes import client, config

config.load_kube_config()

apps_v1 = client.AppsV1Api()
core_v1 = client.CoreV1Api()

deployment = client.V1Deployment(
    metadata=client.V1ObjectMeta(name="banking-agent"),
    spec=client.V1DeploymentSpec(
        replicas=2,
        selector=client.V1LabelSelector(match_labels={"app": "banking-agent"}),
        template=client.V1PodTemplateSpec(
            metadata=client.V1ObjectMeta(labels={"app": "banking-agent"}),
            spec=client.V1PodSpec(
                containers=[
                    client.V1Container(
                        name="agent",
                        image="registry.example.com/banking-agent:1.0.0",
                        ports=[client.V1ContainerPort(container_port=8000)],
                    )
                ]
            ),
        ),
    ),
)

apps_v1.create_namespaced_deployment(namespace="banking-ai", body=deployment)

Create the service:

service = client.V1Service(
    metadata=client.V1ObjectMeta(name="banking-agent"),
    spec=client.V1ServiceSpec(
        selector={"app": "banking-agent"},
        ports=[client.V1ServicePort(port=80, target_port=8000)],
        type="ClusterIP",
    ),
)

core_v1.create_namespaced_service(namespace="banking-ai", body=service)

5. •Connect runtime config through ConfigMaps and Secrets

Don’t bake model endpoints or credentials into the image.

secret = client.V1Secret(
    metadata=client.V1ObjectMeta(name="llm-secret"),
    string_data={
        "OPENAI_API_KEY": "replace-me",
    },
)

config_map = client.V1ConfigMap(
    metadata=client.V1ObjectMeta(name="agent-config"),
    data={
        "RISK_THRESHOLD": "50",
        "ENVIRONMENT": "prod",
    },
)

core_v1.create_namespaced_secret(namespace="banking-ai", body=secret)
core_v1.create_namespaced_config_map(namespace="banking-ai", body=config_map)

Mount these into your pod spec later if you need environment-specific routing or model selection.

Testing the Integration

Hit the running service from inside the cluster or through port-forwarding.

import requests

payload = {
    "request_id": "deal-1042",
    "counterparty": "high_risk_client",
}

response = requests.post("http://localhost:8000/analyze", json=payload)
print(response.status_code)
print(response.json())

Expected output:

200
{
  "request_id": "deal-1042",
  "counterparty": "high_risk_client",
  "risk_score": 72,
  "decision": "review"
}

If you want to verify Kubernetes objects too:

kubectl get deploy,pod,svc -n banking-ai

You should see:

• •banking-agent Deployment available
• •at least one Running Pod per replica
• •banking-agent Service exposed inside the namespace

Real-World Use Cases

• •Deal intake triage
  • •Route inbound M&A or financing requests through LangGraph nodes for enrichment, credit exposure checks, and approval escalation.
• •Compliance review agents
  • •Run policy-check workflows that inspect counterparties, sanctions lists, and internal limits before a banker sees the case.
• •Analyst copilot services
  • •Expose an internal API that drafts investment memos, summarizes filings, and hands off only when confidence thresholds are met.

The clean pattern is simple: LangGraph owns reasoning flow; Kubernetes owns execution boundaries. If you keep those responsibilities separate, your banking agent system stays debuggable under load and predictable under audit.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.