Skip to content

evalwire.langgraph

evalwire.langgraph

LangGraph node isolation helpers (requires evalwire[langgraph] extra).

build_subgraph(nodes, state_cls, input_cls=None, *, name=None, checkpointer=None)

Assemble a minimal linear StateGraph from a list of node callables.

Edges are added in declaration order: START → nodes[0] → nodes[1] → … → END.

Parameters

nodes: Ordered list of (node_name, node_callable) pairs. state_cls: LangGraph state dataclass for the graph. input_cls: Optional input-schema dataclass passed as input_schema=input_cls to StateGraph. name: Optional graph name used for Phoenix tracing. checkpointer: Optional LangGraph checkpointer.

Returns

CompiledStateGraph The compiled subgraph, ready for await graph.ainvoke(...).

Source code in src/evalwire/langgraph.py 
12
13
14
15
16
17
18
19
20
21
22
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
def build_subgraph(
    nodes: list[tuple[str, Callable]],
    state_cls: type[Any],
    input_cls: type[Any] | None = None,
    *,
    name: str | None = None,
    checkpointer: Any = None,
) -> CompiledStateGraph:
    """Assemble a minimal linear StateGraph from a list of node callables.

    Edges are added in declaration order:
    ``START → nodes[0] → nodes[1] → … → END``.

    Parameters
    ----------
    nodes:
        Ordered list of ``(node_name, node_callable)`` pairs.
    state_cls:
        LangGraph state dataclass for the graph.
    input_cls:
        Optional input-schema dataclass passed as ``input_schema=input_cls``
        to ``StateGraph``.
    name:
        Optional graph name used for Phoenix tracing.
    checkpointer:
        Optional LangGraph checkpointer.

    Returns
    -------
    CompiledStateGraph
        The compiled subgraph, ready for ``await graph.ainvoke(...)``.
    """
    try:
        from langgraph.graph import (
            END,
            START,
            StateGraph,
        )
    except ImportError as exc:
        raise ImportError(
            "evalwire[langgraph] is required to use build_subgraph. "
            "Install it with: pip install 'evalwire[langgraph]'"
        ) from exc

    kwargs: dict[str, Any] = {}
    if input_cls is not None:
        kwargs["input_schema"] = input_cls

    graph = StateGraph(state_cls, **kwargs)

    for node_name, node_fn in nodes:
        graph.add_node(node_name, node_fn)

    node_names = [n for n, _ in nodes]
    graph.add_edge(START, node_names[0])
    for a, b in zip(node_names, node_names[1:]):
        graph.add_edge(a, b)
    graph.add_edge(node_names[-1], END)

    compile_kwargs: dict[str, Any] = {}
    if checkpointer is not None:
        compile_kwargs["checkpointer"] = checkpointer

    compiled = graph.compile(**compile_kwargs)
    if name is not None:
        compiled.name = name

    return compiled

invoke_node(node_fn, query, state_cls, *, message_field='messages', config=None) async

Directly call a single LangGraph node as a standalone async function.

Constructs a state_cls instance with a single HumanMessage placed in message_field, calls await node_fn(state=state, config=config), and returns the raw result.

Parameters

node_fn: The node callable to invoke. query: The human message content to place in message_field. state_cls: LangGraph state dataclass used to build the input state. message_field: Name of the field on state_cls that holds the messages list. config: Optional RunnableConfig. A default empty config is used if None.

Returns

Any The raw return value of the node callable.

Source code in src/evalwire/langgraph.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
async def invoke_node(
    node_fn: Callable,
    query: str,
    state_cls: type[Any],
    *,
    message_field: str = "messages",
    config: Any = None,
) -> Any:
    """Directly call a single LangGraph node as a standalone async function.

    Constructs a ``state_cls`` instance with a single ``HumanMessage`` placed
    in ``message_field``, calls ``await node_fn(state=state, config=config)``,
    and returns the raw result.

    Parameters
    ----------
    node_fn:
        The node callable to invoke.
    query:
        The human message content to place in ``message_field``.
    state_cls:
        LangGraph state dataclass used to build the input state.
    message_field:
        Name of the field on ``state_cls`` that holds the messages list.
    config:
        Optional ``RunnableConfig``. A default empty config is used if ``None``.

    Returns
    -------
    Any
        The raw return value of the node callable.
    """
    try:
        from langchain_core.messages import HumanMessage
        from langchain_core.runnables import (
            RunnableConfig,
        )
    except ImportError as exc:
        raise ImportError(
            "langchain-core is required to use invoke_node. "
            "Install it with: pip install langchain-core"
        ) from exc

    state = state_cls(**{message_field: [HumanMessage(content=query)]})
    resolved_config = config if config is not None else RunnableConfig()
    return await node_fn(state=state, config=resolved_config)