Skip to content

symfonic.core.nodes.elicitation

elicitation

ElicitationNode — interrupt node for ask_user tool calls.

This node is inserted into the graph topology between the react node and the tools node when ask_user_enabled=True. It runs AFTER the react node has returned normally (and its AIMessage has been checkpointed), so the interrupt() call here can never orphan a ToolMessage.

Graph flow when ask_user is called

react -> (conditional: _ask_user_pending set?) yes -> elicitation [interrupt() here] [on resume: inject ToolMessage] no -> tools | END (normal tool/finish routing)

create_elicitation_node 

create_elicitation_node() -> Any

Return the elicitation node function.

The node reads state["_ask_user_pending"], calls interrupt(), and on resume builds a ToolMessage and clears _ask_user_pending.

Source code in src/symfonic/core/nodes/elicitation.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def create_elicitation_node() -> Any:
    """Return the elicitation node function.

    The node reads ``state["_ask_user_pending"]``, calls ``interrupt()``,
    and on resume builds a ``ToolMessage`` and clears ``_ask_user_pending``.
    """

    async def elicitation(state: dict[str, Any]) -> dict[str, Any]:
        from langchain_core.messages import ToolMessage
        from langgraph.types import interrupt

        from symfonic.core.contracts.elicitation import (
            AskUserQuestion,
            AskUserRequest,
        )

        pending: dict[str, Any] = state["_ask_user_pending"]
        tool_call_id: str = pending["tool_call_id"]
        raw_request: dict[str, Any] = pending.get("request_input", {})

        # Server-stamp question IDs (uuid4.hex[:12]) — the LLM never supplies ids.
        questions = [
            AskUserQuestion(**q_data)
            for q_data in raw_request.get("questions", [])
        ]
        stamped_request = AskUserRequest(questions=questions)

        # Call interrupt() — LangGraph surfaces this as an ExtensionEvent
        # with type="on_interrupt" in astream_events(v2).  The engine's
        # stream_typed() intercepts it and calls _mint_pause_token.
        # On resume, interrupt() returns the value passed to graph.invoke()
        # as the Command.resume payload.
        resume_value: Any = interrupt(
            {
                "type": "ask_user",
                "tool_call_id": tool_call_id,
                "request": stamped_request.model_dump(),
            }
        )

        # --- Resume path (re-invocation after graph.update_state / Command) ---
        # LangGraph re-runs this node body from the top; interrupt() returns
        # the resume payload this time instead of raising GraphInterrupt.
        if resume_value is not None:
            # The resume payload is either the raw AskUserResponse dict or
            # an already-parsed object.  The engine's resume() method builds
            # the ToolMessage and injects it via state_overrides["messages"],
            # so on the resume path we just clear the pending marker and let
            # the normal tool->react loop continue.
            tool_msg = ToolMessage(
                content=(
                    resume_value
                    if isinstance(resume_value, str)
                    else str(resume_value)
                ),
                tool_call_id=tool_call_id,
                name="ask_user",
            )
            return {
                "messages": [tool_msg],
                "_ask_user_pending": None,
            }

        # Should not reach here in normal operation, but be safe.
        logger.warning(
            "elicitation node: interrupt() returned None unexpectedly for "
            "tool_call_id=%s", tool_call_id
        )
        return {"_ask_user_pending": None}

    return elicitation

elicitation_condition 

elicitation_condition(state: dict[str, Any]) -> str

Conditional-edge router: routes to 'elicitation' when ask_user is pending.

Used as the conditional edge from 'react'. Returns: - "elicitation" when _ask_user_pending is set (truthy) - "continue" when normal tool calls are present (handled by the existing tools_condition edge) - "finish" when there are no tool calls at all

Source code in src/symfonic/core/nodes/elicitation.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def elicitation_condition(state: dict[str, Any]) -> str:
    """Conditional-edge router: routes to 'elicitation' when ask_user is pending.

    Used as the conditional edge from 'react'.  Returns:
      - ``"elicitation"`` when ``_ask_user_pending`` is set (truthy)
      - ``"continue"``    when normal tool calls are present (handled by
                          the existing tools_condition edge)
      - ``"finish"``      when there are no tool calls at all
    """
    if state.get("_ask_user_pending"):
        return "elicitation"

    # Fall through to the standard tools_condition logic
    from symfonic.core.edges.tool_condition import tools_condition

    return tools_condition(state)