cr

langgraph/prebuilt/agent_executor.py

@@ -0,0 +1,90 @@
+from typing import Annotated, TypedDict
+import operator
+from langchain_core.agents import AgentAction, AgentFinish
+from langgraph.graph import StateGraph, END
+from langgraph.prebuilt.tool_executor import ToolExecutor
+
+
+
+
+def create_agent_executor(agent_runnable, tools, input_schema=None):
+
+    if isinstance(tools, ToolExecutor):
+        tool_executor = tools
+    else:
+        tool_executor = ToolExecutor(tools)
+
+
+    if input_schema is None:
+        class AgentState(TypedDict):
+            input: str
+            agent_outcome: AgentAction | AgentFinish | None
+            intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]
+
+    else:
+        class AgentState(input_schema):
+            agent_outcome: AgentAction | AgentFinish | None
+            intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]
+
+    def should_continue(data):
+        # If the agent outcome is an AgentFinish, then we return `exit` string
+        # This will be used when setting up the graph to define the flow
+        if isinstance(data['agent_outcome'], AgentFinish):
+            return "end"
+        # Otherwise, an AgentAction is returned
+        # Here we return `continue` string
+        # This will be used when setting up the graph to define the flow
+        else:
+            return "continue"
+
+    def run_agent(data):
+        agent_outcome = agent_runnable.invoke(data)
+        return {"agent_outcome": agent_outcome}
+
+    # Define the function to execute tools
+    def execute_tools(data):
+        # Get the most recent agent_outcome - this is the key added in the `agent` above
+        agent_action = data['agent_outcome']
+        output = tool_executor.invoke(agent_action)
+        return {"intermediate_steps": [(agent_action, str(output))]}
+
+    # Define a new graph
+    workflow = StateGraph(AgentState)
+
+    # Define the two nodes we will cycle between
+    workflow.add_node("agent", run_agent)
+    workflow.add_node("action", execute_tools)
+
+    # Set the entrypoint as `agent`
+    # This means that this node is the first one called
+    workflow.set_entry_point("agent")
+
+    # We now add a conditional edge
+    workflow.add_conditional_edges(
+        # First, we define the start node. We use `agent`.
+        # This means these are the edges taken after the `agent` node is called.
+        "agent",
+        # Next, we pass in the function that will determine which node is called next.
+        should_continue,
+        # Finally we pass in a mapping.
+        # The keys are strings, and the values are other nodes.
+        # END is a special node marking that the graph should finish.
+        # What will happen is we will call `should_continue`, and then the output of that
+        # will be matched against the keys in this mapping.
+        # Based on which one it matches, that node will then be called.
+        {
+            # If `tools`, then we call the tool node.
+            "continue": "action",
+            # Otherwise we finish.
+            "end": END
+        }
+    )
+
+    # We now add a normal edge from `tools` to `agent`.
+    # This means that after `tools` is called, `agent` node is called next.
+    workflow.add_edge('action', 'agent')
+
+    # Finally, we compile it!
+    # This compiles it into a LangChain Runnable,
+    # meaning you can use it as you would any other runnable
+    return workflow.compile()

langgraph/prebuilt/messages_executor.py

@@ -0,0 +1,122 @@
+from langchain_core.runnables import RunnablePassthrough
+from langchain_core.messages import FunctionMessage
+from langchain_core.agents import AgentFinish, AgentAction
+import json
+
+from langchain.tools.render import format_tool_to_openai_function
+from langgraph.prebuilt.tool_executor import ToolExecutor
+from langchain_core.utils.function_calling import convert_pydantic_to_openai_function
+from typing import Annotated, TypedDict, Sequence
+from langchain_core.messages import BaseMessage
+import operator
+from langchain_core.agents import AgentAction, AgentFinish
+from langgraph.graph import StateGraph, END
+
+
+def _get_tool_executor_and_functions(tools, response_format):
+    if isinstance(tools, ToolExecutor):
+        tool_executor = tools
+        tool_classes = tools.tools
+    else:
+        tool_executor = ToolExecutor(tools)
+        tool_classes = tools
+
+    functions = [format_tool_to_openai_function(t) for t in tool_classes]
+    if response_format is not None:
+        functions.append(convert_pydantic_to_openai_function(response_format))
+    return tool_executor, functions
+
+
+def create_messages_executor(model, tools, response_format = None):
+    tool_executor, functions = _get_tool_executor_and_functions(tools, response_format)
+    model = model.bind_functions([format_tool_to_openai_function(t) for t in tools])
+
+    # Define the function that determines whether to continue or not
+    def should_continue(state):
+        messages = state['messages']
+        last_message = messages[-1]
+        # If there is no function call, then we finish
+        if "function_call" not in last_message.additional_kwargs:
+            return "end"
+        # Otherwise if there is, we need to check what type of function call it is
+        else:
+            if response_format is None:
+                return "continue"
+            elif last_message.additional_kwargs["function_call"]["name"] == response_format.__name__:
+                return "end"
+            else:
+                return "continue"
+
+    # Define the function that calls the model
+    def call_model(state):
+        messages = state['messages']
+        response = model.invoke(messages)
+        # We return a list, because this will get added to the existing list
+        return {"messages": [response]}
+
+    # Define the function to execute tools
+    def call_tool(state):
+        messages = state['messages']
+        # Based on the continue condition
+        # we know the last message involves a function call
+        last_message = messages[-1]
+        # We construct an AgentAction from the function_call
+        action = AgentAction(
+            tool=last_message.additional_kwargs["function_call"]["name"],
+            tool_input=json.loads(last_message.additional_kwargs["function_call"]["arguments"]),
+            log="",
+        )
+        # We call the tool_executor and get back a response
+        response = tool_executor.invoke(action)
+        # We use the response to create a FunctionMessage
+        function_message = FunctionMessage(content=str(response), name=action.tool)
+        # We return a list, because this will get added to the existing list
+        return {"messages": [function_message]}
+
+    # We create the AgentState that we will pass around
+    # This simply involves a list of messages
+    # We want steps to return messages to append to the list
+    # So we annotate the messages attribute with operator.add
+    class AgentState(TypedDict):
+        messages: Annotated[Sequence[BaseMessage], operator.add]
+
+    # Define a new graph
+    workflow = StateGraph(AgentState)
+
+    # Define the two nodes we will cycle between
+    workflow.add_node("agent", call_model)
+    workflow.add_node("action", call_tool)
+
+    # Set the entrypoint as `agent`
+    # This means that this node is the first one called
+    workflow.set_entry_point("agent")
+
+    # We now add a conditional edge
+    workflow.add_conditional_edges(
+        # First, we define the start node. We use `agent`.
+        # This means these are the edges taken after the `agent` node is called.
+        "agent",
+        # Next, we pass in the function that will determine which node is called next.
+        should_continue,
+        # Finally we pass in a mapping.
+        # The keys are strings, and the values are other nodes.
+        # END is a special node marking that the graph should finish.
+        # What will happen is we will call `should_continue`, and then the output of that
+        # will be matched against the keys in this mapping.
+        # Based on which one it matches, that node will then be called.
+        {
+            # If `tools`, then we call the tool node.
+            "continue": "action",
+            # Otherwise we finish.
+            "end": END
+        }
+    )
+
+    # We now add a normal edge from `tools` to `agent`.
+    # This means that after `tools` is called, `agent` node is called next.
+    workflow.add_edge('action', 'agent')
+
+    # Finally, we compile it!
+    # This compiles it into a LangChain Runnable,
+    # meaning you can use it as you would any other runnable
+    return workflow.compile()

langgraph/prebuilt/tool_executor.py

@@ -0,0 +1,40 @@
+from langchain_core.runnables import RunnableBinding, RunnableLambda
+from typing import Sequence, Any
+from langchain_core.tools import BaseTool
+from langchain_core.agents import AgentAction
+INVALID_TOOL_MSG_TEMPLATE = (
+    "{requested_tool_name} is not a valid tool, "
+    "try one of [{available_tool_names_str}]."
+)
+
+class ToolExecutor(RunnableBinding):
+
+    tools: Sequence[BaseTool]
+    tool_map: dict
+    invalid_tool_msg_template: str
+    def __init__(self, tools: Sequence[BaseTool], invalid_tool_msg_template: str = INVALID_TOOL_MSG_TEMPLATE, **kwargs: Any) -> None:
+
+        bound = RunnableLambda(self._execute, afunc=self._aexecute)
+        super().__init__(bound=bound, tools=tools, tool_map ={t.name: t for t in tools}, invalid_tool_msg_template=invalid_tool_msg_template, **kwargs)
+
+    def _execute(self, tool_invocation: AgentAction) -> Any:
+        if tool_invocation.tool not in self.tool_map:
+            return self.invalid_tool_msg_template.format(
+                requested_tool_name=tool_invocation.tool,
+                available_tool_names_str=", ".join([t.name for t in self.tools])
+            )
+        else:
+            tool = self.tool_map[tool_invocation.tool]
+            output = tool.invoke(tool_invocation.tool_input)
+            return output
+
+    async def _aexecute(self, tool_invocation: AgentAction) -> Any:
+        if tool_invocation.tool not in self.tool_map:
+            return self.invalid_tool_msg_template.format(
+                requested_tool_name=tool_invocation.tool,
+                available_tool_names_str=", ".join([t.name for t in self.tools])
+            )
+        else:
+            tool = self.tool_map[tool_invocation.tool]
+            output = await tool.ainvoke(tool_invocation.tool_input)
+            return output