Routing Multi-Step LangGraph & AutoGen Workflows to Human Closers Before State Desynchronization

Multi-agent frameworks such as LangGraph and AutoGen are increasingly used to draft proposals, negotiate pricing, and review contract language in B2B sales. Long-running, multi-turn negotiations are where they tend to break down. This article calls that breakdown state desynchronization. The term is this article’s shorthand, not an established industry standard. It means the agent’s working picture of the deal (conversation history, tool outputs, approval limits) no longer matches the real deal.

Published research supports the underlying problem. A NeurIPS 2025 study of multi-agent LLM systems (Cemri et al., “Why Do Multi-Agent LLM Systems Fail?”) analyzed traces from seven frameworks and identified 14 failure modes in three categories: specification issues, inter-agent misalignment, and task verification. Several of those modes map directly onto sales workflows, including loss of conversation history, conversation resets, task derailment, ignoring another agent’s input, and premature termination.

The practical answer is not to prompt the model harder. It is to detect the problem with deterministic code, freeze the workflow at a persisted checkpoint, and route that checkpoint to a human Account Executive (AE). This guide shows how to build that handoff with LangGraph’s interrupt() and AutoGen’s Handoff mechanism, plus Slack alerting and safe resumption.

A multi-agent sales workflow typically splits work across specialized roles:

  Discovery Agent ───► Pricing Agent (CPQ) ───► Legal Agent (MSA)
                            │                        │
                            ▼                        ▼
                   [Context drift / loops]   [Unsupported terms]
                            └──────────┬─────────────┘
                                       ▼
                         State desynchronization
                      (wrong quote, stalled negotiation)

GraphRecursionError when the configured recursion_limit is exceeded, and AutoGen provides termination conditions such as MaxMessageTermination. Retrying a drifted agent usually adds more context to the same confused state. A better pattern is to stop, persist the exact state, and hand it to a person.

A reliable handoff needs three properties: durable state, idempotent resumption, and a clear owner for the deal at every moment.

┌──────────────────────────────────────────────────────────────┐
│ AGENT RUNTIME (LangGraph graph / AutoGen team)               │
│   guardrail check ──(threshold hit)──►  / handoff       │
└───────────────┬──────────────────────────────▲───────────────┘
                │ persist state                │ resume with human decision
                ▼                              │
        ┌───────────────────┐        ┌─────────┴──────────┐
        │ Durable store     │        │ Resume gateway     │
        │ (Postgres, etc.)  │        │ (verifies, locks,  │
        └─────────┬─────────┘        │  dedupes, resumes) │
                  │                  └─────────▲──────────┘
                  ▼                            │ button click
        ┌───────────────────┐   Slack message  │
        │ Escalation service├──────────────────┘
        │ (summary + blocks)│   to AE channel
        └───────────────────┘

interrupt() and Command-Based Resumption LangGraph models a workflow as a state graph. Its interrupt() function s execution inside a node and surfaces a JSON-serializable payload to the caller. The graph state is saved through the persistence layer, and the workflow waits indefinitely until you resume it.

Three requirements apply (per the LangGraph interrupts documentation):

thread_id To resume, you invoke the graph again with Command(resume=...). That value becomes the return value of the interrupt() call.

try/except.

import os
from typing import Annotated, Any, TypedDict

from langchain_core.messages import AIMessage
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.types import Command, interrupt

AUTO_APPROVAL_LIMIT = 0.15  # discounts above 15% need an AE

class DealState(TypedDict, total=False):
    messages: Annotated[list, add_messages]
    customer_id: str
    deal_size: float
    requested_discount: float  # fraction: 0.18 means 18%
    approval_required: bool
    human_feedback: dict[str, Any]
    final_discount: float

def check_guardrails(state: DealState) -> dict[str, Any]:
    """Deterministic policy check. No LLM decides whether a limit applies."""
    discount = state.get("requested_discount", 0.0)
    if discount > AUTO_APPROVAL_LIMIT:
        return {
            "approval_required": True,
            "messages": [AIMessage(
                content=f"Requested discount {discount:.0%} exceeds the "
                        f"{AUTO_APPROVAL_LIMIT:.0%} auto-approval limit. Escalating to AE."
            )],
        }
    return {
        "approval_required": False,
        "final_discount": discount,
        "messages": [AIMessage(content=f"Discount {discount:.0%} auto-approved.")],
    }

def human_review(state: DealState) -> dict[str, Any]:
    decision = interrupt({
        "event": "HUMAN_APPROVAL_REQUIRED",
        "customer_id": state["customer_id"],
        "deal_size": state["deal_size"],
        "requested_discount": state["requested_discount"],
        "reason": f"Discount above {AUTO_APPROVAL_LIMIT:.0%}",
    })

    update: dict[str, Any] = {
        "approval_required": False,
        "human_feedback": decision,
        "messages": [AIMessage(
            content=f"AE decision: {decision['action']}. Notes: {decision.get('notes') or 'none'}"
        )],
    }
    if decision["action"] == "APPROVED":
        update["final_discount"] = state["requested_discount"]
    elif decision["action"] == "MODIFIED":
        update["final_discount"] = decision["adjusted_discount"]
    return update  # "REJECTED" leaves final_discount unset

def route_after_guardrails(state: DealState) -> str:
    return "human_review" if state.get("approval_required") else END

builder = StateGraph(DealState)
builder.add_node("check_guardrails", check_guardrails)
builder.add_node("human_review", human_review)
builder.add_edge(START, "check_guardrails")
builder.add_conditional_edges("check_guardrails", route_after_guardrails)
builder.add_edge("human_review", END)

DB_URI = os.environ["POSTGRES_URI"]

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # creates tables; required the first time you use the checkpointer
    graph = builder.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "deal-acme-001"}}
    result = graph.invoke(
        {
            "customer_id": "acme",
            "deal_size": 500_000.0,
            "requested_discount": 0.25,
        },
        config,
    )

    if "__interrupt__" in result:
        payload = result["__interrupt__"][0].value

A few production notes on this code:

with block is for brevity. In a long-running service, create the checkpointer or connection pool once at startup and keep it open.from_conn_string, the LangGraph docs say to pass autocommit=True and row_factory=dict_row. Without them you can see errors such as TypeError: tuple indices must be integers or slices, not str. LANGGRAPH_STRICT_MSGPACK=true, or passing an explicit allowed_msgpack_modules list. This restricts checkpoint deserialization to known-safe types, which matters because the checkpoint store holds your deal data.

from langgraph.types import Command

def resume_deal(graph, thread_id: str, decision: dict[str, Any]) -> None:
    """Resume a d deal with the AE's decision."""
    config = {"configurable": {"thread_id": thread_id}}

    if not graph.get_state(config).next:
        return

    for event in graph.stream(Command(resume=decision), config=config):
        print("Resumed event:", event)

The get_state(...).next check reduces duplicate resumes but does not eliminate the race between check and resume. Pair it with the locking pattern in section 6.

Status note. AutoGen 0.4 was a ground-up redesign built around an event-driven core, with the high-level AgentChat API on top. Microsoft has since put AutoGen in maintenance mode (since October 2025: bug and security fixes only). The official successor is Microsoft Agent Framework, which reached 1.0 general availability on April 3, 2026 and merges Semantic Kernel’s enterprise foundations with AutoGen’s orchestration patterns. Microsoft describes its workflows as graph-based, with checkpointing and human-in-the-loop support. The latest AgentChat release at the time of writing is 0.7.5. Existing AutoGen projects can keep using the patterns below. For new builds, evaluate Agent Framework or LangGraph.

AutoGen’s documentation describes two ways to involve a human:

UserProxyAgent during a run. For the second approach, give the agent a Handoff to a target name. That generates a transfer tool, and the team stops when HandoffTermination detects the resulting HandoffMessage. The model must support tool calling for handoffs to work.

import asyncio

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.base import Handoff
from autogen_agentchat.conditions import HandoffTermination, TextMentionTermination
from autogen_agentchat.messages import HandoffMessage
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient

AE_TARGET = "human_ae"

def build_team(model_client: OpenAIChatCompletionClient) -> RoundRobinGroupChat:
    sales_agent = AssistantAgent(
        "sales_ai",
        model_client=model_client,
        handoffs=[Handoff(target=AE_TARGET, message="Escalating to the human account executive.")],
        system_message=(
            "You are an enterprise sales assistant. If the prospect asks for a discount "
            "above 15%, a custom SLA, or changes to liability terms, transfer to the human "
            "account executive. When the deal is finished, reply with DEAL_CLOSED."
        ),
    )
    termination = HandoffTermination(target=AE_TARGET) | TextMentionTermination("DEAL_CLOSED")
    return RoundRobinGroupChat([sales_agent], termination_condition=termination)

async def run_until_handoff(task: str) -> dict:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")  # any tool-calling model works
    try:
        team = build_team(model_client)
        result = await team.run(task=task)

        last = result.messages[-1]
        if isinstance(last, HandoffMessage) and last.target == AE_TARGET:
            state = await team.save_state()  # JSON-serializable dict; persist to your database
            return {"status": "escalated", "state": state, "summary": last.content}
        return {"status": "completed", "stop_reason": result.stop_reason}
    finally:
        await model_client.close()

async def resume_after_human(saved_state: dict, ae_message: str):
    model_client = OpenAIChatCompletionClient(model="gpt-4o")
    try:
        team = build_team(model_client)
        await team.load_state(saved_state)
        return await team.run(task=ae_message)
    finally:
        await model_client.close()

Swarm with HandoffTermination targeting the user, the docs say you resume by setting the task to a HandoffMessage whose target is the next agent to run.max_turns for conversational loops.RoundRobinGroupChat, SelectorGroupChat, or An AE should be able to decide from the alert alone, with a link to the full trace for anyone who wants more. Slack’s Block Kit supports this with a header, summary fields, and buttons.

Slack limits to design around:

value to 2,000 characters.action_id and block_id are limited to 255 characters. Keep value small, such as an ID, and look up everything else server-side from the thread ID.

def build_escalation_blocks(
    thread_id: str,
    customer_name: str,
    deal_value: str,
    stall_reason: str,
    agent_summary: str,
) -> dict:
    """Build a Slack Block Kit message for an AE handoff."""
    return {
        "text": f"Human review needed: {customer_name} ({deal_value})",
        "blocks": [
            {
                "type": "header",
                "text": {"type": "plain_text", "text": "Human review required: AI sales handoff"},
            },
            {
                "type": "section",
                "fields": [
                    {"type": "mrkdwn", "text": f"*Customer:*\n{customer_name}"},
                    {"type": "mrkdwn", "text": f"*Deal value:*\n{deal_value}"},
                    {"type": "mrkdwn", "text": f"*Thread ID:*\n`{thread_id}`"},
                    {"type": "mrkdwn", "text": f"*Trigger:*\n{stall_reason}"},
                ],
            },
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": f"*Agent summary:*\n>{agent_summary}"},
            },
            {"type": "divider"},
            {
                "type": "actions",
                "block_id": f"escalation_{thread_id}",
                "elements": [
                    {"type": "button", "action_id": "deal_approve", "style": "primary",
                     "text": {"type": "plain_text", "text": "Approve"}, "value": thread_id},
                    {"type": "button", "action_id": "deal_modify",
                     "text": {"type": "plain_text", "text": "Modify terms"}, "value": thread_id},
                    {"type": "button", "action_id": "deal_takeover", "style": "danger",
                     "text": {"type": "plain_text", "text": "Take over"}, "value": thread_id},
                ],
            },
        ],
    }

Practical notes:

&, <, and >. response_url accepts follow-up messages, so you can update the card afterward (“Approved by …”).

The right choice depends on your stack and risk tolerance. For approval workflows, the key question is whether you can reliably persist, inspect, and resume state. Explicit state and first-class interrupts make that easier.

Long multi-agent sales workflows fail in ways that more prompting rarely fixes. A sturdier design:

LangGraph’s interrupt() and checkpointers cover most of this with explicit state, and AutoGen’s handoff and save_state() pattern can do it too, though you provide more of the plumbing yourself. If you are starting a new project today, weigh AutoGen’s maintenance status and consider LangGraph or Microsoft Agent Framework.

Originally published at https://instachime.com/blog/routing-multi-step-langgraph-autogen-workflows-to-human-closers-before-state-desynchronization