16 Practice Lab: Build a Stateful AI Workflow
16.1 What You’ll Build
A Multi-Agent Report Generator that: - Researches a topic across multiple dimensions - Reviews and revises content iteratively - Formats a structured report - Allows human-in-the-loop approval
16.2 Step 1: Define State and Agents
# file: report_workflow.py
from typing import TypedDict, Annotated, List
from operator import add
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
class ReportState(TypedDict):
topic: str
sections: dict # section_name -> content
quality_scores: dict # section_name -> score
revision_count: int
final_report: str
messages: Annotated[List, add]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
reviewer_llm = ChatOpenAI(model="gpt-4o", temperature=0) # Stronger for review
REPORT_SECTIONS = [
"executive_summary",
"market_analysis",
"technical_assessment",
"recommendations"
]
# --- Research Agent ---
def research_agent(state: ReportState) -> ReportState:
"""Research the topic and generate all sections."""
topic = state["topic"]
sections = {}
section_prompts = {
"executive_summary": f"Write a 150-word executive summary on: {topic}",
"market_analysis": f"Analyse the market opportunity for: {topic} in 200 words",
"technical_assessment": f"Assess the technical requirements and challenges of: {topic} in 200 words",
"recommendations": f"Provide 5 strategic recommendations for: {topic}"
}
for section, prompt in section_prompts.items():
response = llm.invoke(prompt)
sections[section] = response.content
print(f" ✅ Generated: {section}")
return {"sections": sections, "revision_count": 0, "quality_scores": {}}
# --- Quality Review Agent ---
def review_agent(state: ReportState) -> ReportState:
"""Review each section and score quality."""
quality_scores = {}
for section, content in state["sections"].items():
prompt = f"""Rate this {section.replace('_', ' ')} section from 1-10.
Return only a number.
Content: {content}"""
response = reviewer_llm.invoke(prompt)
try:
score = float(response.content.strip())
except:
score = 7.0
quality_scores[section] = score
print(f" 📊 {section}: {score}/10")
return {"quality_scores": quality_scores}
# --- Revision Agent ---
def revision_agent(state: ReportState) -> ReportState:
"""Revise sections that scored below threshold."""
threshold = 7.5
revised_sections = dict(state["sections"])
for section, score in state["quality_scores"].items():
if score < threshold:
print(f" 🔄 Revising: {section} (score: {score})")
response = llm.invoke(
f"Improve this content significantly:\n\n{state['sections'][section]}"
)
revised_sections[section] = response.content
return {
"sections": revised_sections,
"revision_count": state["revision_count"] + 1
}
# --- Report Assembly Agent ---
def assembly_agent(state: ReportState) -> ReportState:
"""Assemble the final report."""
sections = state["sections"]
avg_score = sum(state["quality_scores"].values()) / len(state["quality_scores"]) if state["quality_scores"] else 0
report = f"""# {state['topic'].title()}
## Strategic Analysis Report
*Generated by AI | Quality Score: {avg_score:.1f}/10 | Revisions: {state['revision_count']}*
---
## Executive Summary
{sections.get('executive_summary', '')}
---
## Market Analysis
{sections.get('market_analysis', '')}
---
## Technical Assessment
{sections.get('technical_assessment', '')}
---
## Strategic Recommendations
{sections.get('recommendations', '')}
---
*Report generated using RAG + LangGraph workflow*
"""
return {"final_report": report}
# --- Routing Logic ---
def should_revise(state: ReportState) -> str:
"""Decide: revise more or finalise?"""
scores = state["quality_scores"]
avg_score = sum(scores.values()) / len(scores) if scores else 0
low_quality = [s for s, score in scores.items() if score < 7.5]
if state["revision_count"] >= 3 or (avg_score >= 8.0 and not low_quality):
print(f" ✅ Quality sufficient (avg: {avg_score:.1f}). Finalising...")
return "assemble"
print(f" 🔄 Needs revision (avg: {avg_score:.1f}, low: {low_quality})")
return "revise"
# --- Build the Graph ---
def build_report_workflow():
workflow = StateGraph(ReportState)
workflow.add_node("research", research_agent)
workflow.add_node("review", review_agent)
workflow.add_node("revise", revision_agent)
workflow.add_node("assemble", assembly_agent)
workflow.set_entry_point("research")
workflow.add_edge("research", "review")
workflow.add_conditional_edges(
"review",
should_revise,
{"assemble": "assemble", "revise": "revise"}
)
workflow.add_edge("revise", "review")
workflow.add_edge("assemble", END)
return workflow.compile()
# --- Run it ---
if __name__ == "__main__":
app = build_report_workflow()
topic = input("Enter report topic: ") or "AI adoption in African banking"
print(f"\n🚀 Generating report on: '{topic}'\n")
result = app.invoke({"topic": topic})
print("\n" + "=" * 60)
print(result["final_report"])
# Save to file
with open(f"report_{topic[:30].replace(' ', '_')}.md", "w") as f:
f.write(result["final_report"])
print(f"\n📄 Report saved!")16.3 Lab Challenges 🏆
- Easy: Add a “conclusion” section to the report
- Medium: Add web search capability using the Tavily API as a research tool
- Hard: Add human-in-the-loop approval: pause before final assembly and allow the user to edit any section