Im don’t work for langchain, so please qualify my response
Your graph is standard react style setup so
llm <—> tool_call
so once a tool is call the control goes back to the node which calls the llm to decide what tool to call next.
And remember at the end of the day llm input is text not a pd.DataFrame. So you have a few options
the most obvious: have a single tool process data that takes in the file location, reads it AND does the processing
have the read_file tool read the csv and return the datafram flattened as a dict. This will append the response to messages (as text), which gets sent to the llm. Them rely on the llm to extract the arg from messages to form up the second tool call
be very explicit and have read_file return a Command which updates a specific state variable
Thank you @darthShana for helping out, your suggestions on using InjectedState are correct!
@IamExperimenting you were almost there! I noticed a few syntax errors but probably due to copy pasting, you just need to create a new field in the State that contains the uploaded data.
Here’s a full example to get started (parallel_tool_calls=False avoids process data tool to be called at the same time as read_file):
from langchain.chat_models import init_chat_model
from langgraph.prebuilt import ToolNode, InjectedState
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import ToolMessage
from langgraph.types import Command
from typing import Annotated
from langchain_core.tools import InjectedToolCallId
import pandas as pd
from dotenv import load_dotenv
load_dotenv()
class State(MessagesState):
input_data: pd.DataFrame
def read_file(file_location: str, tool_call_id: Annotated[str, InjectedToolCallId]):
"""
This function read a csv file from a given location, serializes it to a JSON string, and returns it.
Args:
file_location: The location of the csv file to read.
Returns:
A JSON string of the dataframe.
"""
df_ = pd.read_csv(file_location)
state_update = {
"messages": [ToolMessage(content="loaded csv file", tool_call_id=tool_call_id)],
"input_data": df_
}
return Command(update=state_update)
def process_data(state: Annotated[dict, InjectedState], tool_call_id: Annotated[str, InjectedToolCallId]):
"""
This function processes the dataframe from the state.
Returns:
The processed DataFrame or a Command with error message if processing fails
"""
try:
output_data = state['input_data']
print(output_data.head())
return output_data
except Exception as e:
print(f"Error processing data: {e}")
return Command(update={"messages": [ToolMessage(content=f"Error processing data: {e}", tool_call_id=tool_call_id)]})
tools = [read_file, process_data]
tool_node = ToolNode(tools)
model = init_chat_model(model="gpt-4o-mini", parallel_tool_calls=False)
model_with_tools = model.bind_tools(tools)
def should_continue(state: State):
messages = state["messages"]
last_message = messages[-1]
if last_message.tool_calls:
return "tools"
return END
def call_model(state: State):
messages = state["messages"]
response = model_with_tools.invoke(messages)
return {"messages": [response]}
builder = StateGraph(State)
# Define the two nodes we will cycle between
builder.add_node("call_model", call_model)
builder.add_node("tools", tool_node)
builder.add_edge(START, "call_model")
builder.add_conditional_edges("call_model", should_continue, ["tools", END])
builder.add_edge("tools", "call_model")
graph = builder.compile()
graph.invoke({"messages": [{"role": "user", "content": "read the file /Users/mperini/Projects/agents/examples/sample_data.csv and process it"}]})
