Case 6 — Migrate a Linear Chain to a LangGraph State Machine

# Case 6 — Migrate a Linear Chain to a LangGraph State Machine > *"You have a content-generation pipeline as an LCEL chain. Now you need: retry on bad outputs, human review of drafts, optional reruns, persistence across user sessions. Refactor to LangGraph."* --- ## Step 1 — The original LCEL chain ```python # Original — works for happy path only chain = ( {"topic": RunnablePassthrough()} | research_prompt | research_llm | parser_research | {"research": RunnablePassthrough(), "topic": lambda x: x["topic"]} | draft_prompt | writer_llm | parser_draft | review_prompt | reviewer_llm | parser_review ) # Use result = chain.invoke({"topic": "LLM scaling laws"}) ``` Linear flow: research → draft → review → done. No retries, no human input, no state across calls. ## Step 2 — Identify why it's insufficient Production needs: 1. **Retry on bad drafts.** If reviewer says "REVISE", loop back to writer. 2. **Human review before publishing.** Pause for editorial approval. 3. **Resume across sessions.** User might come back tomorrow to approve. 4. **Inspect intermediate state.** Debugging requires seeing research, drafts, reviews. 5. **Different model per step.** Cheap for research, mid-tier for writing, frontier for review. None of these fit cleanly in LCEL. Migrate to LangGraph. ## Step 3 — Design the state schema ```python from typing import TypedDict, Annotated, Literal import operator class ContentState(TypedDict): topic: str research: str draft: str review: str review_decision: Literal["APPROVED", "REVISE", ""] revision_count: int human_approval: Literal["pending", "approved", "rejected", ""] messages: Annotated[list, operator.add] # for audit trail ``` Each piece corresponds to a step's output. The flags (`review_decision`, `human_approval`) drive routing. ## Step 4 — Decompose into nodes ```python def research_node(state): response = research_llm.invoke(research_prompt.format(topic=state["topic"])) research = parser_research(response) return { "research": research, "messages": [SystemMessage(f"Research complete: {len(research)} chars")], } def draft_node(state): revision_context = "" if state["review_decision"] == "REVISE": revision_context = f"\nPrevious review feedback: {state['review']}\nPlease address." response = writer_llm.invoke(draft_prompt.format( topic=state["topic"], research=state["research"], revision_context=revision_context, )) draft = parser_draft(response) return { "draft": draft, "revision_count": state["revision_count"] + 1, "messages": [SystemMessage(f"Draft v{state['revision_count'] + 1} complete")], } def review_node(state): response = reviewer_llm.invoke(review_prompt.format( topic=state["topic"], research=state["research"], draft=state["draft"], )) review = parser_review(response) if "APPROVED" in review: decision = "APPROVED" elif "REVISE" in review: decision = "REVISE" else: decision = "APPROVED" # default return { "review": review, "review_decision": decision, "messages": [SystemMessage(f"Review: {decision}")], } def await_human_approval(state): # This is a pass-through node; interrupt_before pauses execution here return {} def publish_node(state): # Final step — could push to CMS, send email, etc. return {"messages": [SystemMessage("Published.")]} ``` ## Step 5 — Define routing ```python MAX_REVISIONS = 3 def after_review(state): if state["review_decision"] == "APPROVED": return "await_human" # Goes to human approval elif state["revision_count"] >= MAX_REVISIONS: return "await_human" # Give up after 3 revisions return "draft" # Revise def after_human(state): if state["human_approval"] == "approved": return "publish" elif state["human_approval"] == "rejected": # Could loop back to draft with feedback, or end return END return END # No approval = no publish builder = StateGraph(ContentState) builder.add_node("research", research_node) builder.add_node("draft", draft_node) builder.add_node("review", review_node) builder.add_node("await_human", await_human_approval) builder.add_node("publish", publish_node) builder.set_entry_point("research") builder.add_edge("research", "draft") builder.add_edge("draft", "review") builder.add_conditional_edges("review", after_review) builder.add_conditional_edges("await_human", after_human) builder.add_edge("publish", END) # Pause for human review graph = builder.compile( checkpointer=PostgresSaver.from_conn_string(os.environ["DATABASE_URL"]), interrupt_before=["await_human"], # Pause before human approval ) ``` ## Step 6 — Run it end-to-end ```python config = {"configurable": {"thread_id": "article-001"}} # Start the pipeline state = graph.in