LangGraph
This post is mainly based on
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
Step 1. Understanding the tool
- A tool consists of several components:
name
(str): unique within a set of tools provided to an agentdescription
(str): used by an agent to determine tool useargs_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 sethandle_tool_error
accordingly
ToolException
- When
ToolException
is thrown, the agent will not stop working, but will handle the exception according to thehandle_tool_error
variable of the tool - Processing result will be returned to the agent as observation, and printed in red
- When
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
andToolException
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 aToolInvocation
and calls that tool, returning the outputToolInvocation
: 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 stateADD
to the existing attribute
In the example below, the AgentState
will only track messages
- Each node add message to
messages
- The
AgentState
is aTypedDict
- The
TypedDict
has one key:messages
messages
is aSequence[BaseMessage]
messages
is annotated byoperator.add
- The
- How
Annotated
works?Annotated
is used to specify how themessages
attribute in theAgentState
should be updated- When nodes interact with the
messages
attribute, they should perform anADD
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)