How to Integrate LangGraph for pension funds with Kubernetes for startups

Combining LangGraph for pension funds with Kubernetes gives you a clean way to run regulated, stateful AI workflows on infrastructure that can scale, recover, and isolate workloads properly. For startups building pension operations tooling, that means you can orchestrate multi-step agent flows for member queries, contribution checks, document review, and compliance routing without turning your app into a pile of ad hoc scripts.

Prerequisites

• •Python 3.10+
• •A Kubernetes cluster:
  • •local: kind, minikube, or docker desktop
  • •cloud: EKS, GKE, or AKS
• •kubectl configured and pointing at your cluster
• •Access to your pension fund data services or sandbox APIs
• •LangGraph installed:
  • •pip install langgraph langchain-openai
• •Kubernetes Python client installed:
  • •pip install kubernetes
• •Environment variables set:
  • •OPENAI_API_KEY
  • •KUBECONFIG or in-cluster service account access

Integration Steps

1. •Build the LangGraph workflow for pension fund operations

Start with a graph that handles a simple pension workflow: classify the request, fetch data, then produce a response. In production, each node should map to a real business capability like benefits lookup, contribution validation, or escalation.

from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

class PensionState(TypedDict):
    query: str
    intent: str
    result: str

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def classify_intent(state: PensionState):
    prompt = f"Classify this pension request: {state['query']}"
    intent = llm.invoke(prompt).content.strip()
    return {"intent": intent}

def answer_request(state: PensionState):
    prompt = f"Answer as a pension ops assistant. Intent={state['intent']}. Query={state['query']}"
    result = llm.invoke(prompt).content.strip()
    return {"result": result}

graph = StateGraph(PensionState)
graph.add_node("classify_intent", classify_intent)
graph.add_node("answer_request", answer_request)
graph.set_entry_point("classify_intent")
graph.add_edge("classify_intent", "answer_request")
graph.add_edge("answer_request", END)

app = graph.compile()

2. •Add Kubernetes-aware execution context

Your agent needs to know where it runs so it can call internal services safely. Use the Kubernetes client to read namespace metadata and cluster state before the graph executes.

from kubernetes import client, config

def load_k8s_context():
    try:
        config.load_incluster_config()
    except Exception:
        config.load_kube_config()

    v1 = client.CoreV1Api()
    namespace = open("/var/run/secrets/kubernetes.io/serviceaccount/namespace").read().strip() \
        if os.path.exists("/var/run/secrets/kubernetes.io/serviceaccount/namespace") else "default"

    pods = v1.list_namespaced_pod(namespace=namespace)
    return {
        "namespace": namespace,
        "pod_count": len(pods.items),
        "cluster_ready": True,
    }

3. •Wire Kubernetes context into the LangGraph input

Pass cluster metadata into your graph state so nodes can make decisions based on deployment context. This is useful for routing high-risk pension requests to dedicated workers or throttling expensive tasks when the cluster is under pressure.

import os

def build_input(query: str):
    k8s_context = load_k8s_context()
    return {
        "query": query,
        "intent": "",
        "result": "",
        **k8s_context,
    }

state = build_input("How much can I withdraw from my pension this year?")
output = app.invoke(state)
print(output["result"])

4. •Expose the workflow as a service inside Kubernetes

Package the graph behind an HTTP API and deploy it as a pod. The startup pattern is straightforward: one container for the API, one deployment for horizontal scaling, and one service for internal access.

from fastapi import FastAPI
from pydantic import BaseModel

api = FastAPI()

class QueryRequest(BaseModel):
    query: str

@api.post("/pension/query")
def pension_query(req: QueryRequest):
    state = build_input(req.query)
    result = app.invoke(state)
    return {
        "intent": result["intent"],
        "result": result["result"],
        "namespace": state["namespace"],
        "pod_count": state["pod_count"],
    }

5. •Deploy with Kubernetes primitives

Use a Deployment for replicas and a Service for stable access. Keep secrets out of code and inject them through environment variables or mounted secrets.

from kubernetes import client, config

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

deployment = client.V1Deployment(
    metadata=client.V1ObjectMeta(name="pension-agent"),
    spec=client.V1DeploymentSpec(
        replicas=2,
        selector=client.V1LabelSelector(match_labels={"app": "pension-agent"}),
        template=client.V1PodTemplateSpec(
            metadata=client.V1ObjectMeta(labels={"app": "pension-agent"}),
            spec=client.V1PodSpec(containers=[
                client.V1Container(
                    name="api",
                    image="your-registry/pension-agent:latest",
                    ports=[client.V1ContainerPort(container_port=8000)],
                    env=[
                        client.V1EnvVar(name="OPENAI_API_KEY", value_from=client.V1EnvVarSource(
                            secret_key_ref=client.V1SecretKeySelector(name="openai-secret", key="api_key")
                        ))
                    ]
                )
            ])
        )
    )
)

apps_v1.create_namespaced_deployment(namespace="default", body=deployment)

Testing the Integration

Run the API locally or port-forward it from Kubernetes, then hit the endpoint with a real pension-style query.

import requests

response = requests.post(
    "http://localhost:8000/pension/query",
    json={"query": "Can I transfer my preserved benefits to another scheme?"}
)

print(response.status_code)
print(response.json())

Expected output:

200
{
  "intent": "...",
  "result": "...",
  "namespace": "default",
  "pod_count": 2
}

If you want a stronger test in-cluster, verify both sides separately:

• •LangGraph returns a non-empty result
• •Kubernetes returns valid namespace/pod metadata
• •The service stays responsive after pod restarts

Real-World Use Cases

• •

  Member support automation

  • •Answer pension contribution questions.
  • •Route complex cases to human ops teams.
  • •Keep every step auditable through graph state.
• •

  Compliance triage

  • •Classify requests that may trigger regulatory review.
  • •Run policy checks in separate nodes.
  • •Scale review workers independently in Kubernetes.
• •

  Document-driven workflows

  • •Extract data from benefit statements or transfer forms.
  • •Validate against internal systems.
  • •Push exceptions into dedicated queues or pods for manual handling.

The main pattern here is simple: keep LangGraph focused on orchestration and decision flow, and let Kubernetes handle runtime isolation, scaling, and rollout control. That split is what makes this stack usable in production instead of just working in a notebook.

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.