--- title: "Multi-step LLM Pipeline" description: "Multi-step pipelines are how real AI products are actually built — this is the skill that gets you past 'I can call an API' to 'I can design an AI system.'" source: "https://confidentprep.com/paths/ai-engineering-fundamentals/multi-step-llm-pipeline/" path: "AI Engineering Fundamentals" repository: "https://github.com/confidentprep14-web/ai-engineering-projects/tree/main/multi-step-llm-pipeline" token_estimate: 3111 --- # Multi-step LLM Pipeline ## Overview A 3-stage pipeline: extract entities from any topic, enrich them with live Wikipedia data, then synthesise a professional briefing document. Single-prompt thinking only gets you so far — this is your first real exercise in decomposing a problem into stages, passing structured data between them, and handling failure at each step instead of one giant prompt that's impossible to debug. ## What to do **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.** ```bash 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 ```bash 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** ```python @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 "" [--output ]` 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 ```markdown # 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 ```bash 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 ```bash 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. ## Source code A reference implementation of Multi-step LLM Pipeline — README, guide, source and tests — is at https://github.com/confidentprep14-web/ai-engineering-projects/tree/main/multi-step-llm-pipeline.