_add_middleware_edge() — langchain Function Reference

Architecture documentation for the _add_middleware_edge() function in factory.py from the langchain codebase.

Function python LangChainCore ApiManagement calls 1 called by 1

Entity Profile

LangChainCore→ ApiManagement→ _add_middleware_edge() — langchain Function Reference

Dependency Diagram

graph TD
  b82fcc4d_63ed_966a_2618_0b729f787b4e["_add_middleware_edge()"]
  fd7a28b1_3772_169b_6524_1342f35143b1["factory.py"]
  b82fcc4d_63ed_966a_2618_0b729f787b4e -->|defined in| fd7a28b1_3772_169b_6524_1342f35143b1
  f4b66c38_651c_807a_caca_41f73fbbe516["create_agent()"]
  f4b66c38_651c_807a_caca_41f73fbbe516 -->|calls| b82fcc4d_63ed_966a_2618_0b729f787b4e
  63280d36_c373_21de_7184_3a4b4f8606bb["_resolve_jump()"]
  b82fcc4d_63ed_966a_2618_0b729f787b4e -->|calls| 63280d36_c373_21de_7184_3a4b4f8606bb
  style b82fcc4d_63ed_966a_2618_0b729f787b4e fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain_v1/langchain/agents/factory.py lines 1767–1812

def _add_middleware_edge(
    graph: StateGraph[
        AgentState[ResponseT], ContextT, _InputAgentState, _OutputAgentState[ResponseT]
    ],
    *,
    name: str,
    default_destination: str,
    model_destination: str,
    end_destination: str,
    can_jump_to: list[JumpTo] | None,
) -> None:
    """Add an edge to the graph for a middleware node.

    Args:
        graph: The graph to add the edge to.
        name: The name of the middleware node.
        default_destination: The default destination for the edge.
        model_destination: The destination for the edge to the model.
        end_destination: The destination for the edge to the end.
        can_jump_to: The conditionally jumpable destinations for the edge.
    """
    if can_jump_to:

        def jump_edge(state: dict[str, Any]) -> str:
            return (
                _resolve_jump(
                    state.get("jump_to"),
                    model_destination=model_destination,
                    end_destination=end_destination,
                )
                or default_destination
            )

        destinations = [default_destination]

        if "end" in can_jump_to:
            destinations.append(end_destination)
        if "tools" in can_jump_to:
            destinations.append("tools")
        if "model" in can_jump_to and name != model_destination:
            destinations.append(model_destination)

        graph.add_conditional_edges(name, RunnableCallable(jump_edge, trace=False), destinations)

    else:
        graph.add_edge(name, default_destination)