Creative Feedback Intelligence (CFI)
A LangGraph and FastMCP pipeline that detects opposing stakeholder revisions across emails, Looms, and voice notes, pausing for human resolution before compiling editor briefs.
The Problem: Contradictory Client Feedback
Client revision rounds in creative production are notorious for disorganized stakeholder alignment. Feedback never arrives as a single coherent document.
Instead, the agency receives an email from the Brand Lead stating: “The closing logo is too small—make it 25% larger.” Simultaneously, the Client Creative Director sends a 4-minute Loom video commenting at 0:52: “The outro feels cluttered, please keep the logo subtle and minimal.” Meanwhile, the client CEO forwards an audio memo over WhatsApp requesting a change to the background music track.
When editors attempt to address these revisions, they spend 45 to 90 minutes cross-referencing conflicting timecodes, scheduling clarification calls, or worse—implementing one stakeholder’s request only to be scolded by another in the next review meeting.
Detecting Contradictions and Pausing for Review
When two stakeholders give mutually opposing instructions for the same timecode window, any autonomous AI resolution will be wrong for at least one of them. Picking a compromise autonomously is unacceptable in commercial client relationships.
The core design decision: the system must detect the contradiction, isolate both stakeholder positions with exact timestamp attribution, and halt execution completely.
When stakeholders contradict each other, the pipeline stops and surfaces both positions. The producer decides which direction to take before any brief is generated.
Stage 06 invokes LangGraph’s native interrupt() pattern to suspend execution. Stage 07 (generate_revision_brief) enforces a hard runtime guard: it asserts the human review signature, raising an explicit RuntimeError if invoked without verified producer sign-off.
How the Pipeline Works
CFI aggregates feedback across email threads, Loom transcripts, and WhatsApp audio notes; transcribes voice memos via Groq Whisper Large v3 in under 800 milliseconds; normalizes all remarks to exact sequence timecodes; cross-references ground-truth project specs via 5 read-only FastMCP tools; detects semantic contradictions; and pauses for human resolution before compiling an editor-ready checklist.
Editorial clarification calls are eliminated, intended to reduce manual alignment time during revision rounds and ensuring editors never implement conflicting notes.
{ "project": "Maison_Living_Commercial_v3", "feedback_sources_ingested": 3, "contradictions_detected": 1, "conflict_timecode": "00:54 - 00:58", "conflict_parties": ["Sarah (Brand Lead)", "Marcus (Creative Dir)"], "human_checkpoint_cleared": true, "producer_resolution": "Approved Marcus minimal lockup with white outline", "action_items_count": 7 }
7-Stage LangGraph Feedback Synthesis Graph
Multi-channel input streams converge into a typed state graph. When semantic contradictions arise, execution pauses via native LangGraph checkpointing until producer clearance. Click any stage card to inspect its implementation, input state, and output state.
Ingest Multi-Channel ($0.00) → Groq Whisper (<800ms) → Timecode Mapping ($0.00) → FastMCP Context ($0.00) → Conflict Check (~$0.01) → Human Checkpoint (interrupt()) → Action Brief ($0.00)
Stage 01 — Multi-Channel Ingest
Ingests messy, unstandardized client feedback across multiple agency communication channels: raw client email threads, Loom video links, PDF annotation markups, and WhatsApp audio memo attachments.
Key Design Decisions
FastMCP stdio Server for Read-Only Ground-Truth State
When an AI agent analyzes client requests like “replace with the approved product render”, it needs ground-truth context: what are the actual asset dimensions? What sequence markers exist in the editor’s timeline? Giving the model unrestricted database or filesystem write access is dangerous.
CFI communicates through an isolated FastMCP stdio server exposing five strictly read-only tools: get_project_brief, get_client_context, list_project_assets, get_current_creative, and get_previous_revision. No mutating tools exist in the schema, protecting the project from unintended changes.
from mcp.server.fastmcp import FastMCP mcp = FastMCP("cfi_context_server") # Exclusively read-only tool contracts — zero mutation methods @mcp.tool() def locate_timeline_marker(timecode_sec: float) -> dict: """Look up NLE sequence marker and clip asset at the specified timecode.""" return get_active_timeline().query_at(timecode_sec)
The Dual-Enforced Human Checkpoint (interrupt() + RuntimeError)
Soft warnings in prompt completions are easily ignored by downstream automation. If contradictory requests are detected, execution must physically stop.
CFI enforces this with two independent mechanisms: First, Stage 06 calls LangGraph’s native interrupt() pattern, suspending the graph until a verified producer resume payload is received. Second, Stage 07 asserts state.human_review_completed, raising a hard Python RuntimeError if invoked directly without the clearance flag.
def human_review_node(state: FeedbackState): if state.conflicts: return interrupt({ "reason": "CONTRADICTION_DETECTED", "conflicts": state.conflicts, "action_required": "Producer resolution sign-off" }) return {"human_review_completed": True} def generate_revision_brief(state: FeedbackState) -> RevisionBrief: if not state.human_review_completed: raise RuntimeError("Security Violation: human review checkpoint incomplete.") return render_actionable_checklist(state.resolved_items)
Sub-800ms Whisper Audio Transcription via Groq Cloud
Client voice notes and Loom recordings are frequently sent while on the go. Traditional cloud transcription APIs introduce 8 to 20 seconds of latency per memo, destroying real-time developer ergonomics.
CFI routes extracted audio buffers to Groq Cloud’s hosted whisper-large-v3 endpoint. By streaming raw WAV audio to Groq’s custom LPU hardware, transcription latency drops to under 800 milliseconds with complete word-level alignment, mapping spoken phrases to sequence timecodes near-instantly.
async def transcribe_audio(audio_bytes: bytes) -> str: response = await groq_client.audio.transcriptions.create( model="whisper-large-v3", file=("audio.wav", audio_bytes), response_format="json" ) return response.text # Returns in < 800ms
Run the Full Suite Locally in Under 4 Seconds
The full test suite runs offline against pre-recorded multi-stakeholder fixture transcripts. Zero API keys, zero network calls, zero spend required to verify the implementation.