Backends

The backends module enable ebiose agents to be executed with differents frameworks. A backendclass implement 3 sub classes :

1. Agent
2. LLMNode
3. ValidatorNode

LangGraph

Bases: Backend

This class handles the compilation from an Agent to a compiled runnable ready to be executed.

Source code in ebiose/backends/langgraph.py

class LangGraph(Backend):
    """This class handles the compilation from an Agent to a compiled runnable ready to be executed."""

    class Agent(Agent):  # noqa: D106
        state: (dict | BaseModel) | None = None
        state_type: Literal["pydantic", "dict"] = "pydantic"
        model: str = "gpt-3.5-turbo"
        compiled_graph: CompiledGraph | None = None

        @model_validator(mode="after")
        def __set_llm_models(self) -> LangGraph.Agent:
            if self.state_type == "dict":
                msg = (
                    "Dict state type is not fully supported yet. Use pydantic instead."
                )
                raise NotImplementedError(
                    msg,
                )
            for i, node in enumerate(self.graph.nodes):
                if isinstance(node, LLMNode):
                    temp_node = LangGraph.LLMNode(
                        id=node.id,
                        name=node.name,
                        purpose=node.purpose,
                        prompt=node.prompt,
                        model=self.model,
                    )
                    self.graph.nodes[i] = temp_node
            return self

        @classmethod
        def from_ebiose_agent(
            cls,
            agent: Agent,
            model: str = "gpt-3.5-turbo",
            state_type: Literal["pydantic", "dict"] = "pydantic",
        ) -> Self:
            """Create an Agent that can be later a runnable agent from an ebiose Agent."""
            return cls(
                graph=agent.graph,
                input_model=agent.input_model,
                output_model=agent.output_model,
                model=model,
                state_type=state_type,
            )

        def _build_state(self) -> dict | BaseModel:
            """Build dynamically the state of the agent with the nodes of the graph.

            For each nodes it creates a node_id_prompt, node_it_responses where all responses are stored,
            """
            if self.state is not None:
                return self.state

            if self.state_type == "pydantic":
                state_attributes = {
                    "shared_context_prompt": (str, self.graph.shared_context_prompt),
                    "input": (self.input_model, ...),
                    "output": (Optional[self.output_model], None),
                    "history": (Annotated[Sequence[str], add], []),
                    "node_sequence": (Annotated[Sequence[str], add], []),
                }

                for node in self.graph.nodes:
                    if isinstance(node, StartNode | EndNode):
                        continue

                    if isinstance(node, LLMNode):
                        self.__build_llm_node_output_model(node)
                        state_attributes[f"{node.id}_prompt"] = (str, node.prompt)
                        state_attributes[f"{node.id}_responses"] = (
                            Annotated[Sequence[node.output_model], add],
                            [],
                        )
                    else:
                        msg = f"Node type {type(node)} not supported"
                        raise NotImplementedError(
                            msg,
                        )
                self.state = create_model("AgentState", **state_attributes)

            elif self.state_type == "dict":
                state_attributes = {
                    "shared_context_prompt": str,
                    "input": self.input_model,
                    "output": Optional[self.output_model],
                    "history": Annotated[Sequence[str], add],
                    "node_sequence": Annotated[Sequence[str], add],
                }

                for node in self.graph.nodes:
                    if isinstance(node, StartNode | EndNode):
                        continue
                    if isinstance(node, LLMNode):
                        node.state_type = self.state_type
                        self.__build_llm_node_output_model(node)
                        state_attributes[f"{node.id}_prompt"] = str
                        state_attributes[f"{node.id}_responses"] = Annotated[
                            Sequence[node.output_model],
                            add,
                        ]
                        state_attributes[f"{node.id}_prompt"] = str
                        state_attributes[f"{node.id}_responses"] = Annotated[
                            Sequence[node.output_model],
                            add,
                        ]
                    else:
                        msg = f"Node type {type(node)} not supported"

                        raise NotImplementedError(
                            msg,
                        )
                self.state = TypedDict("AgentState", state_attributes)
            else:
                msg = f"Mode {self.mode} not supported."
                raise ValueError(msg)

            return self.state

        def __build_llm_node_output_model(self, node: BaseNode) -> None:
            """This function builds the pydantic schema of the ouput of the node.

            - If there are conditions, it adds the condition field to the schema.
            - If it is the last node, it adds the agent_output field to the schema.
            """
            if node.output_model is not None:
                return

            conditions = set()
            is_last_node = False
            for edge in self.graph.get_outgoing_edges(node.id, conditional=None):
                if edge.condition is not None:
                    conditions.add(edge.condition)
                if edge.end_node_id == self.graph.get_end_node_id():
                    is_last_node = True

            llm_node_output_attributes = (
                {
                    "response": (
                        str,
                        Field(description="The response of the LLM model"),
                    ),
                }
                if self.state_type == "pydantic"
                else {
                    "response": str,
                }
            )

            if conditions:
                llm_node_output_attributes["condition"] = (
                    (
                        Literal[tuple(conditions)],
                        Field(
                            description="The chosen condition to reach the next node according to the response",
                        ),
                    )
                    if self.state_type == "pydantic"
                    else {
                        "condition": Literal[tuple(conditions)],
                    }
                )

            if is_last_node:
                llm_node_output_attributes["agent_output"] = (
                    (
                        self.output_model,
                        Field(
                            description="The final output of the agent to meet the user's request",
                        ),
                    )
                    if self.state_type == "pydantic"
                    else {
                        "agent_output": self.output_model,
                    }
                )

            if self.state_type == "pydantic":
                node.output_model = create_model(
                    f"{node.id.title()}Output",
                    __doc__=f"The model for formatting the output of the {node.id} LLM node",
                    **llm_node_output_attributes,
                )
            else:
                node.output_model = TypedDict(
                    f"{node.id.title()}Output",
                    llm_node_output_attributes,
                )

        def compile(self) -> CompiledGraph:
            """Compile the agent into a runnable graph."""
            # Cache it if it has not been compiled yet
            if self.compiled_graph is not None:
                return self.compiled_graph

            self._build_state()
            compiled_graph = self.__to_compiled_graph()
            self.compiled_graph = compiled_graph
            return compiled_graph

        def invoke_graph(
            self,
            agent_input: BaseModel,
        ) -> BaseModel:
            """Compile and run the agent.

            Args:
                agent_input: The input that goes in first trough the graph

            Returns: the final updated graph state

            """
            compiled_graph = self.compile()
            initial_state = self.state(
                input=agent_input,
                shared_context_prompt=self.graph.shared_context_prompt,
            )
            return compiled_graph.invoke(initial_state)

        def run(
            self,
            agent_input: BaseModel,
        ) -> BaseModel:
            """Compile and run the agent.

            Args:
                agent_input: The input that goes in first trough the graph

            Returns: the agent last response at the end of the graph

            """
            final_state = self.invoke_graph(agent_input)
            last_node_id = final_state["node_sequence"][-1]
            agent_output = final_state[f"{last_node_id}_responses"][-1].agent_output

            if agent_output is None:
                msg = "No agent output found."
                raise ValueError(msg)

            return agent_output

        def run_in_batch(
            self,
            agent_inputs: list[BaseModel],
            max_concurrency: int = 5,
        ) -> BaseModel:
            """Run the agents multiple times with a batch of inputs."""
            self._build_state()
            compiled_graph: CompiledGraph = self.__to_compiled_graph()
            initial_states = [
                self.state(
                    input=agent_input,
                    shared_context_prompt=self.graph.shared_context_prompt,
                )
                for agent_input in agent_inputs
            ]

            return compiled_graph.batch(
                initial_states,
                config={"max_concurrency": max_concurrency},
            )

        def __to_workflow(self) -> StateGraph:
            workflow = StateGraph(self.state)

            # add nodes to the workflow
            nodes_dict = {node.id: node for node in self.graph.nodes}
            for node_id, node in nodes_dict.items():
                if isinstance(node, (EndNode, StartNode)):
                    continue
                # We first retrieve the function that represents the node it does not depend
                # if we have a llm, a rag or whatnot
                workflow.add_node(node_id, node.call_node)

            # add edges to the workflow
            for node_id, node in nodes_dict.items():
                if isinstance(node, EndNode):
                    continue

                # ---------------------- Not conditional edges ----------------------------
                outgoing_nodes = self.graph.get_outgoing_nodes(
                    node_id,
                    conditional=False,
                )
                outgoing_node_ids = [
                    node.id if not isinstance(node, EndNode) else END
                    for node in outgoing_nodes
                ]

                for outgoing_nodes_id in outgoing_node_ids:
                    workflow.add_edge(
                        node_id if not isinstance(node, StartNode) else START,
                        outgoing_nodes_id,
                    )

                # -----------------------Conditional edges ---------------------
                # we get the destination node of the current node that pass through a conditional edge
                # along with their corresponding condition
                # get_outgoing_edges(self: Self, node_id: str, conditional: Optional[bool]=None)
                outgoing_conditional_edges = self.graph.get_outgoing_edges(
                    node_id,
                    conditional=True,
                )
                if not outgoing_conditional_edges:
                    continue
                path, path_map = get_path(
                    outgoing_conditional_edges,
                    self.graph.get_end_node_id(),
                )
                workflow.add_conditional_edges(
                    source=node_id,
                    path=path,
                    path_map=path_map,
                )

            return workflow

        def __to_compiled_graph(self) -> CompiledGraph:
            """Compile an agent into a runnable graph."""
            return self.__to_workflow().compile()

    class LLMNode(LLMNode):  # noqa: D106
        model: str | None = Field(default=None, exclude=True)
        output_model: (type[BaseModel] | dict) | None = None
        retry_after_validation_errors: bool = False
        max_retries: int = 1
        state_type: Literal["pydantic", "dict"] = "pydantic"
        temperature: float = 0

        def call_node(self, state: BaseModel) -> dict:  # noqa: D102
            # All nodes have access to the shared context prompt
            prompts = (
                [
                    SystemMessage(
                        state.shared_context_prompt.format(
                            **state.input.model_dump(),
                        ),
                    ),
                ]
                if self.state_type == "pydantic"
                else [
                    SystemMessage(
                        state["shared_context_prompt"].format(
                            **state["input"],
                        ),
                    ),
                ]
            )

            # Add history if it exists (only responses are included in the history, see Issue #32)
            is_history_empty = (
                len(state.history) == 0
                if self.state_type == "pydantic"
                else len(state["history"]) == 0
            )
            if not is_history_empty:
                prompts.append(HumanMessage("History of the conversation:"))
                if self.state_type == "pydantic":
                    prompts.append(AIMessage("\n".join(state.history)))
                else:
                    prompts.append(AIMessage("\n".join(state["history"])))

            has_one_field = (
                len(self.output_model.model_fields) == 1
                if self.state_type == "pydantic"
                else len(self.output_model.__annotations__) == 1
            )
            has_conditions = (
                "condition" in self.output_model.model_fields
                if self.state_type == "pydantic"
                else "condition" in self.output_model.__annotations__
            )
            node_prompt = self.prompt.format(
                **state.input.model_dump(),
            )
            if has_one_field:
                prompts.append(
                    HumanMessage(
                        node_prompt,
                    ),
                )
            elif has_conditions:
                conditions = (
                    get_args(
                        self.output_model.model_fields["condition"].annotation,
                    )
                    if self.state_type == "pydantic"
                    else get_args(
                        self.output_model["condition"].__annotations__.values(),
                    )
                )
                prompts.append(
                    HumanMessage(
                        node_prompt
                        + f"\nUse the given tool to provide your response. Based on your response, fill the 'condition' field with the chosen condition to get to the next node amongst the following: {conditions}.",
                    ),
                )
            else:
                prompts.append(
                    HumanMessage(
                        node_prompt + "\nUse the given tool to provide your response.",
                    ),
                )

            # instantiate model
            llm = instantiate_llm_model(self.model, temperature=self.temperature)

            if has_one_field:
                response = llm.invoke(prompts)
                node_output = {"response": response.content}
            else:
                response = llm.bind_tools(
                    [
                        model_to_openai_json_schema(self.output_model),
                    ],
                    strict=False,
                ).invoke(prompts)
                node_output = self.validate_output(response, state)

            return {
                f"{self.id}_responses": [node_output],
                "history": [f"Response of node '{self.id}': {node_output!s}"],
                "node_sequence": [self.id],
            }

        def validate_output(self, response: AIMessage, state: BaseModel) -> dict:  # noqa: D102
            history = state.history
            prompts = [
                HumanMessage("History of the conversation:"),
                AIMessage("\n".join(history)),
            ]

            validation_retry_prompt = """
                You provided the following response:
                '{tool_call}'
                which cannot be parsed because of the following {error_count} errors:
                '{errors}'
                Please fix all these errors and return a corrected response which respects the schema given as a tool."
            """
            no_tool_retry_prompt = """
                You did not call the tool to format the following previous response:
                {response}
                Please call the tool and return a corrected response which respects the schema given as a tool."
            """

            llm = instantiate_llm_model(self.model, temperature=1).bind_tools(
                [
                    model_to_openai_json_schema(self.output_model),
                ],
            )

            n_retry = 0
            while n_retry <= self.max_retries:
                n_retry += 1
                if not response.tool_calls:
                    response = llm.invoke(
                        input=[
                            *prompts,
                            HumanMessage(
                                no_tool_retry_prompt.format(response=response.content),
                            ),
                        ],
                    )
                else:
                    try:
                        return self.output_model.model_validate(
                            response.tool_calls[0]["args"],
                            context=state.model_dump(),
                        )
                    except ValidationError as e:
                        str_e = ""
                        for error in e.errors():
                            str_e += "- {msg} at {loc}\n".format(**error)
                        response = llm.invoke(
                            [
                                *prompts,
                                HumanMessage(
                                    validation_retry_prompt.format(
                                        tool_call=response.tool_calls[0]["args"],
                                        error_count=e.error_count(),
                                        errors=str_e,
                                    ),
                                ),
                            ],
                        )

            return self.output_model.model_validate(
                response.tool_calls[0]["args"],
                context=state.model_dump(),
            )

Agent

Bases: Agent

Source code in ebiose/backends/langgraph.py

class Agent(Agent):  # noqa: D106
    state: (dict | BaseModel) | None = None
    state_type: Literal["pydantic", "dict"] = "pydantic"
    model: str = "gpt-3.5-turbo"
    compiled_graph: CompiledGraph | None = None

    @model_validator(mode="after")
    def __set_llm_models(self) -> LangGraph.Agent:
        if self.state_type == "dict":
            msg = (
                "Dict state type is not fully supported yet. Use pydantic instead."
            )
            raise NotImplementedError(
                msg,
            )
        for i, node in enumerate(self.graph.nodes):
            if isinstance(node, LLMNode):
                temp_node = LangGraph.LLMNode(
                    id=node.id,
                    name=node.name,
                    purpose=node.purpose,
                    prompt=node.prompt,
                    model=self.model,
                )
                self.graph.nodes[i] = temp_node
        return self

    @classmethod
    def from_ebiose_agent(
        cls,
        agent: Agent,
        model: str = "gpt-3.5-turbo",
        state_type: Literal["pydantic", "dict"] = "pydantic",
    ) -> Self:
        """Create an Agent that can be later a runnable agent from an ebiose Agent."""
        return cls(
            graph=agent.graph,
            input_model=agent.input_model,
            output_model=agent.output_model,
            model=model,
            state_type=state_type,
        )

    def _build_state(self) -> dict | BaseModel:
        """Build dynamically the state of the agent with the nodes of the graph.

        For each nodes it creates a node_id_prompt, node_it_responses where all responses are stored,
        """
        if self.state is not None:
            return self.state

        if self.state_type == "pydantic":
            state_attributes = {
                "shared_context_prompt": (str, self.graph.shared_context_prompt),
                "input": (self.input_model, ...),
                "output": (Optional[self.output_model], None),
                "history": (Annotated[Sequence[str], add], []),
                "node_sequence": (Annotated[Sequence[str], add], []),
            }

            for node in self.graph.nodes:
                if isinstance(node, StartNode | EndNode):
                    continue

                if isinstance(node, LLMNode):
                    self.__build_llm_node_output_model(node)
                    state_attributes[f"{node.id}_prompt"] = (str, node.prompt)
                    state_attributes[f"{node.id}_responses"] = (
                        Annotated[Sequence[node.output_model], add],
                        [],
                    )
                else:
                    msg = f"Node type {type(node)} not supported"
                    raise NotImplementedError(
                        msg,
                    )
            self.state = create_model("AgentState", **state_attributes)

        elif self.state_type == "dict":
            state_attributes = {
                "shared_context_prompt": str,
                "input": self.input_model,
                "output": Optional[self.output_model],
                "history": Annotated[Sequence[str], add],
                "node_sequence": Annotated[Sequence[str], add],
            }

            for node in self.graph.nodes:
                if isinstance(node, StartNode | EndNode):
                    continue
                if isinstance(node, LLMNode):
                    node.state_type = self.state_type
                    self.__build_llm_node_output_model(node)
                    state_attributes[f"{node.id}_prompt"] = str
                    state_attributes[f"{node.id}_responses"] = Annotated[
                        Sequence[node.output_model],
                        add,
                    ]
                    state_attributes[f"{node.id}_prompt"] = str
                    state_attributes[f"{node.id}_responses"] = Annotated[
                        Sequence[node.output_model],
                        add,
                    ]
                else:
                    msg = f"Node type {type(node)} not supported"

                    raise NotImplementedError(
                        msg,
                    )
            self.state = TypedDict("AgentState", state_attributes)
        else:
            msg = f"Mode {self.mode} not supported."
            raise ValueError(msg)

        return self.state

    def __build_llm_node_output_model(self, node: BaseNode) -> None:
        """This function builds the pydantic schema of the ouput of the node.

        - If there are conditions, it adds the condition field to the schema.
        - If it is the last node, it adds the agent_output field to the schema.
        """
        if node.output_model is not None:
            return

        conditions = set()
        is_last_node = False
        for edge in self.graph.get_outgoing_edges(node.id, conditional=None):
            if edge.condition is not None:
                conditions.add(edge.condition)
            if edge.end_node_id == self.graph.get_end_node_id():
                is_last_node = True

        llm_node_output_attributes = (
            {
                "response": (
                    str,
                    Field(description="The response of the LLM model"),
                ),
            }
            if self.state_type == "pydantic"
            else {
                "response": str,
            }
        )

        if conditions:
            llm_node_output_attributes["condition"] = (
                (
                    Literal[tuple(conditions)],
                    Field(
                        description="The chosen condition to reach the next node according to the response",
                    ),
                )
                if self.state_type == "pydantic"
                else {
                    "condition": Literal[tuple(conditions)],
                }
            )

        if is_last_node:
            llm_node_output_attributes["agent_output"] = (
                (
                    self.output_model,
                    Field(
                        description="The final output of the agent to meet the user's request",
                    ),
                )
                if self.state_type == "pydantic"
                else {
                    "agent_output": self.output_model,
                }
            )

        if self.state_type == "pydantic":
            node.output_model = create_model(
                f"{node.id.title()}Output",
                __doc__=f"The model for formatting the output of the {node.id} LLM node",
                **llm_node_output_attributes,
            )
        else:
            node.output_model = TypedDict(
                f"{node.id.title()}Output",
                llm_node_output_attributes,
            )

    def compile(self) -> CompiledGraph:
        """Compile the agent into a runnable graph."""
        # Cache it if it has not been compiled yet
        if self.compiled_graph is not None:
            return self.compiled_graph

        self._build_state()
        compiled_graph = self.__to_compiled_graph()
        self.compiled_graph = compiled_graph
        return compiled_graph

    def invoke_graph(
        self,
        agent_input: BaseModel,
    ) -> BaseModel:
        """Compile and run the agent.

        Args:
            agent_input: The input that goes in first trough the graph

        Returns: the final updated graph state

        """
        compiled_graph = self.compile()
        initial_state = self.state(
            input=agent_input,
            shared_context_prompt=self.graph.shared_context_prompt,
        )
        return compiled_graph.invoke(initial_state)

    def run(
        self,
        agent_input: BaseModel,
    ) -> BaseModel:
        """Compile and run the agent.

        Args:
            agent_input: The input that goes in first trough the graph

        Returns: the agent last response at the end of the graph

        """
        final_state = self.invoke_graph(agent_input)
        last_node_id = final_state["node_sequence"][-1]
        agent_output = final_state[f"{last_node_id}_responses"][-1].agent_output

        if agent_output is None:
            msg = "No agent output found."
            raise ValueError(msg)

        return agent_output

    def run_in_batch(
        self,
        agent_inputs: list[BaseModel],
        max_concurrency: int = 5,
    ) -> BaseModel:
        """Run the agents multiple times with a batch of inputs."""
        self._build_state()
        compiled_graph: CompiledGraph = self.__to_compiled_graph()
        initial_states = [
            self.state(
                input=agent_input,
                shared_context_prompt=self.graph.shared_context_prompt,
            )
            for agent_input in agent_inputs
        ]

        return compiled_graph.batch(
            initial_states,
            config={"max_concurrency": max_concurrency},
        )

    def __to_workflow(self) -> StateGraph:
        workflow = StateGraph(self.state)

        # add nodes to the workflow
        nodes_dict = {node.id: node for node in self.graph.nodes}
        for node_id, node in nodes_dict.items():
            if isinstance(node, (EndNode, StartNode)):
                continue
            # We first retrieve the function that represents the node it does not depend
            # if we have a llm, a rag or whatnot
            workflow.add_node(node_id, node.call_node)

        # add edges to the workflow
        for node_id, node in nodes_dict.items():
            if isinstance(node, EndNode):
                continue

            # ---------------------- Not conditional edges ----------------------------
            outgoing_nodes = self.graph.get_outgoing_nodes(
                node_id,
                conditional=False,
            )
            outgoing_node_ids = [
                node.id if not isinstance(node, EndNode) else END
                for node in outgoing_nodes
            ]

            for outgoing_nodes_id in outgoing_node_ids:
                workflow.add_edge(
                    node_id if not isinstance(node, StartNode) else START,
                    outgoing_nodes_id,
                )

            # -----------------------Conditional edges ---------------------
            # we get the destination node of the current node that pass through a conditional edge
            # along with their corresponding condition
            # get_outgoing_edges(self: Self, node_id: str, conditional: Optional[bool]=None)
            outgoing_conditional_edges = self.graph.get_outgoing_edges(
                node_id,
                conditional=True,
            )
            if not outgoing_conditional_edges:
                continue
            path, path_map = get_path(
                outgoing_conditional_edges,
                self.graph.get_end_node_id(),
            )
            workflow.add_conditional_edges(
                source=node_id,
                path=path,
                path_map=path_map,
            )

        return workflow

    def __to_compiled_graph(self) -> CompiledGraph:
        """Compile an agent into a runnable graph."""
        return self.__to_workflow().compile()

__build_llm_node_output_model(node)

This function builds the pydantic schema of the ouput of the node.

• If there are conditions, it adds the condition field to the schema.
• If it is the last node, it adds the agent_output field to the schema.

Source code in ebiose/backends/langgraph.py

def __build_llm_node_output_model(self, node: BaseNode) -> None:
    """This function builds the pydantic schema of the ouput of the node.

    - If there are conditions, it adds the condition field to the schema.
    - If it is the last node, it adds the agent_output field to the schema.
    """
    if node.output_model is not None:
        return

    conditions = set()
    is_last_node = False
    for edge in self.graph.get_outgoing_edges(node.id, conditional=None):
        if edge.condition is not None:
            conditions.add(edge.condition)
        if edge.end_node_id == self.graph.get_end_node_id():
            is_last_node = True

    llm_node_output_attributes = (
        {
            "response": (
                str,
                Field(description="The response of the LLM model"),
            ),
        }
        if self.state_type == "pydantic"
        else {
            "response": str,
        }
    )

    if conditions:
        llm_node_output_attributes["condition"] = (
            (
                Literal[tuple(conditions)],
                Field(
                    description="The chosen condition to reach the next node according to the response",
                ),
            )
            if self.state_type == "pydantic"
            else {
                "condition": Literal[tuple(conditions)],
            }
        )

    if is_last_node:
        llm_node_output_attributes["agent_output"] = (
            (
                self.output_model,
                Field(
                    description="The final output of the agent to meet the user's request",
                ),
            )
            if self.state_type == "pydantic"
            else {
                "agent_output": self.output_model,
            }
        )

    if self.state_type == "pydantic":
        node.output_model = create_model(
            f"{node.id.title()}Output",
            __doc__=f"The model for formatting the output of the {node.id} LLM node",
            **llm_node_output_attributes,
        )
    else:
        node.output_model = TypedDict(
            f"{node.id.title()}Output",
            llm_node_output_attributes,
        )

__to_compiled_graph()

Compile an agent into a runnable graph.

Source code in ebiose/backends/langgraph.py

def __to_compiled_graph(self) -> CompiledGraph:
    """Compile an agent into a runnable graph."""
    return self.__to_workflow().compile()

compile()

Compile the agent into a runnable graph.

Source code in ebiose/backends/langgraph.py

def compile(self) -> CompiledGraph:
    """Compile the agent into a runnable graph."""
    # Cache it if it has not been compiled yet
    if self.compiled_graph is not None:
        return self.compiled_graph

    self._build_state()
    compiled_graph = self.__to_compiled_graph()
    self.compiled_graph = compiled_graph
    return compiled_graph

from_ebiose_agent(agent, model='gpt-3.5-turbo', state_type='pydantic') classmethod

Create an Agent that can be later a runnable agent from an ebiose Agent.

Source code in ebiose/backends/langgraph.py

@classmethod
def from_ebiose_agent(
    cls,
    agent: Agent,
    model: str = "gpt-3.5-turbo",
    state_type: Literal["pydantic", "dict"] = "pydantic",
) -> Self:
    """Create an Agent that can be later a runnable agent from an ebiose Agent."""
    return cls(
        graph=agent.graph,
        input_model=agent.input_model,
        output_model=agent.output_model,
        model=model,
        state_type=state_type,
    )

invoke_graph(agent_input)

Compile and run the agent.

Parameters:

Name	Type	Description	Default
agent_input	BaseModel	The input that goes in first trough the graph	required

Returns: the final updated graph state

Source code in ebiose/backends/langgraph.py

def invoke_graph(
    self,
    agent_input: BaseModel,
) -> BaseModel:
    """Compile and run the agent.

    Args:
        agent_input: The input that goes in first trough the graph

    Returns: the final updated graph state

    """
    compiled_graph = self.compile()
    initial_state = self.state(
        input=agent_input,
        shared_context_prompt=self.graph.shared_context_prompt,
    )
    return compiled_graph.invoke(initial_state)

run(agent_input)

Compile and run the agent.

Parameters:

Name	Type	Description	Default
agent_input	BaseModel	The input that goes in first trough the graph	required

Returns: the agent last response at the end of the graph

Source code in ebiose/backends/langgraph.py

def run(
    self,
    agent_input: BaseModel,
) -> BaseModel:
    """Compile and run the agent.

    Args:
        agent_input: The input that goes in first trough the graph

    Returns: the agent last response at the end of the graph

    """
    final_state = self.invoke_graph(agent_input)
    last_node_id = final_state["node_sequence"][-1]
    agent_output = final_state[f"{last_node_id}_responses"][-1].agent_output

    if agent_output is None:
        msg = "No agent output found."
        raise ValueError(msg)

    return agent_output

run_in_batch(agent_inputs, max_concurrency=5)

Run the agents multiple times with a batch of inputs.

Source code in ebiose/backends/langgraph.py

def run_in_batch(
    self,
    agent_inputs: list[BaseModel],
    max_concurrency: int = 5,
) -> BaseModel:
    """Run the agents multiple times with a batch of inputs."""
    self._build_state()
    compiled_graph: CompiledGraph = self.__to_compiled_graph()
    initial_states = [
        self.state(
            input=agent_input,
            shared_context_prompt=self.graph.shared_context_prompt,
        )
        for agent_input in agent_inputs
    ]

    return compiled_graph.batch(
        initial_states,
        config={"max_concurrency": max_concurrency},
    )