Module 6: Prompt Composition and Chaining
2. Prompt Chaining: Output → Input
Overview
Prompt chaining is the fundamental pattern of prompt composition: the output of one prompt becomes the input of the next. This technique lets you build pipelines where each stage is a specialized prompt that transforms the information incrementally.
In this capsule you'll learn the three main chaining patterns: sequential (A→B→C), conditional (branching based on content), and parallel (multiple simultaneous prompts), along with robust error handling strategies.
Why Prompt Chaining
A single prompt has limitations:
❌ One prompt for a complex analysis:
"Read this contract, identify the parties, extract risk clauses,
determine whether the deadlines are reasonable, suggest changes, and generate
an executive summary in 3 languages."
Result: A shallow answer that tries to do everything halfway.
✓ Specialized chaining:
Prompt 1: Extract parties and structure → clean data
Prompt 2: Identify risk clauses → deep analysis
Prompt 3: Assess deadlines → focused evaluation
Prompt 4: Suggest changes → based on the real analysis
Prompt 5: Summary in 3 languages → translation of the final synthesis
Result: A detailed analysis on every dimension.
Pattern 1: Sequential Chaining
Base Implementation
from openai import OpenAI
from typing import Optional, Callable, Any
import json
client = OpenAI()
def call_llm(
prompt: str,
temperature: float = 0,
max_tokens: int = 500,
json_mode: bool = False
) -> str:
"""Basic helper to call the LLM."""
kwargs = {
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": prompt}],
"temperature": temperature,
"max_tokens": max_tokens
}
if json_mode:
kwargs["response_format"] = {"type": "json_object"}
return client.chat.completions.create(**kwargs).choices[0].message.content
def chain_sequential(
stages: list[dict],
initial_input: str,
verbose: bool = True
) -> dict:
"""
Runs a sequential chain of prompts.
Args:
stages: List of {name: str, prompt_template: str, temperature: float}
The prompt_template uses {input} to receive the previous output
initial_input: The input of the first stage
verbose: If True, prints the process
Returns:
dict with every output keyed by stage name
"""
outputs = {"_initial_input": initial_input}
current_input = initial_input
for stage in stages:
name = stage["name"]
template = stage["prompt_template"]
temperature = stage.get("temperature", 0)
json_mode = stage.get("json_mode", False)
max_tokens = stage.get("max_tokens", 500)
if verbose:
print(f"[Chain] Running stage: {name}")
# Format the template with the current input
prompt = template.format(
input=current_input,
**{k: v for k, v in outputs.items() if not k.startswith("_")}
)
output = call_llm(prompt, temperature, max_tokens, json_mode)
outputs[name] = output
current_input = output # The output becomes the input of the next stage
if verbose:
print(f" Output ({len(output)} chars): {output[:100]}...")
return outputs
# Example: Product review analysis pipeline
review_stages = [
{
"name": "extraction",
"prompt_template": """Extract from this product review:
1. Aspects mentioned (quality, price, shipping, etc.)
2. Opinion on each aspect (positive/negative/neutral)
3. Implicit rating (1-5)
Review: {input}
Respond in JSON: {{"aspects": [{{"name": str, "opinion": str, "rating": 1-5}}], "overall_rating": 1-5}}""",
"json_mode": True,
"temperature": 0
},
{
"name": "analysis",
"prompt_template": """Based on this analysis of a product review:
{input}
Identify:
1. Is this a genuine review or could it be fake? (authenticity signals)
2. Which aspects of the product need urgent improvement?
3. Which strengths should marketing highlight?
Answer concisely.""",
"temperature": 0
},
{
"name": "action",
"prompt_template": """Based on this analysis:
{input}
Generate 3 concrete actions the product team should take.
Each action: who does it, what they do, in how much time.
Format: bullet points.""",
"temperature": 0
}
]
review = "The product arrived in perfect condition and works exactly as described. The price is a bit high but the quality justifies it. Shipping took longer than expected (5 days instead of 2), but the seller was very attentive and kept me informed."
result = chain_sequential(review_stages, review, verbose=True)
print("\n=== FINAL OUTPUT (ACTIONS) ===")
print(result["action"])Chain with Format Transformation
from pydantic import BaseModel
from typing import Union
class ReviewExtraction(BaseModel):
aspects: list[dict]
overall_rating: int
def chain_with_validation(
review_text: str
) -> dict:
"""
Chain with schema validation between stages using Pydantic.
"""
# Stage 1: Structured extraction
extraction_prompt = f"""Extract from this review:
- aspects mentioned along with the opinion on each
- overall implicit rating (1-5)
Review: {review_text}
Respond in JSON: {{"aspects": [{{"name": str, "opinion": "positive|negative|neutral", "rating": 1-5}}], "overall_rating": int}}"""
output_json = call_llm(extraction_prompt, json_mode=True)
# Validate with Pydantic
try:
extraction = ReviewExtraction(**json.loads(output_json))
except Exception as e:
print(f"Validation error in stage 1: {e}. Retrying...")
output_json = call_llm(
extraction_prompt + "\n\nIMPORTANT: The response MUST be valid JSON with 'aspects' and 'overall_rating'.",
json_mode=True
)
extraction = ReviewExtraction(**json.loads(output_json))
# Stage 2: Analysis based on the validated extraction
aspects_str = json.dumps([a for a in extraction.aspects], ensure_ascii=False)
analysis_prompt = f"""Analyze these aspects of a product review:
{aspects_str}
Overall rating: {extraction.overall_rating}/5
What are the 2 most critical points that need improvement?"""
analysis = call_llm(analysis_prompt)
return {
"extraction": extraction.model_dump(),
"analysis": analysis
}Pattern 2: Conditional Chaining
from enum import Enum
class DocumentType(Enum):
CONTRACT = "contract"
INVOICE = "invoice"
EMAIL = "email"
REPORT = "report"
OTHER = "other"
def classify_document(text: str) -> DocumentType:
"""First step of the conditional chain: classify the document type."""
prompt = f"""Classify this text into one of these categories:
CONTRACT, INVOICE, EMAIL, REPORT, OTHER
Text (first 500 chars):
{text[:500]}
Respond with ONE word only: CONTRACT, INVOICE, EMAIL, REPORT, or OTHER"""
classification = call_llm(prompt).strip().upper()
mapping = {
"CONTRACT": DocumentType.CONTRACT,
"INVOICE": DocumentType.INVOICE,
"EMAIL": DocumentType.EMAIL,
"REPORT": DocumentType.REPORT
}
return mapping.get(classification, DocumentType.OTHER)
# Specialized prompts per document type
SPECIALIZED_PROMPTS = {
DocumentType.CONTRACT: """Analyze this legal contract. Identify:
1. Parties involved
2. Subject matter of the contract
3. Main obligations of each party
4. Termination clauses
5. Penalties, if any
Contract: {input}""",
DocumentType.INVOICE: """Extract the following information from this invoice:
1. Issuer and recipient
2. Invoice number and date
3. Invoice line items (description, quantity, unit price, total)
4. Taxable base, VAT, total
Invoice: {input}
Respond in structured JSON.""",
DocumentType.EMAIL: """Analyze this business email:
1. Sender and recipient
2. Main purpose
3. Action required (if any)
4. Tone and urgency
5. Suggested reply
Email: {input}""",
DocumentType.REPORT: """Analyze this report:
1. Main topic
2. Key metrics or data
3. Main conclusions
4. Recommendations, if any
Report: {input}""",
DocumentType.OTHER: """Summarize the following document:
- Type of content
- Main points
- Relevant information
Document: {input}"""
}
def chain_conditional_document(text: str, verbose: bool = True) -> dict:
"""
Conditional chain: classify the document and then apply
the specialized analysis for that type.
"""
# Stage 1: Classification (always runs)
doc_type = classify_document(text)
if verbose:
print(f"[Conditional Chain] Detected type: {doc_type.value}")
# Stage 2: Specialized analysis (conditional on the type)
specialized_prompt = SPECIALIZED_PROMPTS[doc_type].format(input=text)
analysis = call_llm(
specialized_prompt,
json_mode=(doc_type == DocumentType.INVOICE)
)
# Stage 3: Recommended actions (always runs, but with the type as context)
actions_prompt = f"""Given this analysis of a {doc_type.value} document:
{analysis}
Which 2-3 concrete actions should the recipient of this document take?"""
actions = call_llm(actions_prompt)
return {
"document_type": doc_type.value,
"specialized_analysis": analysis,
"recommended_actions": actions
}
# Example:
documents = [
"Dear customer, attached you will find invoice #2026-0342 for services rendered in February...",
"By this contract, PARTY A undertakes to deliver the services described in Annex I...",
]
for doc in documents:
result = chain_conditional_document(doc, verbose=True)
print(f" Analysis: {result['specialized_analysis'][:100]}...\n")Pattern 3: Parallel Chaining
import asyncio
from openai import AsyncOpenAI
async def chain_parallel_analysis(
article: str,
requested_analyses: list[str] = None
) -> dict:
"""
Runs multiple analyses in parallel on the same article.
Reduces total latency significantly.
Args:
article: Text of the article to analyze
requested_analyses: List of desired analyses. Default: all of them.
Returns:
dict with every completed analysis
"""
client_async = AsyncOpenAI()
if requested_analyses is None:
requested_analyses = ["summary", "keywords", "sentiment", "entities"]
PARALLEL_PROMPTS = {
"summary": f"Summarize in 3 concise sentences: {article}",
"keywords": f"""Extract 5-8 keywords from the following article.
Format: comma-separated list.
Article: {article}""",
"sentiment": f"""Analyze the sentiment and tone of this article.
State: sentiment (POSITIVE/NEGATIVE/NEUTRAL), tone (formal/informal/technical), and emotional intensity (high/medium/low).
Article: {article}""",
"entities": f"""Extract named entities from this article.
Categories: people, organizations, places, dates/periods, products.
Article: {article}
JSON format.""",
"questions": f"""What are the 3 most important questions this article answers?
Article: {article}""",
"audience": f"""Who is this article mainly written for?
Describe the target audience and why.
Article: {article}"""
}
async def run_analysis(name: str) -> tuple[str, str]:
"""Runs an individual analysis."""
if name not in PARALLEL_PROMPTS:
return name, f"Analysis '{name}' not available"
response = await client_async.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": PARALLEL_PROMPTS[name]}],
temperature=0,
max_tokens=300
)
return name, response.choices[0].message.content
# Run all of them in parallel
import time
t0 = time.time()
tasks = [run_analysis(name) for name in requested_analyses]
raw_results = await asyncio.gather(*tasks)
total_time = time.time() - t0
results = dict(raw_results)
results["_metadata"] = {
"total_time": total_time,
"n_analyses": len(requested_analyses),
"avg_time_if_sequential": total_time * len(requested_analyses)
}
return results
# To run in a normal script (not a notebook):
async def parallel_example():
article = """Artificial intelligence is transforming the financial industry.
Machine learning algorithms make it possible to detect fraud in real time
with 99.5% accuracy, reducing losses by 60%..."""
import time
t0 = time.time()
results = await chain_parallel_analysis(
article,
requested_analyses=["summary", "keywords", "sentiment", "entities"]
)
t_total = time.time() - t0
print(f"Total time: {t_total:.2f}s (vs ~{t_total * 4:.1f}s if it were sequential)")
for name, result in results.items():
if not name.startswith("_"):
print(f"\n{name.upper()}:\n{result[:150]}...")
# asyncio.run(parallel_example()) # Uncomment to runPattern 4: Fan-Out / Fan-In
async def chain_fan_out_fan_in(
document: str,
n_perspectives: int = 3
) -> dict:
"""
Fan-Out: Analyzes the document from multiple perspectives in parallel.
Fan-In: Synthesizes every analysis into one coherent output.
Useful for multi-dimensional analysis where each perspective
can reveal different aspects of the same content.
"""
client_async = AsyncOpenAI()
perspectives = [
("technical", "Analyze the technical aspects, accuracy and methodology"),
("business", "Analyze the implications for business and ROI"),
("end_user", "Analyze the impact on the end user"),
("ethical", "Analyze ethical and privacy considerations"),
("competitive", "Analyze it in the context of the competitive landscape")
][:n_perspectives]
async def analyze_perspective(name: str, instruction: str) -> tuple[str, str]:
response = await client_async.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"""Analyze this document from the {name} perspective:
{instruction}
Document: {document[:3000]}
Provide 3 key points from this perspective."""
}],
temperature=0,
max_tokens=300
)
return name, response.choices[0].message.content
# Fan-Out: multiple analyses in parallel
tasks = [analyze_perspective(n, i) for n, i in perspectives]
perspective_results = dict(await asyncio.gather(*tasks))
# Fan-In: synthesize every analysis
perspectives_str = "\n\n".join([
f"{name.upper()} PERSPECTIVE:\n{analysis}"
for name, analysis in perspective_results.items()
])
synthesis = client_async # Reuse the async client
synthesis_resp = await synthesis.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"""You have {n_perspectives} different analyses of the same document:
{perspectives_str}
Synthesize a holistic 3-5 paragraph assessment that integrates every perspective.
Highlight: agreements between perspectives, tensions or contradictions, and an overall conclusion."""
}],
temperature=0,
max_tokens=600
)
return {
"perspectives": perspective_results,
"synthesis": synthesis_resp.choices[0].message.content
}Robust Error Handling in Chains
from typing import Optional
import logging
logger = logging.getLogger(__name__)
class StageError(Exception):
"""Error specific to one stage of the chain."""
def __init__(self, stage: str, message: str, input_data: str = ""):
self.stage = stage
self.message = message
self.input_data = input_data
super().__init__(f"Error in stage '{stage}': {message}")
def chain_with_error_handling(
stages: list[dict],
initial_input: str,
max_retries: int = 2,
fallback_output: Optional[dict] = None
) -> dict:
"""
Sequential chain with robust error handling.
Features:
- Automatic retry per stage
- Improved prompt on retry
- Fallback when retries run out
- Error logging
"""
outputs = {"_initial_input": initial_input}
current_input = initial_input
for stage in stages:
name = stage["name"]
template = stage["prompt_template"]
temperature = stage.get("temperature", 0)
json_mode = stage.get("json_mode", False)
validator = stage.get("validator", None) # Optional validation function
last_error = None
for attempt in range(max_retries + 1):
try:
prompt = template.format(
input=current_input,
**{k: v for k, v in outputs.items() if not k.startswith("_")}
)
# On retries, make the prompt more explicit
if attempt > 0:
prompt += f"\n\nIMPORTANT: Attempt {attempt+1}. Previous error: {last_error}. Please respond in the exact requested format."
output = call_llm(prompt, temperature, 600, json_mode)
# Optional validation
if validator:
is_valid, validation_error = validator(output)
if not is_valid:
raise ValueError(f"Validation failed: {validation_error}")
outputs[name] = output
current_input = output
break # Success, exit the retry loop
except Exception as e:
last_error = str(e)
logger.warning(f"Attempt {attempt+1} failed in stage '{name}': {e}")
if attempt == max_retries:
# Retries exhausted
if fallback_output and name in fallback_output:
outputs[name] = fallback_output[name]
current_input = fallback_output[name]
outputs[f"_{name}_failed"] = True
logger.error(f"Using fallback for stage '{name}'")
else:
raise StageError(name, last_error, current_input)
return outputs
# Example validator for JSON
def validate_json_with_fields(required_fields: list[str]):
"""Builds a validator that checks the JSON has the required fields."""
def validate(output: str) -> tuple[bool, str]:
try:
data = json.loads(output)
for field in required_fields:
if field not in data:
return False, f"Field '{field}' is missing from the JSON"
return True, ""
except json.JSONDecodeError as e:
return False, f"Invalid JSON: {e}"
return validate
# Usage:
stages_with_validation = [
{
"name": "extraction",
"prompt_template": "Extract the entities from: {input}\nJSON: {{\"persons\": [], \"organizations\": [], \"dates\": []}}",
"json_mode": True,
"temperature": 0,
"validator": validate_json_with_fields(["persons", "organizations"])
}
]
try:
result = chain_with_error_handling(
stages_with_validation,
"Apple launched the iPhone 17 on March 15, 2026. Tim Cook presented the device.",
max_retries=2
)
print(result)
except StageError as e:
print(f"Unrecoverable error in stage '{e.stage}': {e.message}")Real Use Cases
Case 1: Resume Processing Pipeline
def pipeline_cv(cv_text: str) -> dict:
"""Complete pipeline to analyze a resume."""
stages = [
{
"name": "extraction",
"prompt_template": """Extract from the following resume:
1. Name and contact details
2. Work experience (company, role, duration)
3. Education
4. Technical skills
5. Languages
Resume: {input}
Respond in structured JSON.""",
"json_mode": True
},
{
"name": "evaluation",
"prompt_template": """Based on this resume extraction:
{input}
Evaluate:
1. Total years of experience
2. Estimated technical level (junior/mid/senior)
3. Main area of specialization
4. Strengths (3 points)
5. Areas for improvement (2 points)
Be objective and base everything on evidence from the resume."""
},
{
"name": "interview_questions",
"prompt_template": """Given this resume analysis:
{input}
Generate 5 interview questions specific to this candidate.
Include: 2 technical, 2 about experience, 1 about a difficult situation.
Each question in the format: "Question: [question] (Goal: [what it assesses])"
"""
}
]
return chain_with_error_handling(stages, cv_text)
# Test:
example_cv = """
Juan García
Backend Developer | juan@email.com | LinkedIn: /in/jgarcia
5+ years of experience in Python, FastAPI, PostgreSQL.
Working at TechCorp (2021-present) as Lead Backend Developer...
"""
result = pipeline_cv(example_cv)
print("EVALUATION:", result["evaluation"][:300])Case 2: Parallel Competitor Analysis
async def analyze_competitors(company: str, competitors: list[str]) -> dict:
"""
Analyzes multiple competitors in parallel and compares them with the main company.
"""
client_async = AsyncOpenAI()
async def analyze_competitor(competitor_name: str) -> tuple[str, str]:
response = await client_async.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"""Briefly analyze {competitor_name} as a competitor of {company}:
1. Main strengths (2-3 points)
2. Weaknesses (2 points)
3. Differentiators vs {company}
(Use general knowledge; state when something is uncertain)"""
}],
temperature=0,
max_tokens=250
)
return competitor_name, response.choices[0].message.content
# Fan-out: analyze all of them in parallel
tasks = [analyze_competitor(c) for c in competitors]
individual_analyses = dict(await asyncio.gather(*tasks))
# Fan-in: comparative synthesis
comparisons = "\n\n".join([f"### {comp}:\n{analysis}"
for comp, analysis in individual_analyses.items()])
synthesis_resp = await client_async.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"""Competitor analysis for {company}:
{comparisons}
Generate:
1. Comparison table with strengths/weaknesses
2. Main competitive threats
3. Differentiation opportunities for {company}"""
}],
temperature=0,
max_tokens=600
)
return {
"individual_analyses": individual_analyses,
"comparative_synthesis": synthesis_resp.choices[0].message.content
}
# asyncio.run(analyze_competitors("FastAPI", ["Django", "Flask", "Express.js"]))Troubleshooting
Problem 1: Truncated output between stages
Symptom: The output of stage 1 is cut off and stage 2 works with incomplete information.
Causes:
max_tokensset too low- The model generates more tokens than expected
Solution:
def chain_with_truncation_check(
template: str,
input_data: str,
max_tokens: int = 500
) -> str:
"""Detects and handles truncated outputs."""
output = call_llm(template.format(input=input_data), max_tokens=max_tokens)
# Truncation signals
truncation_signals = [
output.endswith("..."),
output.endswith(","),
len(output.split()) > max_tokens * 0.9,
output.count("{") != output.count("}"), # Unbalanced JSON
]
if any(truncation_signals):
# Try again with more tokens, or ask it to complete
continuation = call_llm(
f"{output}\n\n[CONTINUE from where you were cut off]",
max_tokens=max_tokens * 2
)
return output + continuation
return outputProblem 2: Inconsistent format between stages
Symptom: Stage 2 expects JSON but stage 1 returns free text.
Solution: Use json_mode=True when you need JSON, and validate with Pydantic:
def stage_with_guaranteed_json(prompt: str) -> dict:
"""Guarantees the output is valid JSON."""
max_attempts = 3
for attempt in range(max_attempts):
output = call_llm(
prompt if attempt == 0 else f"{prompt}\n\nCRITICAL: Respond with valid JSON ONLY.",
json_mode=True
)
try:
return json.loads(output)
except json.JSONDecodeError:
if attempt == max_attempts - 1:
return {"error": "Could not parse JSON", "raw": output}
return {}Problem 3: Error cascade
Symptom: An error in stage 2 makes stage 3 fail too, and so on down the chain.
Solution: Every stage must have an independent fallback:
def stage_with_fallback(
template: str,
input_data: str,
fallback_message: str = "Analysis not available for this stage"
) -> str:
"""Individual stage with a fallback."""
try:
return call_llm(template.format(input=input_data))
except Exception as e:
print(f"⚠ Stage failed: {e}. Using fallback.")
return fallback_messageExercises
Exercise 1: Feedback analysis chain
Implement a 3-stage chain to analyze user feedback:
- Categorize the feedback (bug, feature request, complaint, compliment)
- Assess the priority (high/medium/low) based on category and urgency
- Generate an automatic reply appropriate to the type
See solution
def chain_user_feedback(feedback: str) -> dict:
stages = [
{
"name": "categorization",
"prompt_template": """Categorize this user feedback:
Categories: BUG, FEATURE_REQUEST, COMPLAINT, COMPLIMENT, QUESTION
Feedback: {input}
Respond in JSON: {{"category": str, "subcategory": str, "urgency": "high|medium|low", "sentiment": "positive|negative|neutral"}}""",
"json_mode": True
},
{
"name": "priority",
"prompt_template": """Given this feedback analysis:
{input}
Determine:
1. Attention priority (1-5, where 5 is the most urgent)
2. Responsible team (engineering, product, support, marketing)
3. Suggested response time (immediate/24h/72h/week)
JSON: {{"priority": int, "team": str, "response_time": str, "reason": str}}""",
"json_mode": True
},
{
"name": "auto_reply",
"prompt_template": """Original feedback: {_initial_input}
Analysis: {priority}
Generate an empathetic, professional automatic reply for this feedback.
- If it's a BUG: Thank them for the report, say it will be investigated
- If it's a FEATURE_REQUEST: Thank them for the suggestion, say it will be considered
- If it's a COMPLAINT: Empathize, apologize if it applies, state the action
- If it's a COMPLIMENT: Thank them genuinely
Maximum 3 sentences."""
}
]
return chain_with_error_handling(stages, feedback, max_retries=1)
test_feedback = "I've spent 3 days trying to export my data to CSV and the button doesn't work. It's urgent because I have a presentation tomorrow and I need that data."
result = chain_user_feedback(test_feedback)
print("Reply:", result["auto_reply"])Exercise 2: Chain with branching by complexity
Implement a chain that:
- Analyzes the complexity of a technical question (simple/complex)
- If it's simple: answer directly
- If it's complex: decompose into sub-questions, answer each one, synthesize
See solution
def adaptive_question_chain(question: str) -> dict:
# Step 1: Assess complexity
classification = call_llm(f"Is this question SIMPLE (direct answer) or COMPLEX (requires multiple steps)?\nQuestion: {question}\nRespond with only: SIMPLE or COMPLEX")
if "SIMPLE" in classification.upper():
answer = call_llm(f"Answer concisely: {question}")
return {"complexity": "simple", "answer": answer}
# Step 2 (only for complex ones): Decompose
sub_questions_raw = call_llm(f"Break this complex question down into 2-4 simpler sub-questions:\n{question}\nFormat: '1. ...' '2. ...'")
sub_questions = [l.lstrip('0123456789.-) ').strip()
for l in sub_questions_raw.split('\n')
if l.strip() and l.strip()[0].isdigit()]
# Step 3: Answer each sub-question
partial_answers = []
for sq in sub_questions[:4]:
resp = call_llm(f"Answer briefly: {sq}")
partial_answers.append(f"Q: {sq}\nA: {resp}")
# Step 4: Synthesize
context = "\n\n".join(partial_answers)
synthesis = call_llm(f"Original question: {question}\n\nPartial answers:\n{context}\n\nSynthesize a complete answer.")
return {
"complexity": "complex",
"sub_questions": sub_questions,
"partial_answers": partial_answers,
"synthesis": synthesis
}Exercise 3: Parallel content analysis pipeline
Implement an async pipeline that analyzes an article in parallel for: summary, keywords, target audience, and topic classification. The output must be structured JSON.
See solution
async def article_analysis_pipeline(article: str) -> dict:
client_async = AsyncOpenAI()
ANALYSES = {
"summary": f"Generate a 2-sentence executive summary: {article[:2000]}",
"keywords": f"List the 5 main keywords, comma-separated: {article[:2000]}",
"audience": f"Who is this article written for? (2-3 sentences): {article[:2000]}",
"topic": f"Classify into ONE topic: technology, business, science, politics, culture, sports, other. One word only: {article[:500]}"
}
async def run(name: str, prompt: str) -> tuple[str, str]:
r = await client_async.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0, max_tokens=200
)
return name, r.choices[0].message.content.strip()
results = dict(await asyncio.gather(*[run(n, p) for n, p in ANALYSES.items()]))
return results
# To run it:
# result = asyncio.run(article_analysis_pipeline("The text of the article..."))Summary
- Sequential chaining: A→B→C, the output of each stage feeds the next one. Simple but effective for linear processes.
- Conditional chaining: Branching based on the content. Enables specialization per input type.
- Parallel chaining: Multiple simultaneous analyses with asyncio. Cuts latency from N*T down to ~T.
- Fan-out/fan-in: Multiple parallel analyses + synthesis. Ideal for multi-perspective analysis.
- Error handling: Retry per stage, validation with Pydantic, fallback, logging. Never propagate errors without context.