get_graph() — langchain Function Reference

Architecture documentation for the get_graph() function in base.py from the langchain codebase.

Function python LangChainCore Runnables calls 3 called by 1

Entity Profile

LangChainCore→ Runnables→ get_graph() — langchain Function Reference

Dependency Diagram

graph TD
  dd68934b_b24b_922a_dfc2_d96ad2d88349["get_graph()"]
  17599172_8889_afc6_2237_4429f3439071["RunnableParallel"]
  dd68934b_b24b_922a_dfc2_d96ad2d88349 -->|defined in| 17599172_8889_afc6_2237_4429f3439071
  92646703_74f4_de33_6f1d_887ee39c4db1["get_graph()"]
  92646703_74f4_de33_6f1d_887ee39c4db1 -->|calls| dd68934b_b24b_922a_dfc2_d96ad2d88349
  faf9f914_2571_c802_8b6f_3c1dbc8bcacd["get_input_schema()"]
  dd68934b_b24b_922a_dfc2_d96ad2d88349 -->|calls| faf9f914_2571_c802_8b6f_3c1dbc8bcacd
  38e57780_7b6f_748a_55bd_bddc272457df["get_output_schema()"]
  dd68934b_b24b_922a_dfc2_d96ad2d88349 -->|calls| 38e57780_7b6f_748a_55bd_bddc272457df
  4b92cace_dc70_6808_9c96_2880d0125ea4["get_graph()"]
  dd68934b_b24b_922a_dfc2_d96ad2d88349 -->|calls| 4b92cace_dc70_6808_9c96_2880d0125ea4
  style dd68934b_b24b_922a_dfc2_d96ad2d88349 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/base.py lines 3787–3823

    def get_graph(self, config: RunnableConfig | None = None) -> Graph:
        """Get the graph representation of the `Runnable`.

        Args:
            config: The config to use.

        Returns:
            The graph representation of the `Runnable`.

        Raises:
            ValueError: If a `Runnable` has no first or last node.

        """
        # Import locally to prevent circular import
        from langchain_core.runnables.graph import Graph  # noqa: PLC0415

        graph = Graph()
        input_node = graph.add_node(self.get_input_schema(config))
        output_node = graph.add_node(self.get_output_schema(config))
        for step in self.steps__.values():
            step_graph = step.get_graph()
            step_graph.trim_first_node()
            step_graph.trim_last_node()
            if not step_graph:
                graph.add_edge(input_node, output_node)
            else:
                step_first_node, step_last_node = graph.extend(step_graph)
                if not step_first_node:
                    msg = f"Runnable {step} has no first node"
                    raise ValueError(msg)
                if not step_last_node:
                    msg = f"Runnable {step} has no last node"
                    raise ValueError(msg)
                graph.add_edge(input_node, step_first_node)
                graph.add_edge(step_last_node, output_node)

        return graph