Module 9: Human-in-the-Loop
Evolving Project: Human Approvals (v4)
Project overview
In Module 8, you built v3 of the AI Research Assistant: an agent with persistent memory, personalized greetings, preference detection, and multi-user support. If you close the terminal and open it again, the agent remembers who you are and what you researched. It's a product with an identity.
But it makes every decision on its own. If it decides to search a premium API that costs $5.00, it does it without asking. If it generates a report with an incorrect fact, it presents that fact as truth. If its research plan is inefficient — searching 6 sources when 2 would do — you have no way to tell it "no, focus on just these two." The agent is competent but unsupervised.
v4 turns the Research Assistant from an autonomous agent into a supervised one. Before searching, it shows its plan: "I plan to search these 3 sources. Estimated cost: $2.50. Shall I proceed?" Before generating the final report, it shows you a draft so you can review it and give feedback. If a fact is wrong, you can edit it directly in the agent's state. The result is an agent that keeps the human in the loop for the decisions that matter, without bothering you about the ones that don't.
The moment that defines this version: the agent shows its research plan, you remove an unnecessary source and add one it hadn't considered. The agent runs your corrected plan, generates a draft, you point out a factual error, the agent fixes it and produces the final report. That iterative conversation between human and agent — that's what effective supervision looks like.
Project goal
Evolve the AI Research Assistant from v3 (smart but autonomous) to v4 (smart and supervised) by adding plan approval, a cost gate, a feedback loop, and state editing.
By the end of this project:
- 🔧 You'll implement an interrupt in the planning phase to show the plan before executing
- 🔧 You'll add a cost gate that auto-approves free sources and interrupts for paid ones
- 🔧 You'll create a feedback loop where the user reviews the draft and gives feedback (3 rounds max)
- 🔧 You'll enable state editing so the user can correct factual data before the final report
- 🔧 You'll build an interactive CLI that handles every HITL flow naturally
Before and after
v3 (Module 8): smart but autonomous
User: "Research AI agents"
Agent: [decides the plan internally — 5 sources, $3.50]
Agent: [searches every source without asking]
Agent: [generates a report with an incorrect fact]
Agent: "Here's your report."
User: "... it spent $3.50 and the second finding is wrong"
v4 (This module): smart and supervised
User: "Research AI agents"
Agent: "Plan: search [web, arXiv, news]. Cost: $0.00. Shall I proceed?"
User: "Add academic, drop news"
Agent: [searches web, arXiv, academic — free sources, auto-approved]
Agent: "I can also search Premium Research ($2.50). Shall I proceed?"
User: "No, the free ones are enough"
Agent: "Report draft ready. Any feedback?"
User: "The second finding says 2023 but it's 2025. Fix it."
Agent: [fixes it] → "Final report with the correction."
Technical specs
| Component | Version | Purpose |
|---|---|---|
| Python | 3.11+ | Runtime |
| LangChain | v1.2+ | LLM framework |
| LangGraph | v1.0+ | Functional API + interrupt + Command |
| langchain-openai | latest | Model provider |
| pydantic | v2+ | Structured models |
Project structure
research-assistant/
├── .env
├── requirements.txt
├── agents/
│ └── researcher.py # MODIFIED — v4 with interrupts + feedback
├── tools/
│ ├── web_search.py # UNCHANGED (from v2)
│ └── calculator.py # UNCHANGED (from v1)
├── state/
│ └── research_state.py # EXTENDED — HumanFeedback model
├── config/
│ └── settings.py # EXTENDED — per-source cost config
├── memory/
│ └── user_store.py # UNCHANGED (from v3)
├── hitl/ # NEW
│ └── risk_assessment.py # Risk decision framework
├── utils/
│ ├── retry.py # UNCHANGED (from v2)
│ └── logger.py # UNCHANGED (from v2)
└── main.py # MODIFIED — CLI with HITL flows
Step 1: The risk framework (hitl/risk_assessment.py)
The module that decides which actions need human approval and which get auto-approved. It applies the 4-criteria framework from the previous capsule.
"""
hitl/risk_assessment.py
Risk decision framework for the AI Research Assistant v4.
"""
from dataclasses import dataclass
from enum import Enum
class RiskDecision(Enum):
AUTO_APPROVE = "auto_approve"
QUICK_APPROVE = "quick_approve"
FULL_REVIEW = "full_review"
@dataclass
class ActionRisk:
action_name: str
cost: float
reversible: bool
external_impact: bool
confidence: float
@property
def risk_score(self) -> float:
if self.cost < 0.01:
cost_score = 0.0
elif self.cost < 1.0:
cost_score = 0.3
elif self.cost < 10.0:
cost_score = 0.6
else:
cost_score = 1.0
rev_score = 0.0 if self.reversible else 0.8
impact_score = 0.7 if self.external_impact else 0.0
base = max(cost_score, rev_score, impact_score)
confidence_penalty = (1.0 - self.confidence) * 0.5
return min(round(base + confidence_penalty, 2), 1.0)
@property
def decision(self) -> RiskDecision:
score = self.risk_score
if score < 0.3:
return RiskDecision.AUTO_APPROVE
elif score <= 0.7:
return RiskDecision.QUICK_APPROVE
else:
return RiskDecision.FULL_REVIEW
SOURCE_COSTS = {
"web": 0.0,
"academic": 0.0,
"news": 0.0,
"premium_research": 2.50,
"patent_db": 1.00,
"financial_data": 5.00,
}
def assess_source(source_type: str, confidence: float = 0.9) -> ActionRisk:
cost = SOURCE_COSTS.get(source_type, 0.0)
return ActionRisk(
action_name=f"search_{source_type}",
cost=cost,
reversible=True,
external_impact=False,
confidence=confidence,
)
def assess_report_send(recipients: int, confidence: float = 0.8) -> ActionRisk:
return ActionRisk(
action_name="send_report",
cost=0.0,
reversible=False,
external_impact=recipients > 1,
confidence=confidence,
)
def partition_sources(sources: list[str], confidence: float = 0.9) -> dict:
"""Splits sources into auto-approved and requires-approval."""
auto = []
needs_approval = []
total_cost = 0.0
for src in sources:
risk = assess_source(src, confidence)
if risk.decision == RiskDecision.AUTO_APPROVE:
auto.append({"source": src, "cost": SOURCE_COSTS.get(src, 0.0)})
else:
needs_approval.append({"source": src, "cost": SOURCE_COSTS.get(src, 0.0), "risk": risk.risk_score})
total_cost += SOURCE_COSTS.get(src, 0.0)
return {
"auto_approved": auto,
"needs_approval": needs_approval,
"total_pending_cost": total_cost,
}Step 2: Extend the data models (state/research_state.py)
Add the HumanFeedback model to track human interactions during execution.
"""
state/research_state.py
Data models for the AI Research Assistant v4.
"""
from pydantic import BaseModel, Field
from datetime import datetime
class Source(BaseModel):
name: str
source_type: str
content: str
class KeyFinding(BaseModel):
title: str
description: str
confidence: float = Field(ge=0.0, le=1.0)
class SourceStatus(BaseModel):
source_type: str
status: str
attempts: int = Field(default=1)
error: str = Field(default="")
duration_ms: float = Field(default=0)
class UserProfile(BaseModel):
user_id: str
total_sessions: int = 0
top_topics: list[tuple[str, int]] = Field(default_factory=list)
last_topic: str = ""
preferred_format: str = "paragraphs"
is_new_user: bool = True
class HumanFeedback(BaseModel):
"""A record of one human interaction during execution (v4)."""
feedback_type: str
content: str
timestamp: str = Field(default_factory=lambda: datetime.now().isoformat())
round_number: int = 1
class ResearchReport(BaseModel):
topic: str
summary: str
key_findings: list[KeyFinding] = Field(min_length=1)
sources: list[Source] = Field(min_length=1)
sub_queries: list[str]
confidence: float = Field(ge=0.0, le=1.0)
generated_at: str = Field(default_factory=lambda: datetime.now().isoformat())
source_availability: list[SourceStatus] = Field(default_factory=list)
version: str = Field(default="v4")
user_id: str = Field(default="anonymous")
session_number: int = Field(default=0)
human_feedback_rounds: int = Field(default=0)
plan_was_edited: bool = Field(default=False)
facts_corrected: int = Field(default=0)
class SubQuery(BaseModel):
query: str
rationale: strStep 3: Cost configuration (config/settings.py)
"""
config/settings.py
Configuration for the AI Research Assistant v4.
"""
from dotenv import load_dotenv
load_dotenv()
MODEL_NAME = "openai:gpt-4.1-mini"
MODEL_TEMPERATURE = 0.2
MAX_SUB_QUERIES = 4
SEARCH_SOURCES = ["web", "academic", "news"]
RETRY_MAX_ATTEMPTS = 3
RETRY_BASE_DELAY = 1.0
RETRY_MAX_DELAY = 10.0
RETRY_JITTER = True
MIN_SOURCES_FOR_REPORT = 1
# v3: Memory
MAX_CONVERSATION_MESSAGES = 20
MEMORY_PERSIST_FILE = "memory_store.json"
DEFAULT_FORMAT = "paragraphs"
DEFAULT_DETAIL_LEVEL = "standard"
DEFAULT_SOURCES_PRIORITY = ["web", "academic", "news"]
# v4: HITL
MAX_FEEDBACK_ROUNDS = 3
PAID_SOURCES = {"premium_research": 2.50, "patent_db": 1.00, "financial_data": 5.00}
FREE_SOURCES = ["web", "academic", "news"]
AUTO_APPROVE_COST_THRESHOLD = 0.01Step 4: The v4 agent with HITL (agents/researcher.py)
The agent now has four interruption points:
- Plan approval — shows the plan and waits for approval/editing
- Cost gate — interrupts for paid sources
- Draft feedback — shows the draft and waits for feedback (up to 3 rounds)
- Fact correction — lets the user edit findings before the final report
"""
agents/researcher.py
AI Research Assistant v4 — human-in-the-loop, supervised execution.
"""
import json
import time
import uuid
from langchain.chat_models import init_chat_model
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.base import BaseStore
from langgraph.types import interrupt
import sys
sys.path.insert(0, ".")
from config.settings import (
MODEL_NAME, MODEL_TEMPERATURE, MAX_SUB_QUERIES,
MIN_SOURCES_FOR_REPORT, MAX_FEEDBACK_ROUNDS,
FREE_SOURCES, PAID_SOURCES,
)
from state.research_state import (
ResearchReport, Source, KeyFinding, SubQuery, SourceStatus, HumanFeedback,
)
from tools.web_search import search_with_retry
from tools.calculator import calculate_confidence
from utils.logger import ResearchLogger
from memory.user_store import (
get_user_profile, get_preferences, record_session, detect_preferences_from_input,
)
from hitl.risk_assessment import partition_sources, assess_source, RiskDecision
model = init_chat_model(MODEL_NAME, temperature=MODEL_TEMPERATURE)
agent_logger = ResearchLogger("research_agent_v4")
@task
def generate_greeting(user_id: str, profile: dict) -> str:
if profile["is_new_user"]:
return "Hi! I'm your research assistant. In this version, I'll check with you before any important decision."
parts = ["Welcome back!"]
if profile["last_session"]:
topic = profile["last_session"].get("topic", "")
date = profile["last_session"].get("timestamp", "")[:10]
parts.append(f"Last research: '{topic}' ({date}).")
if profile["total_sessions"] > 0:
parts.append(f"You're at {profile['total_sessions']} sessions.")
if profile["top_topics"] and profile["top_topics"][0][1] >= 2:
top = profile["top_topics"][0]
parts.append(f"Favorite topic: '{top[0]}' ({top[1]} times).")
return " ".join(parts)
@task
def decompose_query(topic: str) -> list[dict]:
response = model.invoke(
f"Break this topic into {MAX_SUB_QUERIES} researchable sub-questions.\n\n"
f"Topic: {topic}\n\n"
f'Answer in JSON: [{{"query": "...", "rationale": "..."}}]\nJSON only.'
)
try:
return json.loads(response.content)[:MAX_SUB_QUERIES]
except json.JSONDecodeError:
return [{"query": topic, "rationale": "Fallback"}, {"query": f"advances in {topic}", "rationale": "Trends"}]
@task
def plan_research(topic: str, sub_queries: list[dict], available_sources: list[str]) -> dict:
"""Builds the research plan and asks for human approval."""
partition = partition_sources(available_sources)
plan = {
"topic": topic,
"sub_queries": [sq["query"] for sq in sub_queries],
"free_sources": [s["source"] for s in partition["auto_approved"]],
"paid_sources": [s for s in partition["needs_approval"]],
"total_free_cost": 0.0,
"total_paid_cost": partition["total_pending_cost"],
}
response = interrupt({
"type": "plan_approval",
"message": (
f"Research plan for '{topic}':\n"
f" Sub-queries: {len(sub_queries)}\n"
f" Free sources: {plan['free_sources']}\n"
f" Paid sources: {[s['source'] for s in plan['paid_sources']]} "
f"(${plan['total_paid_cost']:.2f})\n"
f"Shall I proceed? Options: approve / edit / cancel"
),
"plan": plan,
})
action = response.get("action", "approve")
if action == "cancel":
return {**plan, "status": "cancelled"}
elif action == "edit":
edited_sources = response.get("sources", plan["free_sources"])
plan["free_sources"] = [s for s in edited_sources if s in FREE_SOURCES]
plan["paid_sources"] = [
{"source": s, "cost": PAID_SOURCES.get(s, 0.0)}
for s in edited_sources if s in PAID_SOURCES
]
plan["total_paid_cost"] = sum(s["cost"] for s in plan["paid_sources"])
plan["was_edited"] = True
return {**plan, "status": "approved_edited"}
else:
return {**plan, "status": "approved"}
@task
def approve_paid_sources(paid_sources: list[dict]) -> list[str]:
"""Cost gate: asks for individual approval on each paid source."""
approved = []
for source_info in paid_sources:
src = source_info["source"]
cost = source_info["cost"]
response = interrupt({
"type": "cost_approval",
"message": f"Searching '{src}' costs ${cost:.2f}. Approve?",
"source": src,
"cost": cost,
})
if response.get("approved", False):
approved.append(src)
print(f" ✅ {src} (${cost:.2f}) — approved")
else:
print(f" ❌ {src} (${cost:.2f}) — rejected")
return approved
@task
def search_all_sources(query: str, sources: list[str], logger: ResearchLogger) -> list[dict]:
start = time.time()
futures = [search_with_retry(query, src, logger) for src in sources]
results = [f.result() for f in futures]
return results
@task
def merge_and_deduplicate(all_results: list[dict]) -> list[dict]:
seen, unique = set(), []
for r in all_results:
if r["search_status"] != "ok":
continue
key = f"{r['source_type']}:{r['content'][:100]}"
if key not in seen:
seen.add(key)
unique.append(r)
unique.sort(key=lambda r: r.get("relevance", 0), reverse=True)
return unique
@task
def synthesize_findings(topic: str, results: list[dict]) -> list[dict]:
text = "\n".join(f"Source {i} ({r['source_type']}): {r['content']}" for i, r in enumerate(results, 1))
response = model.invoke(
f"Identify 3-5 key findings about '{topic}'.\n\nSources:\n{text}\n\n"
f'JSON: [{{"title": "...", "description": "...", "confidence": 0.8}}]\nJSON only.'
)
try:
return json.loads(response.content)[:5]
except json.JSONDecodeError:
return [{"title": "General finding", "description": f"Relevant research on {topic}.", "confidence": 0.6}]
@task
def generate_summary(topic: str, findings: list[dict], source_info: str, user_prefs: dict) -> str:
findings_text = "\n".join(f"- {f['title']}: {f['description']}" for f in findings)
fmt = ""
if user_prefs.get("format") == "bullet_points":
fmt = "Use bullet points. "
if user_prefs.get("detail_level") == "brief":
fmt += "Be brief (1-2 sentences). "
elif user_prefs.get("detail_level") == "detailed":
fmt += "Be detailed (4-5 sentences). "
response = model.invoke(
f"Executive summary about '{topic}'.\nFindings:\n{findings_text}\n"
f"Sources: {source_info}\n{fmt}The summary only."
)
return response.content.strip()
@task
def collect_draft_feedback(topic: str, summary: str, findings: list[dict], round_num: int) -> dict:
"""Shows the draft and collects the user's feedback."""
findings_display = "\n".join(
f" {i}. {f['title']} [{f['confidence']:.0%}]: {f['description']}"
for i, f in enumerate(findings, 1)
)
response = interrupt({
"type": "draft_feedback",
"message": (
f"--- Report draft (round {round_num}/{MAX_FEEDBACK_ROUNDS}) ---\n\n"
f"Topic: {topic}\n\n"
f"Summary:\n{summary}\n\n"
f"Findings:\n{findings_display}\n\n"
f"Any feedback? Options: approve / feedback <text> / edit_facts"
),
"summary": summary,
"findings": findings,
"round": round_num,
})
return response
@task
def apply_feedback_to_summary(topic: str, current_summary: str, feedback: str, findings: list[dict], user_prefs: dict) -> str:
"""Regenerates the summary incorporating the user's feedback."""
findings_text = "\n".join(f"- {f['title']}: {f['description']}" for f in findings)
response = model.invoke(
f"Rewrite this summary about '{topic}' incorporating the user's feedback.\n\n"
f"Current summary:\n{current_summary}\n\n"
f"Findings:\n{findings_text}\n\n"
f"User feedback: {feedback}\n\n"
f"Produce the improved summary only."
)
return response.content.strip()
@task
def collect_fact_corrections(findings: list[dict]) -> list[dict]:
"""Lets the user correct factual findings."""
response = interrupt({
"type": "fact_correction",
"message": (
"You can correct the findings. Send a list of corrections:\n"
' {"corrections": [{"index": 0, "field": "description", "new_value": "..."}]}\n'
"Or send {\"corrections\": []} if everything is correct."
),
"findings": findings,
})
corrections = response.get("corrections", [])
corrected = [f.copy() for f in findings]
for correction in corrections:
idx = correction.get("index", -1)
field = correction.get("field", "")
new_value = correction.get("new_value", "")
if 0 <= idx < len(corrected) and field in corrected[idx]:
corrected[idx][field] = new_value
return corrected
def create_research_agent(checkpointer, store):
"""Factory that builds the v4 research agent."""
@entrypoint(checkpointer=checkpointer, store=store)
def research_agent(topic: str, *, store: BaseStore) -> dict:
config = entrypoint.get_config()
user_id = config["configurable"].get("user_id", "anonymous")
request_id = uuid.uuid4().hex[:8]
agent_logger.set_request_id(request_id)
pipeline_start = time.time()
profile = get_user_profile(store, user_id)
user_prefs = get_preferences(store, user_id)
greeting = generate_greeting(user_id, profile).result()
print(f"\n{'=' * 60}")
print(f" 🔬 AI Research Assistant v4 — Supervised")
print(f" {greeting}")
print(f" Topic: {topic} | User: {user_id} | Request: {request_id}")
print(f"{'=' * 60}")
detect_preferences_from_input(store, user_id, topic)
# --- PHASE 1: Decompose ---
print(f"\n📋 Decomposing the topic...")
sub_queries_raw = decompose_query(topic).result()
sub_queries = [SubQuery(**sq) for sq in sub_queries_raw]
for i, sq in enumerate(sub_queries, 1):
print(f" {i}. {sq.query}")
# --- PHASE 2: Plan approval (INTERRUPT) ---
all_available = list(FREE_SOURCES) + list(PAID_SOURCES.keys())
plan = plan_research(topic, sub_queries_raw, all_available).result()
if plan.get("status") == "cancelled":
print(f"\n ❌ Research cancelled by the user.")
return {"error": "Research cancelled", "version": "v4"}
was_edited = plan.get("was_edited", False)
active_sources = plan["free_sources"]
print(f"\n Plan {'edited' if was_edited else 'approved'}: {active_sources}")
# --- PHASE 3: Cost gate for paid sources (INTERRUPT per source) ---
if plan["paid_sources"]:
print(f"\n💰 Evaluating paid sources...")
approved_paid = approve_paid_sources(plan["paid_sources"]).result()
active_sources = active_sources + approved_paid
else:
print(f"\n No paid sources in the plan.")
# --- PHASE 4: Search ---
print(f"\n🔍 Searching {len(active_sources)} sources: {active_sources}")
all_raw, source_statuses = [], []
search_futures = [search_all_sources(sq.query, active_sources, agent_logger) for sq in sub_queries]
for i, future in enumerate(search_futures):
results = future.result()
all_raw.extend(results)
ok = sum(1 for r in results if r["search_status"] == "ok")
print(f" Sub-query {i + 1}: {ok}/{len(results)} OK")
for r in results:
source_statuses.append(SourceStatus(
source_type=r["source_type"], status=r["search_status"],
attempts=r.get("attempts", 1), error=r.get("error", ""),
duration_ms=r.get("duration_ms", 0),
))
total_ok = sum(1 for r in all_raw if r["search_status"] == "ok")
total_failed = len(all_raw) - total_ok
if total_ok < MIN_SOURCES_FOR_REPORT:
print(f"\n ❌ Not enough sources ({total_ok}). Aborting.")
return {"error": f"Only {total_ok} sources. Minimum: {MIN_SOURCES_FOR_REPORT}", "version": "v4"}
# --- PHASE 5: Merge + Synthesize ---
unique = merge_and_deduplicate(all_raw).result()
print(f"\n🔀 {len(all_raw)} raw → {len(unique)} unique")
findings_raw = synthesize_findings(topic, unique).result()
findings = [KeyFinding(**f) for f in findings_raw]
print(f"🧠 {len(findings)} findings identified")
source_info = f"{total_ok}/{total_ok + total_failed} sources OK"
summary = generate_summary(topic, findings_raw, source_info, user_prefs).result()
# --- PHASE 6: Draft feedback loop (INTERRUPT — max 3 rounds) ---
feedback_rounds = 0
human_feedbacks = []
for round_num in range(1, MAX_FEEDBACK_ROUNDS + 1):
print(f"\n📝 Showing the draft (round {round_num})...")
feedback_response = collect_draft_feedback(topic, summary, findings_raw, round_num).result()
action = feedback_response.get("action", "approve")
if action == "approve":
print(f" ✅ Draft approved on round {round_num}.")
break
elif action == "feedback":
feedback_text = feedback_response.get("text", "")
human_feedbacks.append(HumanFeedback(
feedback_type="draft_feedback",
content=feedback_text,
round_number=round_num,
))
print(f" 📝 Feedback received: '{feedback_text[:60]}...'")
summary = apply_feedback_to_summary(topic, summary, feedback_text, findings_raw, user_prefs).result()
feedback_rounds += 1
print(f" 🔄 Summary updated.")
elif action == "edit_facts":
print(f"\n ✏️ Fact editing mode...")
corrected_findings_raw = collect_fact_corrections(findings_raw).result()
corrections_count = sum(
1 for orig, corr in zip(findings_raw, corrected_findings_raw)
if orig != corr
)
if corrections_count > 0:
findings_raw = corrected_findings_raw
findings = [KeyFinding(**f) for f in findings_raw]
summary = generate_summary(topic, findings_raw, source_info, user_prefs).result()
human_feedbacks.append(HumanFeedback(
feedback_type="fact_correction",
content=f"{corrections_count} facts corrected",
round_number=round_num,
))
feedback_rounds += 1
print(f" ✅ {corrections_count} facts corrected. Summary regenerated.")
else:
print(f" No corrections. Moving on.")
else:
print(f"\n ⚠️ Max rounds reached ({MAX_FEEDBACK_ROUNDS}). Using the latest version.")
# --- PHASE 7: Build final report ---
avg_rel = sum(r["relevance"] for r in unique) / len(unique) if unique else 0.5
base_conf = calculate_confidence(len(unique), avg_rel, len(findings)).result()
avail_factor = total_ok / (total_ok + total_failed) if (total_ok + total_failed) > 0 else 0.5
confidence = round(base_conf * (0.7 + 0.3 * avail_factor), 2)
sources = [Source(name=r["source_name"], source_type=r["source_type"], content=r["content"]) for r in unique]
facts_corrected = sum(1 for fb in human_feedbacks if fb.feedback_type == "fact_correction")
report = ResearchReport(
topic=topic, summary=summary, key_findings=findings, sources=sources,
sub_queries=[sq.query for sq in sub_queries], confidence=confidence,
source_availability=source_statuses, version="v4",
user_id=user_id, session_number=profile["total_sessions"] + 1,
human_feedback_rounds=feedback_rounds,
plan_was_edited=was_edited,
facts_corrected=facts_corrected,
)
record_session(store, user_id, topic, summary[:200])
pipeline_ms = (time.time() - pipeline_start) * 1000
print(f"\n📄 v4 report generated. Session #{report.session_number}.")
print(f" Feedback rounds: {feedback_rounds} | Facts corrected: {facts_corrected}")
print(f" Plan edited: {'yes' if was_edited else 'no'} | Time: {pipeline_ms:.0f}ms")
print(f"{'=' * 60}")
return report.model_dump()
return research_agentStep 5: An interactive CLI with HITL flows (main.py)
The CLI runs the whole conversation: it receives the agent's interrupts, presents them to the user in readable form, collects the answer, and resumes execution.
"""
main.py
CLI for the AI Research Assistant v4. Interactive human-in-the-loop.
"""
import sys
import json
import uuid
sys.path.insert(0, ".")
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command
from memory.user_store import create_store, persist_store, get_user_profile, get_preferences, save_preference
from agents.researcher import create_research_agent
from config.settings import FREE_SOURCES, PAID_SOURCES
def handle_interrupt(state_snapshot) -> Command:
"""Processes an interrupt and returns the Command to respond with."""
task_data = state_snapshot.tasks
if not task_data:
return Command(resume={"action": "approve"})
for task_item in task_data:
if hasattr(task_item, "interrupts") and task_item.interrupts:
interrupt_data = task_item.interrupts[0].value
interrupt_type = interrupt_data.get("type", "unknown")
print(f"\n{'─' * 58}")
print(f" 🔔 INTERRUPT: {interrupt_type}")
print(f"{'─' * 58}")
print(f" {interrupt_data.get('message', 'No message')}")
print(f"{'─' * 58}")
if interrupt_type == "plan_approval":
return handle_plan_approval(interrupt_data)
elif interrupt_type == "cost_approval":
return handle_cost_approval(interrupt_data)
elif interrupt_type == "draft_feedback":
return handle_draft_feedback(interrupt_data)
elif interrupt_type == "fact_correction":
return handle_fact_correction(interrupt_data)
else:
return Command(resume={"action": "approve"})
return Command(resume={"action": "approve"})
def handle_plan_approval(data: dict) -> Command:
"""Handles approval of the research plan."""
plan = data.get("plan", {})
print(f"\n Free sources: {plan.get('free_sources', [])}")
paid = plan.get("paid_sources", [])
if paid:
print(f" Paid sources: {[s['source'] for s in paid]} (${plan.get('total_paid_cost', 0):.2f})")
while True:
try:
choice = input("\n [approve/edit/cancel] → ").strip().lower()
except (KeyboardInterrupt, EOFError):
return Command(resume={"action": "cancel"})
if choice in ("approve", "a", "ok", "okay", "yes", "y"):
return Command(resume={"action": "approve"})
elif choice in ("cancel", "c", "no", "n"):
return Command(resume={"action": "cancel"})
elif choice.startswith("edit") or choice == "e":
all_sources = list(FREE_SOURCES) + list(PAID_SOURCES.keys())
print(f" Available sources: {all_sources}")
try:
sources_input = input(" Type the sources separated by commas: ").strip()
except (KeyboardInterrupt, EOFError):
return Command(resume={"action": "cancel"})
edited = [s.strip() for s in sources_input.split(",") if s.strip()]
return Command(resume={"action": "edit", "sources": edited})
else:
print(f" Unrecognized option: '{choice}'")
def handle_cost_approval(data: dict) -> Command:
"""Handles approval of one paid source."""
source = data.get("source", "?")
cost = data.get("cost", 0)
print(f"\n Source: {source} | Cost: ${cost:.2f}")
while True:
try:
choice = input(f" Approve ${cost:.2f}? [yes/no] → ").strip().lower()
except (KeyboardInterrupt, EOFError):
return Command(resume={"approved": False})
if choice in ("yes", "yeah", "yep", "y", "ok"):
return Command(resume={"approved": True})
elif choice in ("no", "n"):
return Command(resume={"approved": False})
else:
print(f" Answer 'yes' or 'no'.")
def handle_draft_feedback(data: dict) -> Command:
"""Handles feedback on the report draft."""
round_num = data.get("round", 1)
print(f"\n Round {round_num} — options:")
print(f" approve → Approve the current draft")
print(f" feedback <tx> → Give feedback to improve it")
print(f" edit_facts → Correct factual data")
while True:
try:
choice = input(f"\n → ").strip()
except (KeyboardInterrupt, EOFError):
return Command(resume={"action": "approve"})
if choice.lower() in ("approve", "a", "ok"):
return Command(resume={"action": "approve"})
elif choice.lower().startswith("feedback ") or choice.lower().startswith("fb "):
text = choice.split(" ", 1)[1] if " " in choice else ""
if not text:
try:
text = input(" Your feedback: ").strip()
except (KeyboardInterrupt, EOFError):
return Command(resume={"action": "approve"})
return Command(resume={"action": "feedback", "text": text})
elif choice.lower() in ("edit_facts", "edit", "ef"):
return Command(resume={"action": "edit_facts"})
else:
print(f" Unrecognized option. Use: approve, feedback <text>, edit_facts")
def handle_fact_correction(data: dict) -> Command:
"""Handles fact corrections."""
findings = data.get("findings", [])
print(f"\n Current findings:")
for i, f in enumerate(findings):
print(f" [{i}] {f.get('title', '?')}: {f.get('description', '?')[:70]}...")
print(f"\n To correct one, type: <index> <new text>")
print(f" Type 'done' when you're finished.")
corrections = []
while True:
try:
line = input(" correction → ").strip()
except (KeyboardInterrupt, EOFError):
break
if line.lower() in ("done", "d", "finish", ""):
break
parts = line.split(" ", 1)
if len(parts) == 2 and parts[0].isdigit():
idx = int(parts[0])
if 0 <= idx < len(findings):
corrections.append({"index": idx, "field": "description", "new_value": parts[1]})
print(f" ✅ Finding [{idx}] marked for correction.")
else:
print(f" ⚠️ Index {idx} out of range (0-{len(findings)-1})")
else:
print(f" Format: <index> <new text>")
return Command(resume={"corrections": corrections})
def format_report(report: dict) -> str:
if "error" in report:
return f"\n ❌ {report['error']}"
lines = [
"", "╔" + "═" * 58 + "╗",
"║" + f" 📄 v4 REPORT | {report.get('user_id')} | Session #{report.get('session_number')}".center(58) + "║",
"╚" + "═" * 58 + "╝",
f"\n📌 Topic: {report['topic']}",
f"🎯 Confidence: {report['confidence']:.0%}",
]
hitl_info = []
if report.get("plan_was_edited"):
hitl_info.append("Plan edited by the user")
if report.get("human_feedback_rounds", 0) > 0:
hitl_info.append(f"{report['human_feedback_rounds']} feedback rounds")
if report.get("facts_corrected", 0) > 0:
hitl_info.append(f"{report['facts_corrected']} facts corrected")
if hitl_info:
lines.append(f"👤 Supervision: {' | '.join(hitl_info)}")
lines.extend([f"\n{'─' * 60}", "📋 SUMMARY", f"{'─' * 60}", report["summary"]])
lines.extend([f"\n{'─' * 60}", "💡 FINDINGS", f"{'─' * 60}"])
for i, f in enumerate(report["key_findings"], 1):
lines.append(f" {i}. {f['title']} [{f['confidence']:.0%}]")
lines.append(f" {f['description']}")
lines.extend([f"\n{'─' * 60}", f"📚 SOURCES ({len(report['sources'])})", f"{'─' * 60}"])
for s in report["sources"]:
lines.append(f" • [{s['source_type'].upper()}] {s['name']}")
if report.get("source_availability"):
lines.extend([f"\n{'─' * 60}", "🔌 AVAILABILITY", f"{'─' * 60}"])
for sa in report["source_availability"]:
icon = "✅" if sa["status"] == "ok" else "❌"
retry = f" ({sa['attempts']} attempts)" if sa["attempts"] > 1 else ""
lines.append(f" {icon} {sa['source_type']}: {sa['status']}{retry}")
lines.append(f"\n{'═' * 60}")
return "\n".join(lines)
def run_research_with_hitl(agent, topic: str, user_id: str, store):
"""Runs a full research pass, handling every interrupt along the way."""
thread_id = f"v4-{user_id}-{uuid.uuid4().hex[:8]}"
config = {"configurable": {"thread_id": thread_id, "user_id": user_id}}
result = agent.invoke(topic, config)
while True:
state = agent.get_state(config)
if not state.next:
break
command = handle_interrupt(state)
result = agent.invoke(command, config)
return result
def run_interactive():
print("=" * 60)
print(" 🔬 AI Research Assistant v4 — Human-in-the-Loop")
print("=" * 60)
store = create_store()
checkpointer = MemorySaver()
agent = create_research_agent(checkpointer, store)
current_user = None
print("\n Commands: user <name> | profile | pref <k> <v> | exit\n")
while True:
if not current_user:
try:
current_user = input("👤 User: ").strip()
except (KeyboardInterrupt, EOFError):
break
if not current_user:
continue
profile = get_user_profile(store, current_user)
if profile["is_new_user"]:
print(f" Welcome, {current_user}! First time here.\n")
else:
print(f" Welcome back, {current_user}! ({profile['total_sessions']} sessions)")
if profile["last_session"]:
print(f" Last: '{profile['last_session']['topic']}' ({profile['last_session']['timestamp'][:10]})\n")
continue
try:
user_input = input(f"🔎 [{current_user}] ").strip()
except (KeyboardInterrupt, EOFError):
break
if not user_input:
continue
if user_input.lower() in ("exit", "quit", "bye"):
break
if user_input.lower().startswith("user "):
current_user = user_input[5:].strip()
profile = get_user_profile(store, current_user)
status = "new" if profile["is_new_user"] else f"{profile['total_sessions']} sessions"
print(f" Switched to {current_user} ({status})\n")
continue
if user_input.lower() == "profile":
profile = get_user_profile(store, current_user)
prefs = get_preferences(store, current_user)
print(f"\n Sessions: {profile['total_sessions']}")
for t, c in profile["top_topics"]:
print(f" - {t} ({c}x)")
print(f" Prefs: {json.dumps(prefs, ensure_ascii=False)}\n")
continue
if user_input.lower().startswith("pref "):
parts = user_input.split(maxsplit=2)
if len(parts) == 3:
save_preference(store, current_user, parts[1], parts[2], "Manual")
print(f" ✓ {parts[1]} = {parts[2]}\n")
continue
try:
report = run_research_with_hitl(agent, user_input, current_user, store)
print(format_report(report))
except Exception as e:
print(f"\n ❌ Error: {e}\n")
print("\n Saving memory...")
persist_store(store)
print(" See you next time!")
if __name__ == "__main__":
run_interactive()Running it: the full supervised experience
A typical session with all 4 HITL types
cd research-assistant
python main.py============================================================
🔬 AI Research Assistant v4 — Human-in-the-Loop
============================================================
[Memory] Loaded 5 items from memory_store.json
👤 User: mike
Welcome back, mike! (3 sessions)
Last: 'RAG techniques' (2026-03-07)
🔎 [mike] AI agents frameworks comparison
============================================================
🔬 AI Research Assistant v4 — Supervised
Welcome back! Last research: 'RAG techniques' (2026-03-07). You're at 3 sessions.
Topic: AI agents frameworks comparison | User: mike | Request: a1b2c3d4
============================================================
📋 Decomposing the topic...
1. What are the main AI agent frameworks available?
2. How do LangGraph, CrewAI, and AutoGen compare?
3. What are the production considerations for each?
4. What are real-world use cases for each framework?
──────────────────────────────────────────────────────
🔔 INTERRUPT: plan_approval
──────────────────────────────────────────────────────
Research plan for 'AI agents frameworks comparison':
Sub-queries: 4
Free sources: [web, academic, news]
Paid sources: [premium_research] ($2.50)
Shall I proceed? Options: approve / edit / cancel
──────────────────────────────────────────────────────
Free sources: ['web', 'academic', 'news']
Paid sources: ['premium_research'] ($2.50)
[approve/edit/cancel] → edit
Available sources: ['web', 'academic', 'news', 'premium_research', 'patent_db', 'financial_data']
Type the sources separated by commas: web, academic
Plan edited: ['web', 'academic']
No paid sources in the plan.
🔍 Searching 2 sources: ['web', 'academic']
Sub-query 1: 2/2 OK
Sub-query 2: 2/2 OK
Sub-query 3: 1/2 OK
Sub-query 4: 2/2 OK
🔀 16 raw → 7 unique
🧠 4 findings identified
📝 Showing the draft (round 1)...
──────────────────────────────────────────────────────
🔔 INTERRUPT: draft_feedback
──────────────────────────────────────────────────────
--- Report draft (round 1/3) ---
Topic: AI agents frameworks comparison
Summary:
The main frameworks for AI agents include LangGraph,
CrewAI, and AutoGen. LangGraph stands out for its granular control...
Findings:
1. LangGraph offers granular control [85%]: It lets you define...
2. CrewAI simplifies multi-agent [80%]: A role-oriented framework...
3. AutoGen leads on conversation [75%]: Designed by Microsoft...
4. Production favors LangGraph [82%]: In production environments...
Any feedback? Options: approve / feedback <text> / edit_facts
──────────────────────────────────────────────────────
Round 1 — options:
approve → Approve the current draft
feedback <tx> → Give feedback to improve it
edit_facts → Correct factual data
→ feedback Add a more direct comparison between the three. Mention that CrewAI is newer and has a smaller community.
📝 Feedback received: 'Add a more direct comparison between the three...'
🔄 Summary updated.
📝 Showing the draft (round 2)...
──────────────────────────────────────────────────────
🔔 INTERRUPT: draft_feedback
──────────────────────────────────────────────────────
(improved draft with the direct comparison)
→ edit_facts
✏️ Fact editing mode...
──────────────────────────────────────────────────────
🔔 INTERRUPT: fact_correction
──────────────────────────────────────────────────────
Current findings:
[0] LangGraph offers granular control: It lets you define flows...
[1] CrewAI simplifies multi-agent: A role-oriented framework...
[2] AutoGen leads on conversation: Designed by Microsoft in 2023...
[3] Production favors LangGraph: In production environments...
To correct one, type: <index> <new text>
Type 'done' when you're finished.
correction → 2 Designed by Microsoft Research, with version 0.4 released in 2025. It supports conversational agent patterns and extensible tools.
✅ Finding [2] marked for correction.
correction → done
✅ 1 facts corrected. Summary regenerated.
📝 Showing the draft (round 3)...
→ approve
✅ Draft approved on round 3.
📄 v4 report generated. Session #4.
Feedback rounds: 2 | Facts corrected: 1
Plan edited: yes | Time: 45230ms
============================================================
╔══════════════════════════════════════════════════════════╗
║ 📄 v4 REPORT | mike | Session #4 ║
╚══════════════════════════════════════════════════════════╝
📌 Topic: AI agents frameworks comparison
🎯 Confidence: 78%
👤 Supervision: Plan edited by the user | 2 feedback rounds | 1 facts corrected
...
The full session hits all 4 HITL types:
- ✅ Plan approval → the user edited the sources (dropped
news, droppedpremium_research) - ✅ Cost gate → never fired, because the edited plan has no paid sources
- ✅ Draft feedback → the user gave textual feedback that improved the summary
- ✅ Fact correction → the user fixed an out-of-date fact about AutoGen
Success criteria
- ✅ Plan approval works — the agent shows the plan, and the user can approve, edit the sources, or cancel
- ✅ The cost gate works — free sources auto-approve, paid sources interrupt with a cost estimate
- ✅ The feedback loop improves quality — the user gives textual feedback, the agent regenerates the summary incorporating it
- ✅ State editing fixes factual errors — the user edits findings directly, and the summary is regenerated with the corrected data
- ✅ The CLI handles every flow — the interaction feels natural, with clear prompts and understandable options
- ✅ Supervision metadata lands in the report — the report includes
human_feedback_rounds,plan_was_edited,facts_corrected - ✅ Memory persists across sessions — the v3 data (preferences, history) still works
Test scenarios
Test 1: Plan approved with no changes
🔎 [mike] quantum computing trends
→ Plan: [web, academic, news] + [premium_research $2.50]
→ approve
→ Cost gate: premium_research $2.50 → no
→ Draft → approve
Result: a report with 3 free sources, no feedback rounds, plan not edited.
Test 2: Plan edited + paid source approved
🔎 [mike] patent analysis for AI chips
→ Plan: [web, academic, news] + [premium_research, patent_db]
→ edit → web, academic, patent_db
→ Cost gate: patent_db $1.00 → yes
→ Draft → approve
Result: a report with academic + web + patent_db. Plan edited. Cost: $1.00.
Test 3: Research cancelled
🔎 [mike] something I changed my mind about
→ Plan: [web, academic, news]
→ cancel
Result: {"error": "Research cancelled", "version": "v4"}
Test 4: A full feedback loop (3 rounds)
🔎 [mike] deep learning optimization techniques
→ approve plan
→ Draft round 1 → feedback "More focus on hardware-aware optimization"
→ Draft round 2 → feedback "Add a mention of quantization"
→ Draft round 3 → approve
Result: a report improved by 2 feedback rounds. Approved on the 3rd round.
Test 5: Factual correction
🔎 [mike] LLM benchmarks 2025
→ approve plan
→ Draft → edit_facts
→ Finding [1] says "GPT-4 leads on MMLU" → correct it to "Claude 3.5 Sonnet has led on MMLU since Oct 2024"
→ done → approve
Result: a report with 1 corrected fact reflected in the final summary.
Test 6: Multiple paid sources — partial approval
🔎 [mike] competitive analysis SaaS pricing
→ Plan includes [premium_research $2.50, financial_data $5.00]
→ approve plan
→ Cost gate: premium_research $2.50 → yes
→ Cost gate: financial_data $5.00 → no
Result: the report uses premium_research but not financial_data. Cost: $2.50 instead of $7.50.
Common errors
1. interrupt() doesn't pause — the agent runs straight through
Cause: The graph has no checkpointer configured. Without a checkpointer, interrupt() can't save the state and simply doesn't work.
Fix: Check that @entrypoint(checkpointer=checkpointer) is configured and that checkpointer isn't None:
checkpointer = MemorySaver()
agent = create_research_agent(checkpointer, store)2. Command(resume=...) doesn't resume — the agent starts from scratch
Cause: The thread_id in the resume config doesn't match the one from the original invocation.
Fix: Use exactly the same config to resume:
config = {"configurable": {"thread_id": "v4-mike-abc123", "user_id": "mike"}}
result = agent.invoke(topic, config)
# ... get the interrupt ...
result = agent.invoke(Command(resume=response), config) # same config3. The feedback loop never ends — infinite cycle
Cause: The loop's exit condition doesn't cover every case, or the action in the response doesn't match the expected values.
Fix: Always include an else or a default break, and cap the rounds:
for round_num in range(1, MAX_FEEDBACK_ROUNDS + 1):
response = collect_feedback(...).result()
if response.get("action") == "approve":
break
# ... handle feedback ...
else:
pass # max rounds reached, use the latest version4. Fact corrections don't show up in the summary
Cause: After correcting the findings, you never regenerate the summary with the new data.
Fix: Always regenerate the summary after editing findings:
corrected = collect_fact_corrections(findings_raw).result()
if corrections_count > 0:
findings_raw = corrected
summary = generate_summary(topic, findings_raw, ...).result() # regenerate5. state.tasks is empty when you try to read the interrupt
Cause: You're reading the state before the agent has reached the interrupt point, or the interrupt was already resolved.
Fix: Read the state only when state.next says the agent is paused:
state = agent.get_state(config)
if state.next: # a node is pending → there's an interrupt
for task in state.tasks:
if hasattr(task, "interrupts") and task.interrupts:
data = task.interrupts[0].value6. The CLI can't tell interrupt types apart
Cause: Every interrupt gets handled the same way, without inspecting the type in the payload.
Fix: Each interrupt sends a type in its payload. Use a dispatcher:
interrupt_type = data.get("type", "unknown")
if interrupt_type == "plan_approval":
return handle_plan_approval(data)
elif interrupt_type == "cost_approval":
return handle_cost_approval(data)7. The v3 memory stops working after adding HITL
Cause: The store or the checkpointer isn't passed correctly to the create_research_agent() factory.
Fix: Check that both are created and passed in:
store = create_store()
checkpointer = MemorySaver()
agent = create_research_agent(checkpointer, store) # both arguments8. The paid-source costs don't match reality
Cause: SOURCE_COSTS in risk_assessment.py doesn't reflect the APIs' current prices.
Fix: Keep a single source of truth for the costs:
# config/settings.py — the source of truth
PAID_SOURCES = {"premium_research": 2.50, "patent_db": 1.00}
# hitl/risk_assessment.py — imports from config
from config.settings import PAID_SOURCES
SOURCE_COSTS = {**{s: 0.0 for s in FREE_SOURCES}, **PAID_SOURCES}What's coming: Module 10 — Multi-Agent Systems
Your Research Assistant v4 is a supervised agent: it shows plans, asks approval for expensive actions, accepts iterative feedback, and lets you correct factual data. It's the complete human-in-the-loop experience with a single agent.
But it's a single agent doing everything: planning, searching, synthesizing, writing. What happens when the task is too big or too specialized for one? What happens when you need one agent to search, another to analyze, and a third to write — each with its own tools and model?
Module 10 scales from one supervised agent to multiple coordinated agents. One plans, another runs searches, another synthesizes. The human supervises the "manager" that coordinates the rest. The same HITL techniques you learned — approval, feedback, state editing — now apply at the team level: which agent needs supervision? Which one can run autonomously? How do you coordinate approvals when 5 agents are working in parallel?
Project resources
- LangGraph Human-in-the-Loop — Official HITL concepts
- interrupt() API Reference — Reference for the interrupt function
- LangGraph Command — Using Command to resume
- How to edit graph state — Editing state during execution
- LangGraph Functional API + HITL — Interrupt inside @entrypoint and @task
- How to review tool calls — The tool review pattern
Module 9 — LangChain & LangGraph: From Chains to Agents