Multi-step LLM Pipeline

Path: 1
Position: 5 of 12
Difficulty: 🟢 Local — one API key, no Docker, no cloud
Estimated time: 3–4 hours
AWS cost: None

---

Agent Pickup Instructions

This spec is self-contained. Build this project without reading any other spec file.

mkdir -p path-1/p1-05-multi-step-llm-pipeline
cd path-1/p1-05-multi-step-llm-pipeline

pytest tests/ -v
python src/main.py "The James Webb Space Telescope"
python src/main.py "Python programming language" --output briefing.md

Done when:

• pytest tests/ -v → minimum 4 tests, all green
• Running on any topic produces a 3-stage pipeline trace in stdout
• Each stage's latency is logged independently
• Output is a structured briefing saved to a file when --output is specified
• Pipeline fails gracefully (per-stage error with clear message) if any stage fails
• No file in src/ exceeds 200 lines

---

What this project is

A 3-stage pipeline that: (1) extracts named entities from a topic description, (2) fetches a Wikipedia summary for each entity via HTTP, and (3) synthesises a professional briefing document from all retrieved summaries. The pipeline is fixed-sequence — the learner controls what each stage does and when it runs, unlike an agent where the LLM decides. This is the "fixed gears before automatic transmission" moment: once you've built a fixed chain, the concept of a dynamic agent (Project 6) becomes concrete and motivated.

---

What the learner achieves

"I built a 3-stage LLM pipeline that extracts entities, enriches them with live Wikipedia data, and synthesises a briefing — with per-stage latency logged and graceful error handling when any stage fails."

---

Folder structure

p1-05-multi-step-llm-pipeline/
├── README.md
├── GUIDE.md
├── .env.example
├── requirements.txt
├── src/
│   ├── main.py         ← CLI entry point
│   ├── llm.py          ← provider-agnostic LLM wrapper
│   ├── stages.py       ← three pipeline stage functions
│   └── pipeline.py     ← orchestrator that runs stages in sequence
└── tests/
    └── test_stages.py

---

.env.example

LLM_PROVIDER=anthropic
LLM_API_KEY=
LLM_MODEL=

# Maximum entities to enrich per run (Wikipedia fetch per entity)
MAX_ENTITIES=5

# Wikipedia API timeout in seconds
WIKIPEDIA_TIMEOUT_SECONDS=5

---

requirements.txt

anthropic==0.40.0
openai==1.58.0
ollama==0.4.4
requests==2.32.3
python-dotenv==1.0.1

---

src/ — what to implement

src/llm.py

get_completion(prompt: str, system: str = "") -> str

• Standard provider router (anthropic / openai / ollama)
• Raises ValueError for unknown provider

src/stages.py

Stage 1 — extract_entities(topic: str) -> list[str]

• System prompt: "Extract named entities from the input. Return ONLY a JSON array of strings. Include people, organisations, technologies, places, and concepts central to the topic. Maximum {MAX_ENTITIES} items."
• User prompt: "Topic: {topic}"
• Parse response: strip JSON fences, json.loads(), validate it is a list of strings
• Raises StageError("entity extraction", reason) on parse failure
• Returns deduplicated list of entity names

Stage 2 — fetch_entity_summaries(entities: list[str]) -> list[dict]

• For each entity, call Wikipedia REST API: https://en.wikipedia.org/api/rest_v1/page/summary/{entity_url_encoded}
• Returns [{"entity": name, "summary": text, "url": wiki_url, "found": bool}, ...]
• If entity not found (404): set "found": False, "summary": f"No Wikipedia article found for '{entity}'"
• Timeout: WIKIPEDIA_TIMEOUT_SECONDS (default 5)
• On requests.Timeout: set "found": False, "summary": "Wikipedia fetch timed out"
• Do NOT raise on individual entity failures — record them and continue

Stage 3 — synthesise_briefing(topic: str, entity_summaries: list[dict]) -> str

• System prompt: "You are a professional analyst. Write a concise briefing document."
• User prompt includes: the topic, each entity summary formatted as ## {entity}\n{summary}\n, followed by: "Write a 300–400 word briefing covering: (1) what this topic is about, (2) the key entities involved, (3) why it matters. Cite entities inline."
• Returns the briefing text as a string

StageError(Exception)

• Custom exception: StageError(stage_name: str, reason: str)
• str() returns: "Stage '{stage_name}' failed: {reason}"

src/pipeline.py

PipelineResult dataclass

@dataclass
class PipelineResult:
    topic: str
    entities: list[str]
    entity_summaries: list[dict]
    briefing: str
    stage_latencies_ms: dict[str, int]   # {"extract": 450, "fetch": 1200, "synthesise": 1800}
    total_latency_ms: int
    errors: list[str]                     # non-fatal warnings (e.g., Wikipedia misses)

run_pipeline(topic: str) -> PipelineResult

• Run Stage 1, 2, 3 in sequence
• Time each stage: t = time.perf_counter(); result = stage_fn(...); latency = int((time.perf_counter() - t) * 1000)
• Print progress: [Stage 1/3] Extracting entities... → ✓ Found 4 entities (450ms)
• If any stage raises StageError: print error, raise it (do not continue to next stage)
• Collect non-found Wikipedia entities in errors
• Return PipelineResult

src/main.py

CLI: python src/main.py "<topic>" [--output <file.md>]

1. Print \nRunning pipeline for: "{topic}"\n
2. Call run_pipeline(topic)
3. Print pipeline trace:

   [Stage 1/3] Extracting entities...       ✓ 4 entities (312ms)
   [Stage 2/3] Fetching Wikipedia data...   ✓ 3/4 found (1,240ms)
   [Stage 3/3] Synthesising briefing...     ✓ done (1,820ms)
   
   Total: 3,372ms
   

4. Print briefing to stdout
5. If --output: write briefing to file, print Saved to {file}
6. If any Wikipedia entities not found: print ⚠ Not found: [entity1, entity2]

---

tests/ — what to test

File: tests/test_stages.py — mock HTTP calls and LLM calls.

Test 1 — extract_entities returns list of strings: Mock llm.get_completion to return '["Python", "Guido van Rossum", "CPython"]'. Call extract_entities("Python language"). Assert result == ["Python", "Guido van Rossum", "CPython"].

Test 2 — extract_entities raises StageError on bad JSON: Mock get_completion to return "not json". Assert extract_entities() raises StageError.

Test 3 — fetch_entity_summaries handles 404: Mock requests.get to return 404 for one entity and 200 with a summary for another. Assert result list has "found": False for 404 and "found": True for 200.

Test 4 — fetch_entity_summaries handles timeout: Mock requests.get to raise requests.Timeout. Assert entity in result has "found": False with "timed out" in summary. Assert no exception raised from function.

Test 5 — pipeline stage latencies are recorded: Mock all three stages. Call run_pipeline("test topic"). Assert result.stage_latencies_ms has keys "extract", "fetch", "synthesise" and all values are integers >= 0.

---

README.md content

# Multi-step LLM Pipeline

A 3-stage pipeline that extracts entities from any topic, enriches them with
live Wikipedia data, and synthesises a professional briefing document.

## Setup

```bash
cd p1-05-multi-step-llm-pipeline
cp .env.example .env
pip install -r requirements.txt

Run

python src/main.py "The James Webb Space Telescope"
python src/main.py "Kubernetes" --output k8s_briefing.md

Expected output:

Running pipeline for: "The James Webb Space Telescope"

[Stage 1/3] Extracting entities...       ✓ 5 entities (312ms)
[Stage 2/3] Fetching Wikipedia data...   ✓ 5/5 found (1,102ms)
[Stage 3/3] Synthesising briefing...     ✓ done (2,140ms)

Total: 3,554ms

# Briefing: The James Webb Space Telescope
The James Webb Space Telescope (JWST) is a space telescope designed to...

Tests

pytest tests/ -v

Tests verify entity extraction parsing, Wikipedia error handling, and pipeline latency tracking. No API key required.

What to try next

• Add a Stage 4: generate 3 follow-up questions about the briefing
• Add a cache for Wikipedia results so the same entity isn't fetched twice
• Try topic = a company name and use the briefing for interview prep

---

## GUIDE.md content

```markdown
# Build guide: Multi-step LLM Pipeline

## What you're building and why it matters

Most useful AI applications are not single LLM calls — they are sequences of calls
where each stage transforms or enriches the data. A legal contract analyser might:
extract clauses, classify each clause, flag risk, then summarise. A customer support
tool might: classify the ticket, retrieve relevant docs, generate a draft response,
then check tone. Understanding how to orchestrate these stages — in sequence, with
error handling and latency tracking per stage — is the foundation of production AI
engineering.

## The decision that matters in this build

**How much to pass between stages.** Stage 1 extracts entities. Stage 2 fetches
summaries. You could pass everything from Stage 1 to Stage 2 (all text, all context).
Instead, pass only what Stage 2 needs: the entity names. This keeps stages decoupled.
If Stage 1's output format changes, Stage 2 doesn't break as long as it still receives
a list of strings. Decoupled stages are the only kind you can test independently,
swap out, or run in parallel.

## What will break

**Wikipedia API throttles aggressive requests.** If you fetch summaries for 10 entities
sequentially with no delay, you may get rate-limited. Add a `time.sleep(0.1)` between
calls, or implement concurrent fetching with `ThreadPoolExecutor` for a more robust
solution.

**LLMs return varying JSON formats for entity extraction.** Claude wraps arrays in
```json\n...\n``` fences sometimes. OpenAI returns them clean. Always strip fences
before parsing. Test extraction against both providers if you have access to both.

## How to talk about this in an interview

"I built a 3-stage LLM pipeline where each stage has a clear input/output contract
and fails independently. I time each stage separately — which showed that Wikipedia
fetch is often the bottleneck, not the LLM. I also learned that passing minimal data
between stages (just entity names, not full text) is what makes each stage independently
testable and swappable."

---

Failure modes to handle

Failure	Where	How to handle
Bad JSON from entity extraction	stages.extract_entities()	Raise StageError("entity extraction", f"JSON parse failed: {err}")
Empty entity list	stages.extract_entities()	Raise StageError("entity extraction", "LLM returned empty entity list")
Wikipedia 404	stages.fetch_entity_summaries()	Record found=False, continue with other entities
Wikipedia timeout	stages.fetch_entity_summaries()	Record found=False with timeout note, continue
Stage 3 LLM error	stages.synthesise_briefing()	Raise StageError("synthesis", str(err))
Output file write fails	main.py	Print error message and exit with code 1

---

The metric this project measures

Per-stage latency in milliseconds — printed in the pipeline trace. Format: [Stage N/3] {name}... ✓ done ({latency}ms) Wikipedia hit rate — fraction of entities found: Found 4/5 entities. These show the learner where time is spent (usually Wikipedia or LLM generation, not entity extraction).

Model source currency

• Anthropic model list: https://docs.anthropic.com/en/docs/about-claude/models/overview
• OpenAI model list: https://platform.openai.com/docs/models
• verifiedOn: 2026-08-07
• LLM_MODEL is required and has no implicit default; choose a supported model for the configured provider at setup time.