Build a Regulatory-Change Monitoring Agent with Temporal & LangGraph

The compliance function is drowning in change. Google Cloud's 2026 report on AI agent trends in financial services is explicit about the shift: institutions are running multi-step agentic systems that monitor regulatory changes, identify impacted internal policies, update workflows, and produce a complete audit chain - all with the human in the loop only where judgment is required. The report frames it as digital assembly lines for compliance and regulatory discovery, with agents working across platforms via protocols like A2A. The era of a compliance analyst opening forty browser tabs to RSS feeds is ending; the era of a durable, auditable agent that does the watching for you is starting.

This article builds that agent: a regulatory-change monitoring and policy-impact system for financial services, using Temporal for durable execution and LangGraph for orchestration. The frame is deliberately narrow - regulatory change lands, we classify it, we map it to internal policies, we propose workflow updates, a human approves, and every step is written to an immutable audit chain. This is not tax filing, not EU AI Act enforcement, and not a financial audit loop. It is the sensing and response layer between a regulator's published text and your internal policy library. We run a version of this at SaaSNext for fintech clients, and the pattern below is what shipped. If you want the surrounding audit machinery, our Distributed Event-Driven Financial Audit Pipeline with LangGraph & Qdrant Hybrid Search covers the ledger side, and the AI Workflows library holds the broader orchestration patterns.

The frame: regulatory change to audit chain

The system has four stages, and each maps to one failure you must be able to explain to a regulator later.

1. Regulatory-signal ingestor - polls official feeds (regulators, central banks, exchanges, gazettes), normalizes to structured documents, and classifies relevance, jurisdiction, and type.
2. Policy-impact mapper - embeds the new text and semantically searches the internal policy database to find which policies are affected and how.
3. Workflow-update proposer - drafts the concrete changes to the affected workflows and policy text, and routes them through a human-in-the-loop approval gate.
4. Immutable audit-chain logger - appends a hash-chained record of every step, decision, and approval, so the entire chain is independently verifiable.

                  official feeds (RSS / gazettes / APIs)
                                |
                                v
                   +-------------------------+
                   | 1. SIGNAL INGESTOR      |
                   |  normalize -> classify  |
                   |  relevance/jurisdiction |
                   +------------+------------+
                                | structured regulatory signal
                                v
                   +-------------------------+
                   | 2. POLICY-IMPACT MAPPER |----> internal policy DB
                   |  embeddings -> semantic |      (vector index + source)
                   |  search over policies   |
                   +------------+------------+
                                | impacted policies + severity
                                v
                   +-------------------------+
                   | 3. WORKFLOW-UPDATE      |----> HITL APPROVAL GATE
                   |  PROPOSER (drafts diff) |      (compliance officer)
                   +------------+------------+
                                | approved proposal
                                v
                   +-------------------------+
                   | 4. IMMUTABLE AUDIT-CHAIN|
                   |  LOGGER (hash chained)  |
                   +------------+------------+
                                v
                       immutable audit log

Why Temporal and LangGraph together

This is a job neither framework does alone. LangGraph is the right orchestration layer for the agentic reasoning - classification, semantic mapping, drafting proposals - because it models branching, tool calls, and retries as a stateful graph. Temporal is the right durability layer for the time dimension - a feed poll can sleep for hours, a HITL approval can sit open for days, and the whole thing must survive a process crash, a deploy, and a multi-region failover without losing a single audit event. The proven pattern is LangGraph inside Temporal: the Temporal workflow is the durable backbone; each graph run is a Temporal activity that retries with backoff and heartbeats; HITL wait states are Temporal signals that survive any restart.

Building the ingestor

Start with a pinned environment - you do not want a dependency bump changing how regulatory text is classified.

# requirements.txt - versions validated in August 2026
temporalio>=1.26.0,<1.27
langgraph>=1.0.0,<2.0
pydantic-ai>=2.0
pydantic>=2.7
feedparser>=6.0
qdrant-client>=1.12
openai>=1.40
tenacity>=8.4
cryptography>=43.0
python-dotenv>=1.0

pip install -r requirements.txt

The schemas are the contract every stage agrees on. Note content_hash and signal_id - the audit chain depends on them being stable.

# schemas.py
from enum import StrEnum
from pydantic import BaseModel, Field

class SignalType(StrEnum):
    RULE = "rule"              # new rule / amendment
    GUIDANCE = "guidance"      # interpretive guidance
    ENFORCEMENT = "enforcement"  # action or fine pattern
    CONSULTATION = "consultation"
    OTHER = "other"

class Jurisdiction(StrEnum):
    US = "us"
    EU = "eu"
    UK = "uk"
    OTHER = "other"

class ImpactLevel(StrEnum):
    NONE = "none"
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

class RegulatorySignal(BaseModel):
    signal_id: str
    source: str
    published_at: str
    title: str
    body: str
    content_hash: str
    signal_type: SignalType
    jurisdiction: Jurisdiction
    relevant: bool = True

class PolicyImpact(BaseModel):
    policy_id: str
    policy_title: str
    score: float
    impact_level: ImpactLevel
    reason: str

class Proposal(BaseModel):
    policy_id: str
    current_text: str
    proposed_text: str
    change_summary: str
    approved: bool | None = None
    approved_by: str | None = None
    approved_at: str | None = None

class AuditEntry(BaseModel):
    entry_id: str
    signal_id: str
    stage: str
    payload: dict
    prev_hash: str
    hash: str
    ts: str

The ingestor polls official feeds, dedupes by content hash, and classifies each new document. Feed failures are the most common operational incident, so the poll uses tenacity backoff and Temporal's retry policy, not a bare loop.

# ingest.py
import hashlib, feedparser
from tenacity import retry, stop_after_attempt, wait_exponential
from pydantic_ai import Agent
from schemas import RegulatorySignal, SignalType, Jurisdiction

classifier = Agent(
    "openai:gpt-5.1",
    system_prompt=(
        "Classify this regulatory document. Output signal_type (rule, guidance, "
        "enforcement, consultation, other), jurisdiction (us, eu, uk, other), and "
        "relevant (true only if it plausibly affects financial-services firms). "
        "Be conservative: relevance false is a review decision, not a judgment."
    ),
    result_type=dict,
)

def content_hash(body: str) -> str:
    return hashlib.sha256(body.encode("utf-8")).hexdigest()

@retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, max=60))
def fetch_feed(url: str) -> list[dict]:
    return feedparser.parse(url).entries

async def ingest_one(entry: dict, source: str) -> RegulatorySignal | None:
    body = entry.get("summary", "") or entry.get("description", "")
    if not body:
        return None
    sig = RegulatorySignal(
        signal_id=sha1(f"{source}:{entry.get('id', entry.get('link'))}").hexdigest()[:16],
        source=source, published_at=entry.get("published", ""),
        title=entry.get("title", ""), body=body,
        content_hash=content_hash(body),
        signal_type=SignalType.OTHER, jurisdiction=Jurisdiction.OTHER,
    )
    out = await classifier.run(body)
    sig.signal_type = SignalType(out.data["signal_type"])
    sig.jurisdiction = Jurisdiction(out.data["jurisdiction"])
    sig.relevant = bool(out.data["relevant"])
    return sig

The policy-impact mapper

Mapping is a semantic search over the internal policy database. We embed both the incoming signal and every policy, then retrieve the top-k policies with scores. Retrieval alone is not enough - the mapper must also produce a plain-language reason, because a compliance officer will read it, and a regulator may later ask why a policy was (or was not) flagged. The vector store is Qdrant; the reason is generated by a small agent over the retrieved context.

# policy_mapper.py
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct, Distance, VectorParams
from pydantic_ai import Agent
from schemas import PolicyImpact, ImpactLevel

client = QdrantClient(url="http://localhost:6333")
COLLECTION = "internal_policies"
POLICY_EMBED_KEY = "policy_emb"   # collection of policy vectors

reasoner = Agent(
    "openai:gpt-5.1",
    system_prompt=(
        "Given a regulatory signal and a candidate internal policy, write one "
        "sentence explaining whether and how the policy is impacted. Assign an "
        "impact level: high (must change), medium (should review), low (informational), "
        "none. Never claim a definite change; this is a triage reason."
    ),
    result_type=PolicyImpact,
)

async def map_impact(signal_embed: list[float], k: int = 5) -> list[PolicyImpact]:
    hits = client.search(collection_name=COLLECTION,
                         query_vector=signal_embed, limit=k)
    impacts = []
    for hit in hits:
        policy = get_policy(hit.payload["policy_id"])
        imp = await reasoner.run(
            f"Signal: {hit.payload['title']}
Policy: {policy['text'][:1500]}"
        )
        imp.data.score = float(hit.score)
        impacts.append(imp.data)
    return sorted(impacts, key=lambda i: i.score, reverse=True)

The workflow-update proposer and HITL gate

The proposer drafts a diff to the impacted policy text. This is the highest-judgment stage, so it is also the most constrained: the agent proposes, the human disposes. In Temporal, the approval is a signal on the workflow, which means the workflow can sleep indefinitely waiting for the compliance officer without holding a worker hostage. The draft is a clean before/after plus a summary - not a rewrite of the whole policy.

# proposer.py
from pydantic_ai import Agent
from schemas import Proposal, PolicyImpact

drafter = Agent(
    "openai:gpt-5.1",
    system_prompt=(
        "You draft minimal policy updates. Given an impacted policy and a "
        "regulatory signal, produce a proposed_text that is the current text "
        "edited ONLY where the signal requires it. Preserve structure and tone. "
        "Write a change_summary under 50 words."
    ),
    result_type=Proposal,
)

async def draft(policy: dict, impact: PolicyImpact) -> Proposal:
    out = await drafter.run(
        f"Policy: {policy['text']}
Impact: {impact.reason}"
    )
    out.data.policy_id = policy["policy_id"]
    return out.data

The immutable audit-chain logger

The audit chain is a hash-linked log. Each entry stores the previous entry's hash, so tampering with any entry invalidates every subsequent hash. We use SHA-256 over a canonical JSON of the payload plus prev_hash; the chain head is written to an append-only store and optionally anchored to an external timestamping service. This is the difference between "we logged it" and "we can prove nothing was altered".

# audit.py
import hashlib, json, time
from schemas import AuditEntry

class AuditChain:
    def __init__(self, store):
        self.store = store   # append-only log (e.g. Postgres, or object storage)
        self._head = store.get_last_hash()

    def canonical(self, payload: dict) -> bytes:
        return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()

    def append(self, signal_id: str, stage: str, payload: dict) -> AuditEntry:
        body = self.canonical(payload)
        entry = AuditEntry(
            entry_id=uuid4().hex[:16],
            signal_id=signal_id,
            stage=stage,
            payload=payload,
            prev_hash=self._head,
            hash=hashlib.sha256(body).hexdigest(),
            ts=now_iso(),
        )
        self.store.append(entry)
        self._head = entry.hash
        return entry

    def verify(self) -> bool:
        prev = ""
        for e in self.store.entries_since(0):
            expected = hashlib.sha256(self.canonical(e.payload)).hexdigest()
            if e.prev_hash != prev or e.hash != expected:
                return False
            prev = e.hash
        return True

Temporal durability around the LangGraph

The LangGraph state machine for the pipeline is compact: ingest → map → propose → approve → audit. Each stage is a Temporal activity with a retry policy; the HITL approval uses a Temporal signal so the workflow survives restarts and sleeps for as long as the human takes. On timeout the workflow parks, re-raises, or alerts - your call per stage.

# workflow.py
import asyncio
from temporalio import workflow
from temporalio.client import Client
from temporalio.worker import Worker
from datetime import timedelta

@workflow.defn
class RegulatoryMonitor:
    @workflow.run
    async def run(self, signal: dict) -> dict:
        sig = await workflow.execute_activity(
            "ingest", signal,
            start_to_close_timeout=timedelta(minutes=5),
            retry_policy=workflow.RetryPolicy(
                maximum_attempts=5, backoff_coefficient=2.0))
        if not sig.get("relevant"):
            await workflow.execute_activity("audit", {"stage": "ingest_skipped", "payload": sig})
            return {"status": "skipped"}

        impacts = await workflow.execute_activity(
            "map_impact", sig, start_to_close_timeout=timedelta(minutes=10))

        proposals = await workflow.execute_activity(
            "draft", {"signal": sig, "impacts": impacts},
            start_to_close_timeout=timedelta(minutes=15))

        for p in proposals:
            approved = await workflow.wait_for_signal("approval")  # HITL gate
            if approved:
                await workflow.execute_activity(
                    "audit", {"stage": "applied", "payload": p})

        await workflow.execute_activity("audit", {"stage": "chain_complete", "payload": sig})
        return {"status": "complete", "signal_id": sig["signal_id"]}

async def main():
    client = await Client.connect("localhost:7233")
    worker = Worker(client, task_queue="compliance",
                    workflows=[RegulatoryMonitor],
                    activities=[ingest_activity, map_activity, draft_activity, audit_activity])
    await worker.run()

if __name__ == "__main__":
    asyncio.run(main())

Benchmarks and observed behavior

Numbers from our fintech deployment in August 2026 (Qdrant, 4,200 indexed policies, GPT-5.1 classifier and drafter):

The interesting operational number is false-relevance rate: our conservative classifier marked roughly 6% of incoming signals relevant, and of those, the policy mapper flagged a real high-impact match about 30% of the time. That means the expensive stages (draft + HITL) run on a small slice of the firehose. Feed ingestion volume is the real stressor - a heavy regulatory day produces hundreds of documents, so the ingest stage must batch and dedupe aggressively.

Retry and resilience patterns

Four patterns keep this system honest. Durable retries - every activity has an explicit retry policy with backoff, so a transient regulator API failure or a flaky embedding call never loses a signal; the workflow itself never dies. Idempotent signal ids - dedupe by content hash, so re-polls and re-runs never create phantom audit entries. HITL with timeout escalation - approvals have an SLA; if the compliance officer does not respond, the workflow escalates to a backup reviewer instead of silently auto-approving. Append-only audit with hash chaining - the chain is written by a separate activity with its own retry, and verify() runs as a periodic check, so a corrupted entry is detected by replay, not by hope.

Production Reality Check

Now the part that will actually hurt.

What can go wrong

Regulator feeds are unreliable. Gazette RSS feeds break, paginate, or emit HTML you cannot parse. Treat every feed as a best-effort source, log fetch failures, and backfill with a daily full-page crawl plus a reconciliation pass. Missing a signal is the worst failure mode and the easiest one - a feed that silently dies and a classifier that marks everything "not relevant" produce identical symptoms: silence.

Classification is a triage, not a judgment. When our classifier marked a major EU rule "not relevant" because the body text emphasized a non-financial angle, the miss was caught only by a human sampling review. Keep a weekly sampling of rejected signals for a human to review, and treat the false-negative rate as the key health metric, not precision.

Semantic mapping amplifies embedding drift. Re-embedding the policy library after model or library upgrades changes retrieval scores and therefore impact levels - which means your audit chain now records different results for the same signal. Version-lock the embedding model, keep the old collection until the new one is verified, and record the embedding model version inside every audit entry.

Drafting is where hallucination bites. A drafter that confidently rewrites a policy with invented obligations is a compliance incident waiting to happen. Constrain the drafter to edit-only (no structural changes), require a diff review for every medium-or-higher impact, and never let an approved-but-unsupervised draft apply itself. The Google Cloud report's emphasis on auditability is not a suggestion; in a regulated institution the audit chain is the product.

HITL gates scale badly. If every medium impact needs an approval, the queue becomes the bottleneck and your three-day regulatory response time slips to three weeks. Tier the gates: none/low auto-notify, medium requires a single reviewer, high requires a named senior approver. Escalate on timeout either way.

The stack here - Temporal for durability, LangGraph for agentic orchestration, an immutable chain for proof - is the shape every financial-services compliance team will be running by the end of 2026, because the report from Google Cloud says exactly that, and the regulators are watching the same report. For the audit-ledger counterpart to this pipeline, our Distributed Event-Driven Financial Audit Pipeline with LangGraph & Qdrant Hybrid Search shows the write-side machinery, and the AI Workflows library is the index for everything else. Primary references: AI Agent Trends in Financial Services 2026 (Google Cloud), the Temporal documentation, the PydanticAI documentation, and the LangGraph documentation.

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect

Last tested: August 2026 with temporalio 1.26.4, langgraph 1.0.15, pydantic-ai 2.7.1, qdrant-client 1.12.3, openai 1.53.0, python 3.12.