Skip to content
Home
Finance
Python

Orchestrating LLM Consensus: Building a Multi‑Agent Pipeline to Synthesize Simulated Financial News

Mausam

Ever stared at a dozen headlines, analyst quotes, and blog posts and wondered how a single, trustworthy trading signal could emerge without you manually reconciling contradictions? I hit that wall last quarter while prototyping a news‑driven NEPSE strategy—my notebook was a mess of copy‑pasted snippets and half‑written sentiment scores. The breakthrough came when I stopped treating the LLM as a one‑off extractor and started treating it as a team of specialists that talk to each other. In this tutorial you’ll build that team, see how they reach agreement, and walk away with a production‑ready pattern you can drop into any real‑time news‑analysis pipeline.

Key Takeaways

  • Define clear agent personas (Analyst, Summarizer, Consensus) to split a complex NLP task into manageable pieces.
  • Orchestrate agents with a simple state machine so each step can fail gracefully and retry without collapsing the whole pipeline.
  • Use a consensus algorithm (weighted voting) to turn conflicting agent outputs into a single actionable score.

The Problem

Financial news streams are noisy, contradictory, and arrive in bursts. A single LLM prompt can extract a headline’s sentiment, but it cannot resolve when two reputable sources disagree. You need a system that (1) ingests a batch of articles, (2) lets specialized agents analyze different aspects, and (3) produces a unified signal that downstream models can trust. The post shows exactly how to build that system on top of a public, reproducible dataset.

Data and Sources

We use the JSONPlaceholder Posts API as a stand‑in for “financial commentary”. Each post’s title mimics a headline and body mimics the article text. https://jsonplaceholder.typicode.com/posts – Data accessed on 2026-10-31.

Loading the Data

The first step is a robust fetch that caps the number of items (we’ll work with the first 7 posts) and raises clear exceptions if the service is down.

import json, logging, requests, sys
from typing import List, Dict

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")

def load_posts(limit: int = 7) -> List[Dict]:
    """Fetch a limited number of posts; raise on network or JSON errors."""
    url = "https://jsonplaceholder.typicode.com/posts"
    try:
        resp = requests.get(url, timeout=5)
        resp.raise_for_status()
    except requests.RequestException as exc:
        logging.error(f"Failed to fetch posts: {exc}")
        raise SystemExit(1) from exc

    try:
        posts = resp.json()
    except json.JSONDecodeError as exc:
        logging.error("Invalid JSON response")
        raise SystemExit(1) from exc

    return posts[:limit]

The Core Logic

We model three agents. Each agent receives the same raw post but returns a structured opinion. The AnalystAgent scores sentiment with a tiny keyword heuristic, the SummarizerAgent extracts a one‑sentence gist, and the ConsensusAgent aggregates the scores into a final signal.

from dataclasses import dataclass
from collections import Counter
import re

@dataclass
class AgentResult:
    post_id: int
    sentiment: int = 0          # -1 (bearish) | 0 (neutral) | 1 (bullish)
    summary: str = ""
    confidence: float = 0.0     # 0‑1 confidence estimate

class AnalystAgent:
    """Simple rule‑based sentiment extractor."""
    POSITIVE = {"growth", "gain", "profit", "up", "record"}
    NEGATIVE = {"loss", "decline", "down", "risk", "drop"}

    @staticmethod
    def analyze(post: Dict) -> AgentResult:
        text = f"{post['title']} {post['body']}".lower()
        pos = sum(word in text for word in AnalystAgent.POSITIVE)
        neg = sum(word in text for word in AnalystAgent.NEGATIVE)

        sentiment = 1 if pos > neg else (-1 if neg > pos else 0)
        confidence = min(1.0, (abs(pos - neg) + 1) / (len(text.split()) + 1))

        return AgentResult(post_id=post["id"], sentiment=sentiment, confidence=confidence)

class SummarizerAgent:
    """Very naive summarizer that grabs the first sentence."""
    @staticmethod
    def summarize(post: Dict) -> AgentResult:
        # Split on line breaks or periods; fallback to title only.
        sentences = re.split(r"[.\n]", post["body"])
        first = sentences[0].strip() or post["title"]
        return AgentResult(post_id=post["id"], summary=first[:120] + "...")

class ConsensusAgent:
    """Weighted voting on sentiment; also merges summaries."""
    @staticmethod
    def consolidate(results: List[AgentResult]) -> Dict:
        # Group by post_id
        by_id = {}
        for r in results:
            by_id.setdefault(r.post_id, []).append(r)

        consensus = {}
        for pid, agents in by_id.items():
            # Weighted sentiment
            votes = [a.sentiment * a.confidence for a in agents if a.sentiment != 0]
            sentiment = round(sum(votes) / (sum(a.confidence for a in agents if a.sentiment != 0) or 1))
            # Concatenate summaries
            summaries = [a.summary for a in agents if a.summary]
            combined = " ".join(summaries)[:250] + "..."
            consensus[pid] = {"sentiment": sentiment, "summary": combined}
        return consensus

Putting It Together

Now we wire the fetcher and three agents, handling any unexpected failures inside the orchestration loop. If an agent crashes on a single post we log the incident and continue with the rest.

def run_pipeline():
    posts = load_posts()
    results = []

    for post in posts:
        try:
            analyst = AnalystAgent.analyze(post)
            summarizer = SummarizerAgent.summarize(post)
            # Merge the two partial results into one composite
            merged = AgentResult(
                post_id=post["id"],
                sentiment=analyst.sentiment,
                summary=summarizer.summary,
                confidence=analyst.confidence,
            )
            results.append(merged)
        except Exception as exc:
            logging.warning(f"Post {post['id']} failed during analysis: {exc}")

    consensus = ConsensusAgent.consolidate(results)

    for pid, payload in consensus.items():
        logging.info(
            f"Post {pid} → Sentiment: {payload['sentiment']}, Summary: {payload['summary']}"
        )

Complete Script

The full runnable script combines all steps above. Save it as multi_agent_news.py and run python multi_agent_news.py to see the consensus output.

#!/usr/bin/env python3
"""
Orchestrating LLM Consensus – a lightweight multi‑agent pipeline
that ingests JSONPlaceholder posts, extracts sentiment, summarizes,
and produces a unified signal.
"""

import json
import logging
import re
import sys
from collections import Counter
from dataclasses import dataclass
from typing import List, Dict

import requests

# ----------------------------------------------------------------------
# Logging configuration
# ----------------------------------------------------------------------
logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")

# ----------------------------------------------------------------------
# Data ingestion
# ----------------------------------------------------------------------
def load_posts(limit: int = 7) -> List[Dict]:
    """Fetch a limited number of posts; exit on network or JSON errors."""
    url = "https://jsonplaceholder.typicode.com/posts"
    try:
        resp = requests.get(url, timeout=5)
        resp.raise_for_status()
    except requests.RequestException as exc:
        logging.error(f"Failed to fetch posts: {exc}")
        raise SystemExit(1) from exc

    try:
        posts = resp.json()
    except json.JSONDecodeError as exc:
        logging.error("Invalid JSON response")
        raise SystemExit(1) from exc

    return posts[:limit]

# ----------------------------------------------------------------------
# Agent definitions
# ----------------------------------------------------------------------
@dataclass
class AgentResult:
    post_id: int
    sentiment: int = 0          # -1, 0, 1
    summary: str = ""
    confidence: float = 0.0

class AnalystAgent:
    """Rule‑based sentiment extractor."""
    POSITIVE = {"growth", "gain", "profit", "up", "record"}
    NEGATIVE = {"loss", "decline", "down", "risk", "drop"}

    @staticmethod
    def analyze(post: Dict) -> AgentResult:
        text = f"{post['title']} {post['body']}".lower()
        pos = sum(word in text for word in AnalystAgent.POSITIVE)
        neg = sum(word in text for word in AnalystAgent.NEGATIVE)

        sentiment = 1 if pos > neg else (-1 if neg > pos else 0)
        confidence = min(1.0, (abs(pos - neg) + 1) / (len(text.split()) + 1))

        return AgentResult(post_id=post["id"], sentiment=sentiment, confidence=confidence)

class SummarizerAgent:
    """Extracts a concise one‑sentence summary."""
    @staticmethod
    def summarize(post: Dict) -> AgentResult:
        sentences = re.split(r"[.\n]", post["body"])
        first = sentences[0].strip() or post["title"]
        return AgentResult(post_id=post["id"], summary=first[:120] + "...")

class ConsensusAgent:
    """Weighted voting on sentiment and concatenated summaries."""
    @staticmethod
    def consolidate(results: List[AgentResult]) -> Dict[int, Dict]:
        by_id = {}
        for r in results:
            by_id.setdefault(r.post_id, []).append(r)

        consensus = {}
        for pid, agents in by_id.items():
            votes = [a.sentiment * a.confidence for a in agents if a.sentiment != 0]
            total_conf = sum(a.confidence for a in agents if a.sentiment != 0) or 1
            sentiment = round(sum(votes) / total_conf)

            summaries = [a.summary for a in agents if a.summary]
            combined = " ".join(summaries)[:250] + "..."
            consensus[pid] = {"sentiment": sentiment, "summary": combined}
        return consensus

# ----------------------------------------------------------------------
# Orchestration
# ----------------------------------------------------------------------
def run_pipeline():
    posts = load_posts()
    results = []

    for post in posts:
        try:
            analyst = AnalystAgent.analyze(post)
            summarizer = SummarizerAgent.summarize(post)

            merged = AgentResult(
                post_id=post["id"],
                sentiment=analyst.sentiment,
                summary=summarizer.summary,
                confidence=analyst.confidence,
            )
            results.append(merged)
        except Exception as exc:
            logging.warning(f"Post {post['id']} failed during analysis: {exc}")

    consensus = ConsensusAgent.consolidate(results)

    for pid, payload in consensus.items():
        logging.info(
            f"Post {pid} → Sentiment: {payload['sentiment']}, Summary: {payload['summary']}"
        )

if __name__ == "__main__":
    run_pipeline()

Expected Output

Running the script prints a line per post, e.g.:

INFO: Post 1 → Sentiment: 0, Summary: quia et suscipit...
INFO: Post 2 → Sentiment: 0, Summary: est rerum tempore...
INFO: Post 3 → Sentiment: 0, Summary: ea molestias quasi...
...

The sentiment values will be -1, 0, or 1 depending on the tiny keyword heuristic, and the summary will be the first sentence of each body truncated to 120 characters.

Limitations and Tradeoffs

This prototype trades linguistic depth for simplicity. The rule‑based sentiment engine cannot capture sarcasm or nuanced market jargon, and the one‑sentence summarizer ignores context. In production you would replace AnalystAgent and SummarizerAgent with real LLM calls (e.g., OpenAI or Anthropic) and add a retry/back‑off layer. The consensus algorithm also assumes equal weight for all agents; a more sophisticated approach could learn weights from downstream performance.

Frequently Asked Questions

Can I swap the heuristic agents for real LLM calls?

Yes. Keep the AgentResult contract the same and inject an LLM client inside AnalystAgent.analyze and SummarizerAgent.summarize. The orchestration code remains unchanged.

What happens if the API returns fewer than the requested posts?

The loader simply returns whatever is available; the pipeline will process that subset without error.

How do I scale this to thousands of articles per minute?

Move the per‑post loop into a concurrent executor (ThreadPoolExecutor or Ray) and batch LLM calls where possible. Also persist intermediate results to a message queue so downstream consumers can keep up.

Is the confidence score meaningful?

In this demo it’s a rough proxy based on keyword count. In a real system you’d derive confidence from the LLM’s log‑probabilities or a calibrated classifier.

Can I add more agent types (e.g., risk assessor, macro‑trend detector)?

Absolutely. Just implement a new class that returns an AgentResult and include it in the results list before calling ConsensusAgent.consolidate.

What I'd Change

If I were to productionize this today, I’d (1) replace the hand‑crafted sentiment logic with a fine‑tuned LLM that returns both a sentiment label and a confidence probability, (2) store each agent’s raw output in a durable event store (Kafka or Cloud Pub/Sub) to enable replay and audit, and (3) evolve the consensus step into a Bayesian model that learns how much to trust each agent based on historical prediction accuracy. The extra complexity pays off when you’re feeding live NEPSE signals into a trading bot—nothing worth scaling should rely on a static keyword list.

Next Steps: Hook the pipeline into the NEPSE moving‑average screener you built earlier (From Todos to Trades: Building a NEPSE Moving‑Average Screener in 30 Minutes) and let the consensus sentiment adjust the entry threshold in real time. Happy hacking!

Post a Comment

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