LanggGraph -> Must write to at least one of []

[mrctito]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

https://github.com/langchain-ai/langgraph/blob/main/examples/agent_executor/force-calling-a-tool-first.ipynb

import operator
from typing import Annotated, TypedDict, Union

from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.messages import BaseMessage


class AgentState(TypedDict):
    # The input string
    input: str
    # The list of previous messages in the conversation
    chat_history: list[BaseMessage]
    # The outcome of a given call to the agent
    # Needs `None` as a valid type, since this is what this will start as
    agent_outcome: Union[AgentAction, AgentFinish, None]
    # List of actions and corresponding observations
    # Here we annotate this with `operator.add` to indicate that operations to
    # this state should be ADDED to the existing values (not overwrite it)
    intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]

from langchain_core.agents import AgentFinish

from langgraph.prebuilt.tool_executor import ToolExecutor

# This a helper class we have that is useful for running tools
# It takes in an agent action and calls that tool and returns the result
tool_executor = ToolExecutor(tools)


# Define the agent
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 logic that will be used to determine which conditional edge to go down
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"

from langgraph.graph import END, StateGraph

# 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)
workflow.add_node("first_agent", first_agent)

# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.set_entry_point("first_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")

# After the first agent, we want to take an action
workflow.add_edge("first_agent", "action")

# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable
app = workflow.compile()

Description

Hi @dosu!

Im running this LangGraph demo:
https://github.com/langchain-ai/langgraph/blob/main/examples/agent_executor/force-calling-a-tool-first.ipynb

At this point:

    for s in self.graph.stream(inputs):
        print(list(s.values())[0])
        print("----")

I changed to:

inputs = {"input": input,
"dia_da_semana": "Segunda",
"idioma": "Português",
'email_usuario': self.email_usuario,
'data_hoje': "12/2/2024",
"chat_history": []}
result = await self.graph.ainvoke(inputs, debug=True)

And its raising this error:

Must write to at least one of []

System Info

langchain==0.2.6
langchain-community==0.2.6
langchain-core==0.2.10
langchain-experimental==0.0.62
langchain-openai==0.1.13
langchain-qdrant==0.1.0
langchain-text-splitters==0.2.1

[dosubot[bot]]

The error "Must write to at least one of []" typically indicates that the Chroma vector store is not properly configured to write to a collection. In your code, the Chroma.from_documents method is used to create a vector store, but it seems there might be an issue with the collection name or the documents being passed.

To resolve this error, ensure that the collection_name parameter is correctly specified and that the documents are properly formatted. Here is the relevant part of the code with the Chroma.from_documents method:

# Add to vectorDB
vectorstore = Chroma.from_documents(
    documents=doc_splits,
    collection_name="rag-chroma",
    embedding=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever()

Make sure that doc_splits contains the documents in the correct format and that the collection_name is a valid string. If the issue persists, you might want to check the Chroma documentation or ensure that the Chroma service is running correctly and accessible ^[1].

To continue talking to Dosu, mention @dosu.

[mrctito]

@dosu, it is not my problem. I think you made a mistake.

My error is related with LangGraph, in this offical example:

https://github.com/langchain-ai/langgraph/blob/main/examples/agent_executor/force-calling-a-tool-first.ipynb

The error is raised here:

for s in self.graph.stream(inputs):
print(list(s.values())[0])
print("----")

Thank you

[dosubot[bot]]

The error message "Must write to at least one of []" suggests that the system expects at least one output to be specified but none were provided. This issue can occur if the state dictionary does not contain the expected keys or if the workflow nodes are not correctly modifying the state.

To resolve this issue, ensure that each node in the workflow correctly updates the state dictionary with the necessary keys. Specifically, check that the retrieve function is correctly adding the documents key to the state, and that subsequent nodes are properly handling and updating the state.

Here is a snippet of the retrieve function to ensure it correctly updates the state:

def retrieve(state):
    """
    Retrieve documents

    Args:
        state (dict): The current state of the agent, including all keys.

    Returns:
        dict: New key added to state, documents, that contains documents.
    """
    print("---RETRIEVE---")
    state_dict = state["keys"]
    question = state_dict["question"]
    documents = retriever.invoke(question)
    return {"keys": {"documents": documents, "question": question}}

Ensure that the documents key is correctly added to the state dictionary. Additionally, verify that all other nodes in the workflow are correctly handling the state and updating it as needed.

If the issue persists, you may need to add debugging statements to each node to trace the state and identify where the state is not being updated correctly.

Additionally, if you are using the GraphCypherQAChain class from the LangChain library, ensure that the output_keys property is correctly set. The output_keys property should return a list containing the expected output keys. Here is an example:

class GraphCypherQAChain(Chain):
    # ... other properties and methods ...

    @property
    def output_keys(self) -> List[str]:
        """Return the output keys.

        :meta private:
        """
        _output_keys = [self.output_key]
        return _output_keys

Make sure that the output_key attribute is set to the expected output key name. This will ensure that the system knows which keys to expect in the output ^[1][2].

To continue talking to Dosu, mention @dosu.

[mrctito]

@dosu, this is not my error. I didnt ask you that.

My error is related with LangGraph, in this offical example:

https://github.com/langchain-ai/langgraph/blob/main/examples/agent_executor/force-calling-a-tool-first.ipynb

The error is raised here:

for s in self.graph.stream(inputs):
print(list(s.values())[0])
print("----")

Thank you

[dosubot[bot]]

To resolve the "Must write to at least one of []" error when running the LangGraph demo from the "force-calling-a-tool-first.ipynb" notebook in the "agent_executor" folder of the LangGraph GitHub repository, you need to ensure that the agent is configured correctly with the necessary tools and prompt. Here is an example setup based on the provided context:

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate

# Define the model
model = ChatOpenAI(model="gpt-4")

# Define the tool
@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    return input + 2

tools = [magic_function]

# Define the prompt
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ]
)

# Create the agent
agent = create_tool_calling_agent(model, tools, prompt)

# Create the agent executor
agent_executor = AgentExecutor(agent=agent, tools=tools)

# Define the query
query = "what is the value of magic_function(3)?"

# Invoke the agent executor
result = agent_executor.invoke({"input": query})
print(result)

Ensure that the tools list is not empty and that the prompt is correctly defined with the necessary placeholders. This setup should help you avoid the "Must write to at least one of []" error. Additionally, make sure you have installed LangChain and set up any necessary environment variables as described in the LangChain documentation ^[1][2].

To continue talking to Dosu, mention @dosu.

[mrctito]

Hello @dosu,

The example I mentioned is different from your instructions. It uses Langgraph, and the issue occurs when I execute app.graph.ainvoke.

Your example works very well but does not solve my problem.

Please check the exact source code of the example.
...
graph_builder = StateGraph(StateGraph)
...
graph = graph_builder.compile()
...
result = await self.graph.ainvoke(inputs, debug=True)
error= Must write to at least one of []

Thank you.

[dosubot[bot]]

The error message "Must write to at least one of []" suggests that the StateGraph is not properly configured with the required nodes or edges. Specifically, it seems that the graph might be missing the necessary output nodes or the state is not being updated correctly.

Here is a corrected version of your code:

from langgraph.graph import END, StateGraph

# 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_tools)
workflow.add_node("first_agent", first_agent)

# Set the entrypoint as `first_agent`
workflow.set_entry_point("first_agent")

# Add conditional edges
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "action",
        "end": END,
    },
)

# Add normal edges
workflow.add_edge("action", "agent")
workflow.add_edge("first_agent", "action")

# Compile the workflow
app = workflow.compile()

# Define inputs
inputs = {
    "messages": [
        HumanMessage(
            "what's 3 plus 5 raised to the 2.743. also what's 17.24 - 918.1241"
        )
    ],
    "dia_da_semana": "Segunda",
    "idioma": "Português",
    'email_usuario': self.email_usuario,
    'data_hoje': "12/2/2024",
    "chat_history": []
}

# Invoke the graph
result = await app.ainvoke(inputs, debug=True)

Key changes:

1. Ensure that the app object is used correctly to invoke the graph.
2. Make sure that the StateGraph is properly configured with all necessary nodes and edges.
3. Use call_model and call_tools functions as defined in the new context.
4. Update the input format to match the expected structure, particularly the messages key.

If the issue persists, please verify that all nodes (call_model, call_tools, first_agent) and the should_continue function are correctly defined and implemented. Additionally, ensure that the ainvoke method is correctly implemented and that the input arguments are properly parsed and validated ^[1][2][3].

To continue talking to Dosu, mention @dosu.

[gpiyush2532]

Hey,
I was trying to build customer support agent on a different use case for production , hosted on FAST API, I am stucked with the error " Must write to at least one of [] ". I am Invoking graph wit '''result = abot.graph_compiler.invoke({"messages": ("user",messages)},thread)'''

My Graph is similar to :
builder = StateGraph(State)
#Adding Get Order Info Node
builder.add_node("fetch_user_info", self.user_info)
builder.set_entry_point("fetch_user_info")

#Adding Delivery Preference Asistant
builder.add_node(

"enter_delivery_preference",
create_entry_node("Delivery Preference Update Assistant", "delivery_preference"),

)
builder.add_node("delivery_preference", Assistant(delivery_preference_runnable))
builder.add_edge("enter_delivery_preference", "delivery_preference")
builder.add_node(
"delivery_preference_sensitive_tools",
create_tool_node_with_fallback(delivery_preference_sensitive_tools),
)
builder.add_node(
"delivery_preference_safe_tools",
create_tool_node_with_fallback(delivery_preference_safe_tools),
)
builder.add_edge("delivery_preference_sensitive_tools", "delivery_preference")
builder.add_edge("delivery_preference_safe_tools", "delivery_preference")
builder.add_conditional_edges("delivery_preference", self.route_delivery_preference)
builder.add_node("leave_skill", self.pop_dialog_state)
builder.add_edge("leave_skill", "primary_assistant")

#Adding Update Buyer Info Assistant

builder.add_node(
    "enter_update_buyer_info",
    create_entry_node("Update Buyer Info Assistant", "update_buyer_info"),
)
builder.add_node("update_buyer_info", Assistant(update_buyer_info_runnable))
builder.add_edge("enter_update_buyer_info", "update_buyer_info")
builder.add_node(
    "update_buyer_info_safe_tools",
    create_tool_node_with_fallback(update_buyer_info_safe_tools),
)
builder.add_node(
    "update_buyer_info_sensitive_tools",
    create_tool_node_with_fallback(update_buyer_info_sensitive_tools),
)

builder.add_edge("update_buyer_info_sensitive_tools", "update_buyer_info")
builder.add_edge("update_buyer_info_safe_tools", "update_buyer_info")
builder.add_conditional_edges("update_buyer_info", self.route_update_buyer_info)

# Create Escalation assistant
builder.add_node(
    "enter_create_escalation", create_entry_node("Escalation Creation Agent", "create_escalation")
)
builder.add_node("create_escalation", Assistant(create_escalation_runnable))
builder.add_edge("enter_create_escalation", "create_escalation")
builder.add_node(
    "create_escalation_safe_tools",
    create_tool_node_with_fallback(create_escalation_safe_tools),
)
builder.add_node(
    "create_escalation_sensitive_tools",
    create_tool_node_with_fallback(create_escalation_sensitive_tools),
)

builder.add_edge("create_escalation_sensitive_tools", "create_escalation")
builder.add_edge("create_escalation_safe_tools", "create_escalation")
builder.add_conditional_edges("create_escalation", self.route_create_escalation)

# Cancellation assistant
builder.add_node(
    "enter_cancel_order",
    create_entry_node("Cancel Order Assistant", "cancel_order"),
)
builder.add_node("cancel_order", Assistant(cancel_order_runnable))
builder.add_edge("enter_cancel_order", "cancel_order")
builder.add_node(
    "cancel_order_safe_tools",
    create_tool_node_with_fallback(cancel_order_safe_tools),
)
builder.add_node(
    "cancel_order_sensitive_tools",
    create_tool_node_with_fallback(cancel_order_sensitive_tools),
)
builder.add_edge("cancel_order_sensitive_tools", "cancel_order")
builder.add_edge("cancel_order_safe_tools", "cancel_order")
builder.add_conditional_edges("cancel_order", self.route_cancel_order)

# Primary assistant
builder.add_node("primary_assistant", Assistant(assistant_runnable))
builder.add_node(
    "primary_assistant_tools", create_tool_node_with_fallback(primary_assistant_tools)
)
# The assistant can route to one of the delegated assistants,
# directly use a tool, or directly respond to the user
builder.add_conditional_edges(
    "primary_assistant",
    self.route_primary_assistant,
    {
        "enter_delivery_preference": "enter_delivery_preference",
        "enter_update_buyer_info": "enter_update_buyer_info",
        "enter_create_escalation": "enter_create_escalation",
        "enter_cancel_order": "enter_cancel_order",
        "primary_assistant_tools": "primary_assistant_tools",
        END: END,
    },
)
builder.add_edge("primary_assistant_tools", "primary_assistant")





builder.add_conditional_edges("fetch_user_info", self.route_to_workflow)

# Compile graph
memory = SqliteSaver.from_conn_string(":memory:")
part_4_graph = builder.compile(
    checkpointer=memory,
    # Let the user approve or deny the use of sensitive tools
    interrupt_before=[
        "delivery_preference_sensitive_tools",
        "update_buyer_info_sensitive_tools",
        "create_escalation_sensitive_tools",
        "cancel_order_sensitive_tools",
    ],
)
self.graph_compiler=part_4_graph

Please help!

[HoaNQ98]

Hi @gpiyush2532 ,
What is your State designed?

[HoaNQ98]

Hi @mrctito ,
What is your input?
I guess you should put: {'input': ......} instead of a list inputs

input = {"input": "your input"}

for s in self.graph.stream(input):
        print(list(s.values())[0])
        print("----")