How to Build a policy Q&A Agent Using LangGraph in Python for wealth management

A policy Q&A agent for wealth management answers questions like “Can this client hold this product in this jurisdiction?” or “What’s the approved process for a suitability exception?” It matters because advisors and operations teams need fast, consistent answers grounded in policy, not memory, and every answer needs to be traceable for compliance review.

Architecture

• •

  Policy retrieval layer

  • •Pulls from approved sources: internal policy docs, product restrictions, suitability rules, KYC/AML procedures, and jurisdiction-specific addenda.
  • •Use embeddings plus keyword filters so the agent can find exact clauses, not just semantically similar text.
• •

  LangGraph orchestration

  • •A StateGraph coordinates retrieval, answer drafting, compliance checks, and fallback handling.
  • •This keeps the flow deterministic enough for regulated use cases.
• •

  Policy-aware answer generator

  • •Uses an LLM only after retrieval.
  • •The model must cite source snippets and avoid inventing policy language.
• •

  Compliance guardrail node

  • •Validates the draft against rules like “no advice,” “no ungrounded claims,” and “escalate if policy is ambiguous.”
  • •This is where you enforce wealth management controls.
• •

  Audit logging layer

  • •Stores user question, retrieved documents, final response, timestamps, model version, and decision path.
  • •Required for reviewability and incident analysis.
• •

  Escalation path

  • •Routes uncertain cases to a human compliance or advisory operations queue.
  • •Necessary when policy conflicts or jurisdictional exceptions appear.

Implementation

1. Define the graph state and load your policy corpus

Use a typed state so each node knows what it can read and write. In production, your retriever would point to a vector store backed by approved documents in-region if data residency applies.

from typing import TypedDict, List
from langgraph.graph import StateGraph, END
from langchain_core.documents import Document

class PolicyState(TypedDict):
    question: str
    retrieved_docs: List[Document]
    draft_answer: str
    final_answer: str
    needs_escalation: bool

def retrieve_policies(state: PolicyState) -> dict:
    # Replace with your vector store retriever
    docs = [
        Document(
            page_content="Advisors must not recommend products outside approved jurisdiction lists.",
            metadata={"source": "policy_001", "jurisdiction": "SG"}
        ),
        Document(
            page_content="If suitability is unclear, escalate to compliance before responding.",
            metadata={"source": "policy_014", "jurisdiction": "SG"}
        ),
    ]
    return {"retrieved_docs": docs}

2. Draft an answer from retrieved policy text

Keep generation constrained. The model should summarize what the policy says and cite sources from retrieval. If you use an LLM here, keep temperature low and force a refusal when evidence is weak.

def draft_policy_answer(state: PolicyState) -> dict:
    docs = state["retrieved_docs"]
    context = "\n".join(
        f"- {d.page_content} (source={d.metadata['source']})" for d in docs
    )

    if not docs:
        return {
            "draft_answer": "I couldn't find an approved policy source for this question.",
            "needs_escalation": True,
        }

    answer = (
        f"Based on current policy sources:\n{context}\n\n"
        f"For the question '{state['question']}', the safe interpretation is that "
        f"advisors should stay within approved jurisdiction rules and escalate unclear suitability cases."
    )
    return {"draft_answer": answer}

3. Add a compliance check node and route on risk

This is where LangGraph earns its keep. You can branch based on whether the draft contains unsupported claims or whether the retrieved set is insufficient.

def compliance_check(state: PolicyState) -> dict:
    draft = state["draft_answer"].lower()
    docs_text = " ".join(d.page_content.lower() for d in state["retrieved_docs"])

    needs_escalation = (
        "must" in draft and "approved" not in docs_text
        or "unclear" in draft
        or len(state["retrieved_docs"]) == 0
    )

    return {"needs_escalation": needs_escalation}

def finalize_or_escalate(state: PolicyState) -> dict:
    if state["needs_escalation"]:
        return {
            "final_answer": (
                "I need to escalate this to compliance because the policy basis is incomplete "
                "or the request may involve a restricted judgment call."
            )
        }

    return {"final_answer": state["draft_answer"]}

4. Wire the graph with StateGraph, add_node, add_edge, and conditional routing

This pattern gives you a predictable execution path with explicit escalation behavior. That matters when audit teams ask how an answer was produced.

def route_after_compliance(state: PolicyState) -> str:
    return "escalate" if state["needs_escalation"] else "answer"

graph = StateGraph(PolicyState)

graph.add_node("retrieve_policies", retrieve_policies)
graph.add_node("draft_policy_answer", draft_policy_answer)
graph.add_node("compliance_check", compliance_check)
graph.add_node("finalize_or_escalate", finalize_or_escalate)

graph.set_entry_point("retrieve_policies")
graph.add_edge("retrieve_policies", "draft_policy_answer")
graph.add_edge("draft_policy_answer", "compliance_check")

graph.add_conditional_edges(
    "compliance_check",
    route_after_compliance,
    {
        "answer": "finalize_or_escalate",
        "escalate": "finalize_or_escalate",
    },
)

graph.add_edge("finalize_or_escalate", END)

app = graph.compile()

result = app.invoke({"question": "Can we recommend this product to a Singapore resident?"})
print(result["final_answer"])

Production Considerations

• •

  Data residency

  • •Keep policy indexes, logs, and embeddings in the required region.
  • •For cross-border wealth platforms, separate corpora by jurisdiction instead of mixing them into one global index.
• •

  Auditability

  • •Persist every run with question text, retrieved document IDs, final response, model version, prompt template hash, and graph path taken.
  • •When compliance reviews an answer six months later, they need replayable evidence.
• •

  Guardrails

  • •Block advice-like language unless it’s directly supported by policy.
  • •Force escalation when the question touches suitability exceptions, tax treatment, restricted products, or ambiguous jurisdictional rules.
• •

  Monitoring

  • •Track retrieval hit rate, escalation rate, refusal rate, and false-positive escalations.
  • •A rising escalation rate often means your corpus is stale or your chunking strategy is bad.

Common Pitfalls

• •

  Treating retrieval as optional

  • •Mistake: letting the LLM answer from general knowledge when no good document is found.
  • •Fix: hard-fail to escalation when retrieval confidence is below threshold or no approved source is returned.
• •

  Mixing policies across jurisdictions

  • •Mistake: indexing US, EU, APAC rules into one undifferentiated store.
  • •Fix: filter by client domicile, advisor location, entity type, and product jurisdiction before generation.
• •

  Skipping human escalation paths

  • •Mistake: assuming the agent can resolve all edge cases automatically.
  • •Fix: build explicit handoff states for restricted products, suitability exceptions, complaints handling, and regulatory ambiguity.
• •

  No source-level traceability

  • •Mistake: returning polished answers without citations or source IDs.
  • •Fix: include source metadata in every response payload so operations can verify which clause drove the answer.

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.