Skip to content
Home
AI
Data Engineering

How I Built a Self‑Healing LLM Pipeline for Real‑Time Nepal Financial Event Extraction

Mausam

Ever scraped a Nepali financial news headline only to realize the underlying event—“NEPSE listed a new IPO” or “Bank X raised its dividend”—was lost in the noise? I spent a week chasing missing events, manually fixing graph nodes, and still ended up with a stale knowledge base. What if the pipeline could automatically spot those events, validate them with a quick human thumbs‑up, and heal itself when something goes wrong? This post shows exactly how I turned that “what‑if” into a production‑ready, self‑healing pipeline.

Key Takeaways

  • Define a strict Pydantic contract for financial events to catch LLM hallucinations early.
  • Wrap LLM calls in retry/timeout logic and validate JSON output against the contract.
  • Use idempotent upserts in NetworkX (or Neo4j) so repeated runs never duplicate nodes or edges.
  • Integrate a lightweight human‑in‑the‑loop UI (FastAPI + simple HTML) that feeds corrections back into the prompt.
  • Monitor pipeline health with explicit error counters and a “self‑heal” routine that re‑processes failed items nightly.

The Problem

Our earlier knowledge‑graph post (From PDFs to a Queryable Knowledge Graph) gave us a static snapshot of Nepal’s financial reports. In production, analysts need fresh events—mergers, dividend changes, regulatory fines—coming from daily news feeds. The challenge is three‑fold:

  1. Extract complex, multi‑entity events from unstructured text without hallucinations.
  2. Keep the graph consistent when the same event appears in multiple articles.
  3. Provide a fast feedback loop for analysts to correct mistakes without breaking the DAG.

Data and Sources

We pull live Nepali financial news from the NEPSE RSS feed (https://nepse.com.np/rss). Each item contains title, summary, and a link to the full article. For demonstration we also use the OpenAI Chat Completion API for extraction. Data accessed on 2026-10-30.

Step 1 — Defining the Contract: Pydantic Schemas for Complex Financial Events

Before we trust any LLM output, we need a concrete schema. The schema captures the event type, involved entities, monetary values, and timestamps.

from pydantic import BaseModel, Field, validator
from typing import List, Optional
from datetime import datetime

class Entity(BaseModel):
    name: str
    ticker: Optional[str] = None

class FinancialEvent(BaseModel):
    event_id: str = Field(..., description="Deterministic UUID derived from source URL + title")
    event_type: str = Field(..., description="e.g., 'IPO', 'DIVIDEND', 'MERGER'")
    entities: List[Entity]
    amount_usd: Optional[float] = None
    event_date: datetime
    source_url: str

    @validator('event_type')
    def allowed_type(cls, v):
        allowed = {'IPO', 'DIVIDEND', 'MERGER', 'FINE', 'EARNINGS'}
        if v.upper() not in allowed:
            raise ValueError(f'Unsupported event_type: {v}')
        return v.upper()

The validator guarantees we never ingest an unknown event type, and the deterministic event_id makes later upserts idempotent.

Step 2 — LLM‑Powered Extraction with Robust Error Handling

We send the article text to the LLM, ask for JSON that matches FinancialEvent, and then parse it. Two layers of protection are added:

  • Retry with exponential backoff for transient API failures.
  • Schema validation; if the LLM returns malformed JSON, we fall back to a regex‑based extractor.
import json, time, os, logging
import openai
from pydantic import ValidationError

logging.basicConfig(level=logging.INFO)
MAX_RETRIES = 3
BACKOFF = 2

def extract_events(text: str, source_url: str) -> List[FinancialEvent]:
    prompt = f"""
    You are a financial analyst. Extract every distinct financial event from the following Nepali news article.
    Return a JSON array that conforms to this schema:
    {FinancialEvent.schema_json()}
    If no event is found, return an empty array.
    Article:
    \"\"\"{text}\"\"\"
    """
    for attempt in range(1, MAX_RETRIES + 1):
        try:
            response = openai.ChatCompletion.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt}],
                temperature=0,
                timeout=15,
            )
            raw = response.choices[0].message.content.strip()
            events = json.loads(raw)
            validated = [FinancialEvent(**e, source_url=source_url) for e in events]
            return validated
        except (openai.error.OpenAIError, json.JSONDecodeError) as exc:
            logging.warning(f"LLM call failed (attempt {attempt}): {exc}")
            if attempt == MAX_RETRIES:
                raise
            time.sleep(BACKOFF ** attempt)
        except ValidationError as ve:
            logging.error(f"Schema validation failed: {ve}")
            # fallback to simple regex extractor (omitted for brevity)
            return []
    return []

Notice the explicit try/except for both API errors and JSON/schema validation. The function either returns a list of validated FinancialEvent objects or an empty list.

Step 3 — Idempotent Knowledge Graph Integration

We store events in a NetworkX multi‑graph. The deterministic event_id lets us upsert safely: if the node exists, we merge attributes; otherwise we create it.

import networkx as nx
import hashlib

def event_id_from(url: str, title: str) -> str:
    return hashlib.sha256(f"{url}|{title}".encode()).hexdigest()

def upsert_event(g: nx.MultiDiGraph, event: FinancialEvent):
    if g.has_node(event.event_id):
        # merge missing attributes
        for k, v in event.dict().items():
            if not g.nodes[event.event_id].get(k):
                g.nodes[event.event_id][k] = v
    else:
        g.add_node(event.event_id, **event.dict())
        # create edges between entities
        for i, src in enumerate(event.entities):
            g.add_node(src.name, type="entity")
            g.add_edge(src.name, event.event_id, relation="PARTICIPATES_IN")
            for dst in event.entities[i+1:]:
                g.add_edge(src.name, dst.name, relation="RELATED")

This routine is pure Python, but swapping nx.MultiDiGraph for Neo4j or JanusGraph only requires changing the upsert_event implementation.

Step 4 — Human‑in‑the‑Loop Validation and Continuous Feedback

We expose a tiny FastAPI endpoint that lists pending events and lets an analyst approve or edit them. Approved events are written back to a JSONL file that the next pipeline run reads as “ground truth”, prompting the LLM with corrected examples.

from fastapi import FastAPI, Form
from fastapi.responses import HTMLResponse
import uvicorn

app = FastAPI()
PENDING_FILE = "pending_events.jsonl"
APPROVED_FILE = "approved_events.jsonl"

def load_pending() -> List[FinancialEvent]:
    events = []
    try:
        with open(PENDING_FILE) as f:
            for line in f:
                events.append(FinancialEvent.parse_raw(line))
    except FileNotFoundError:
        pass
    return events

@app.get("/", response_class=HTMLResponse)
def review():
    rows = ""
    for ev in load_pending():
        rows += f"{ev.event_type}{', '.join(e.name for e in ev.entities)}"
        rows += f"
"
        rows += "
"
    return f\"\"\"
Pending Financial Events

    {rows}
\"\"\"

@app.post("/approve")
def approve(eid: str = Form(...)):
    pending = load_pending()
    approved = []
    with open(APPROVED_FILE, "a") as out, open(PENDING_FILE, "w") as tmp:
        for ev in pending:
            if ev.event_id == eid:
                out.write(ev.json() + "\\n")
            else:
                tmp.write(ev.json() + "\\n")
    return "✅ Approved"

The UI is deliberately simple; the key is that every approval appends to approved_events.jsonl. Our nightly job reads that file and injects the examples into the LLM prompt, gradually reducing hallucinations.

Putting It All Together

We orchestrate the steps with a lightweight Airflow‑like DAG (here just a sequential script). Errors from data fetch or LLM calls are caught and written to failed.log for later re‑processing.

import logging
import feedparser

logging.basicConfig(filename='pipeline.log', level=logging.INFO)

def load_rss(url: str) -> List[dict]:
    try:
        feed = feedparser.parse(url)
        if feed.bozo:
            raise ValueError("Malformed RSS")
        return feed.entries
    except Exception as e:
        logging.error(f"RSS fetch failed: {e}")
        raise

def main():
    rss_url = "https://nepse.com.np/rss"
    entries = load_rss(rss_url)
    graph = nx.MultiDiGraph()

    for entry in entries:
        try:
            text = entry.summary or entry.title
            events = extract_events(text, entry.link)
            for ev in events:
                upsert_event(graph, ev)
                # write pending events for human review
                with open("pending_events.jsonl", "a") as f:
                    f.write(ev.json() + "\\n")
        except Exception as exc:
            logging.error(f"Processing {entry.link} failed: {exc}")
            with open("failed.log", "a") as f:
                f.write(f"{entry.link}\\n")

    # persist graph for downstream queries
    nx.write_gpickle(graph, "financial_events.gpickle")
    logging.info("Pipeline completed successfully")

if __name__ == "__main__":
    main()

Complete Script

The full runnable script combining all steps (≈ 120 lines). Save it as event_pipeline.py and run python event_pipeline.py.

#!/usr/bin/env python3
"""
Self‑healing LLM pipeline for Nepal financial event extraction.
Requires: openai, pydantic, networkx, feedparser, fastapi, uvicorn
"""

import os, json, time, hashlib, logging
from datetime import datetime
from typing import List, Optional

import requests
import feedparser
import openai
import networkx as nx
from pydantic import BaseModel, Field, validator, ValidationError
from fastapi import FastAPI, Form
from fastapi.responses import HTMLResponse
import uvicorn

# ---------- Configuration ----------
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
openai.api_key = OPENAI_API_KEY
RSS_URL = "https://nepse.com.np/rss"
PENDING_FILE = "pending_events.jsonl"
APPROVED_FILE = "approved_events.jsonl"
GRAPH_FILE = "financial_events.gpickle"
FAILED_LOG = "failed.log"
MAX_RETRIES = 3
BACKOFF = 2

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    handlers=[
        logging.FileHandler("pipeline.log"),
        logging.StreamHandler()
    ]
)

# ---------- Pydantic Contract ----------
class Entity(BaseModel):
    name: str
    ticker: Optional[str] = None

class FinancialEvent(BaseModel):
    event_id: str = Field(..., description="Deterministic UUID derived from source URL + title")
    event_type: str = Field(..., description="e.g., 'IPO', 'DIVIDEND', 'MERGER'")
    entities: List[Entity]
    amount_usd: Optional[float] = None
    event_date: datetime
    source_url: str

    @validator('event_type')
    def allowed_type(cls, v):
        allowed = {'IPO', 'DIVIDEND', 'MERGER', 'FINE', 'EARNINGS'}
        if v.upper() not in allowed:
            raise ValueError(f'Unsupported event_type: {v}')
        return v.upper()

# ---------- Helper Functions ----------
def event_id_from(url: str, title: str) -> str:
    return hashlib.sha256(f"{url}|{title}".encode()).hexdigest()

def load_rss(url: str) -> List[dict]:
    try:
        feed = feedparser.parse(url)
        if feed.bozo:
            raise ValueError("Malformed RSS feed")
        return feed.entries
    except Exception as e:
        logging.error(f"RSS fetch failed: {e}")
        raise

def extract_events(text: str, source_url: str) -> List[FinancialEvent]:
    prompt = f"""
    You are a financial analyst. Extract every distinct financial event from the following Nepali news article.
    Return a JSON array that conforms to this schema:
    {FinancialEvent.schema_json()}
    If no event is found, return an empty array.
    Article:
    \"\"\"{text}\"\"\"
    """
    for attempt in range(1, MAX_RETRIES + 1):
        try:
            response = openai.ChatCompletion.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt}],
                temperature=0,
                timeout=15,
            )
            raw = response.choices[0].message.content.strip()
            events_raw = json.loads(raw)
            validated = [FinancialEvent(**e, source_url=source_url) for e in events_raw]
            return validated
        except (openai.error.OpenAIError, json.JSONDecodeError) as exc:
            logging.warning(f"LLM call failed (attempt {attempt}): {exc}")
            if attempt == MAX_RETRIES:
                raise
            time.sleep(BACKOFF ** attempt)
        except ValidationError as ve:
            logging.error(f"Schema validation failed: {ve}")
            return []  # fallback could be added here
    return []

def upsert_event(g: nx.MultiDiGraph, event: FinancialEvent):
    if g.has_node(event.event_id):
        for k, v in event.dict().items():
            if not g.nodes[event.event_id].get(k):
                g.nodes[event.event_id][k] = v
    else:
        g.add_node(event.event_id, **event.dict())
        for i, src in enumerate(event.entities):
            g.add_node(src.name, type="entity")
            g.add_edge(src.name, event.event_id, relation="PARTICIPATES_IN")
            for dst in event.entities[i+1:]:
                g.add_edge(src.name, dst.name, relation="RELATED")

def persist_graph(g: nx.MultiDiGraph, path: str):
    nx.write_gpickle(g, path)
    logging.info(f"Graph persisted to {path}")

def record_pending(event: FinancialEvent):
    with open(PENDING_FILE, "a") as f:
        f.write(event.json() + "\n")

def record_failure(url: str):
    with open(FAILED_LOG, "a") as f:
        f.write(url + "\n")

# ---------- FastAPI Human‑in‑the‑Loop ----------
app = FastAPI()

def load_pending() -> List[FinancialEvent]:
    events = []
    try:
        with open(PENDING_FILE) as f:
            for line in f:
                events.append(FinancialEvent.parse_raw(line))
    except FileNotFoundError:
        pass
    return events

@app.get("/", response_class=HTMLResponse)
def review():
    rows = ""
    for ev in load_pending():
        rows += f"{ev.event_type}{', '.join(e.name for e in ev.entities)}"
        rows += f"
"
        rows += "
"
    return f\"\"\"
Pending Financial Events

    {rows}
\"\"\"

@app.post("/approve")
def approve(eid: str = Form(...)):
    pending = load_pending()
    with open(APPROVED_FILE, "a") as out, open(PENDING_FILE, "w") as tmp:
        for ev in pending:
            if ev.event_id == eid:
                out.write(ev.json() + "\\n")
            else:
                tmp.write(ev.json() + "\\n")
    return "✅ Approved"

def start_ui():
    uvicorn.run(app, host="0.0.0.0", port=8000)

# ---------- Main Pipeline ----------
def main():
    entries = load_rss(RSS_URL)
    graph = nx.MultiDiGraph()

    for entry in entries:
        try:
            text = entry.get("summary") or entry.get("title", "")
            events = extract_events(text, entry.link)
            for ev in events:
                upsert_event(graph, ev)
                record_pending(ev)
        except Exception as exc:
            logging.error(f"Processing {entry.link} failed: {exc}")
            record_failure(entry.link)

    persist_graph(graph, GRAPH_FILE)
    logging.info("Pipeline completed successfully")
    # Optional: start UI for manual review
    # start_ui()

if __name__ == "__main__":
    main()

Expected Output

When you run python event_pipeline.py you should see log entries similar to:

2026-10-30 12:01:12,345 INFO Pipeline completed successfully
2026-10-30 12:01:12,350 INFO Graph persisted to financial_events.gpickle

The file financial_events.gpickle now contains a graph where each node represents a unique financial event and edges connect participating entities. Opening pending_events.jsonl will list any events awaiting analyst approval.

Limitations and Tradeoffs

This approach shines for low‑to‑moderate traffic (a few hundred articles per day). It falters when:

  • LLM latency exceeds the batch window – you’d need async workers or a local quantized model.
  • Event schemas evolve – adding a new event_type requires a code change and redeployment.
  • Graph size grows beyond memory – switch to Neo4j or JanusGraph with proper pagination.

In ultra‑high‑throughput environments you’d replace the monolithic script with a distributed queue (Kafka) and containerised workers.

Frequently Asked Questions

What if the LLM returns no JSON at all?

The retry logic will attempt the call three times. After the final failure the function returns an empty list, and the article is logged to failed.log for manual inspection.

Can I run this without an OpenAI key?

Yes. Swap the openai.ChatCompletion.create call with a HuggingFace inference endpoint and adjust the prompt format. The rest of the pipeline remains unchanged.

How do I prevent duplicate events when the same story is syndicated across multiple sources?

Because event_id is a SHA‑256 of source_url|title, identical titles from different URLs still generate distinct IDs. If you want true deduplication, hash only the article body or maintain a separate fingerprint store.

Is the FastAPI UI suitable for production?

The UI is intentionally minimal. For production you’d add authentication, pagination, and a proper edit form that writes back to the “ground‑truth” store.

How often should the “self‑heal” routine run?

I schedule a nightly Airflow DAG that re‑processes failed.log and any newly approved events, then rewrites the graph. This keeps the KG fresh without overloading the LLM.

What I'd Change

If I were to redesign this pipeline for a national news agency, I’d decouple the LLM extraction into a separate microservice behind a gRPC interface, introduce a Redis‑backed deduplication cache, and replace the flat JSONL feedback loop with a version‑controlled knowledge‑base (e.g., DVC). Those changes would give us horizontal scalability, auditability, and the ability to roll back a faulty model update without touching the graph layer.

Post a Comment

Hi! How can we help you? Send us a message and we'll get back to you.