LangGraph

This post is mainly based on

• LangGraph

LangGraph

• A library for building stateful, multi-actor applications
• Built on top of LangChain
• Coordinate multiple chains (actors) across multiple steps of computation in a cyclic manner

Tools

Calling Tools with OpenAI Model

• OpenAI: Chat completion
• OpenAI: Chat completion output
• Multiple tool calls are allowed in a single response

Example

tool = {
	"type": "function",
	"function": {
		"name": "get_current_weather",
		"description": "Get the current weather in a given location",
		"parameters": {
			"type": "object",
			"properties": {
				"location": {
					"type": "string",
					"description": "The city and state, e.g. San Francisco, CA",
				},
				"unit": {
					"type": "string", 
					"enum": ["celsius", "fahrenheit"]
				},
			},
			"required": ["location"],
		},
	},
}

params = {
	'model': 'gpt-4-1106-preview',
	'stream': False,
	'n': 1,
	'temperature': 0.0,
	'tools': [tool],
}
message_dicts = [{'role': 'user', 'content': 'What is the weather in NYC today?'}]

client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
response = client.chat.completions.create(messages=message_dicts, **params)

Defining Tools with LangChain

• Documentations
  • langchain_core.tools.BaseTool
  • langchain_core.callbacks.manager.CallbackManagerForToolRun

Step 1. Understanding the tool

• A tool consists of several components:
  • name (str): unique within a set of tools provided to an agent
  • description (str): used by an agent to determine tool use
  • args_schema (Pydantic BaseModel): optional but recommended, provide information (e.g., few-shot examples) or validation for expected parameters

Step 2. The @tool decorator

• Convert a function into a tool object
• Uses the function name as the tool name by default, but this can be overridden
• Use the function’s docstring as the tool’s description - so a docstring MUST be provided
• return_direct: Whether to return the tool’s output directly. Setting this to True means that after the tool is called, the AgentExecutor will stop looping

Example 1:

# Import things that are needed generically
from langchain.pydantic_v1 import BaseModel, Field
from langchain.tools import BaseTool, StructuredTool, tool

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

Example 2:

class SearchInput(BaseModel):
    query: str = Field(description="search query", example="what's the weather today?")

@tool("search-tool", args_schema=SearchInput, return_direct=True)
def search(query: str) -> str:
    """Look up things online."""
    return "LangChain"

When you call print(search.args), the args will carry your pydantic desciption:

{
	'query': {
		'title': 'Query', 
		'description': 'search query', 
		'example': "what's the weather today?", 
		'type': 'string'
	}
}

Step 3. Formatting tools into OpenAI LLM recognizable format

from langchain.tools.render import format_tool_to_openai_tool

tools_openai = [format_tool_to_openai_tool(t) for t in tools]

Step 4. Bind tool to model

model = ChatOpenAI(model="gpt-4-1106-preview")
model = model.bind(tools=tools_openai)

Handling Tool Errors

• Behavior
  • When a tool encounters an error and the exception is not caught, the agent will stop executing
  • If you want the agent to continue execution, you can raise a ToolException and set handle_tool_error accordingly
• ToolException
  • When ToolException is thrown, the agent will not stop working, but will handle the exception according to the handle_tool_error variable of the tool
  • Processing result will be returned to the agent as observation, and printed in red
• handle_tool_error
  • Can be set to True, a unified string value, or a function
  • The function should take a ToolException as a parameter and return a str value
• You need to set both handle_tool_error and ToolException

Example:

from langchain_core.tools import ToolException

def search_tool1(s: str):
    raise ToolException("The search tool1 is not available.")

def _handle_error(error: ToolException) -> str:
    return (
        "The following errors occurred during tool execution:"
        + error.args[0]
        + "Please try another tool."
    )

search = StructuredTool.from_function(
    func=search_tool1,
    name="Search_tool1",
    description="A bad tool",
    handle_tool_error=_handle_error,
)

search.run("test")

Tool Executor

• ToolExecutor: A class that takes in a ToolInvocation and calls that tool, returning the output
• ToolInvocation: any class with tool and tool_input attribute

Example:

from langgraph.prebuilt import ToolExecutor

tools = [tool1, tool2, tool3]
tool_executor = ToolExecutor(tools)

Agent

• An Agent is StatefulGraph that consists
  • Agent State
  • Nodes
  • Edges / Conditional Edges
  • Entry Point

Agent State

• Parameterized by an object that stores the agent’s current sate and history
• Each node then returns operations to update that state
• An operation can
  • SET specific attributes on the state
  • ADD to the existing attribute

In the example below, the AgentState will only track messages

• Each node add message to messages
• The AgentState is a TypedDict
  • The TypedDict has one key: messages
  • messages is a Sequence[BaseMessage]
  • messages is annotated by operator.add
• How Annotated works?
  • Annotated is used to specify how the messages attribute in the AgentState should be updated
  • When nodes interact with the messages attribute, they should perform an ADD operation
  • Documentation

Define Agent State

from typing import TypedDict, Annotated, Sequence
import operator
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

Nodes

• A node can be
  • A function
  • A runnable
• There are two main nodes in our example
  • agent
    • Call LLM to decide what action to take
    • Defined by call_model()
  • action
    • A function to invoke tools; if the agent decides to take an action, this node will then execute that action
    • Defined by call_tool()

Define Nodes

from langgraph.prebuilt import ToolInvocation
import json
from langchain_core.messages import FunctionMessage

# Define the function that calls the model
def call_model(state):
    messages = state['messages']
    response = model.invoke(messages) # LLM make decision 
    
    return {"messages": [response]}	# To be added to `AgentState.messages`

# Define the function to execute tools
# Hitting this node implies last message involves a function call
# Construct an ToolInvocation from the function_call
def call_tool(state):
    last_message = state['messages'][-1]
    action = ToolInvocation(
        tool=last_message.additional_kwargs["function_call"]["name"],
        tool_input=json.loads(last_message.additional_kwargs["function_call"]["arguments"]),
    )

    # Call the tool_executor and get back a response
    response = tool_executor.invoke(action)

    # Use the response to create a FunctionMessage
    function_message = FunctionMessage(content=str(response), name=action.tool)

    return {"messages": [function_message]}	# To be added to `AgentState.messages`

Graph

• Created by StateGraph(AgentState)
• First Add nodes defined above, then add edges to connect nodes together

Normal Edge

• Output of the starting node will be passed to the ending node
• Arguments
  • start_key
    • A string representing the name of the start node
    • Must be already registered in the graph
  • end_key
    • A string representing the name of the end node
    • Must be already registered in the graph

Conditional Edge

• Only one of the downstream edges will be taken, depends on the results of the starting node
• Arguments
  • start_key
    • A string representing the name of the start node
    • Must be already registered in the graph
  • condition
    • A function to call to decide what to do next
    • The input will be the output of the start node
    • The function should return a string that is present in conditional_edge_mapping and represents the edge to take
  • conditional_edge_mapping
    • A mapping of string to string
    • The keys should be strings that may be returned by condition
    • The values should be the downstream node to call if that condition is returned
• With Conditional Edge, the path that is taken is not known until that node is run

Entry Point

Define which node should the agent be initialized from

END

A special node marking that the graph should finish

Define Conditional Edge

# Define the function that determines whether to continue or not
# If there is no function call, finish
# If there is a function call, continue
def should_continue(state):
    last_message = state['messages'][-1]
    
    if "function_call" not in last_message.additional_kwargs:
		return "end"
    else:
        return "continue"

Configure and Compile Graph

from langgraph.graph import StateGraph, END

# Define a new graph and two nodes to cycle between
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("action", call_tool)

# Set the entrypoint as `agent` / `agent` node is the first one called
workflow.set_entry_point("agent")

# Add conditional edge from `agent` to `continue` or `end`
# After the `agent` node made a decision, 
# We should either: "take an action" or "finish generation"
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "action",		
        "end": END, 		# The special node 
    }
)

# Add a normal edge from `tools` to `agent`
# After the tools are invoked, 
# We should always go back to the `agent` node to decide what to do next
workflow.add_edge('action', 'agent')

# This compiles the graph into a LangChain Runnable
app = workflow.compile()

inputs = {"messages": [
	HumanMessage(content="Write code to print a fibonacci sequence.")
]}
app.invoke(inputs)