langgraph/how-tos/update-state-from-tools/

[giscus[bot]]

langgraph/how-tos/update-state-from-tools/

Build language agents as graphs

https://langchain-ai.github.io/langgraph/how-tos/update-state-from-tools/

[jasontjahjono]

Is this supported for langgraph JS? I can't find the page for it

[vbarda]

yes, https://langchain-ai.github.io/langgraphjs/how-tos/update-state-from-tools/

[rayshen92]

what if the tool is a pydantic model class for structured output. how to set the update state logic (return Command())

[vbarda]

you can't do that, you need a function that explicitly returns a command

[rayshen92]

yep, i change to this way. it works.

class QueryRewriteResult(BaseModel):
    rewrited_query: str = ""
    failed: bool = False
    failed_reason: str = ""


@tool
async def generate_result(
    result: QueryRewriteResult,
    tool_call_id: Annotated[str, InjectedToolCallId],
    config: RunnableConfig,
    state: Annotated[dict, InjectedState],
):
    """
    generate result
    """

    if result.failed:
        content = f"rewrite failed, {result.failed_reason}"
    else:
        content = f"{result.rewrited_query}\n"

    return Command(
        update={
            "rewrited_query": result.rewrited_query,
            "rewrite_failed": result.failed,
            "messages": [ToolMessage(content=content, tool_call_id=tool_call_id)],
        }
    )


llm.bind_tools([generate_result])

[rayshen92]

i found an issue:
when def tools with return Command like the func lookup_user_info,
and use stream_mode = updates bleow
async for stream_mode, data in graph.astream(
inputs, config, stream_mode=["updates"]
):

when LLM call mutliple tools at once, only the last tool call message stream output, others not output.

[rayshen92]

only the last tool message stream out, others not output

[rayshen92]

#2831

create an issue

[lucamoroz]

FYI, in order to update the state from a tool by using Command, it seems that the state MUST be a subclass of AgentState.
Inheriting directly from TypedDict or MessagesState leads to 'KeyError: 'your_custom_property'`

Hopefully, this saves you some time

[vbarda]

could you actually paste an example of the state you tried creating and how you're creating the agent with create_react_agent?

you're right that you need to use AgentState (or any state that contains required keys - messages, is_last_step) but ONLY if you use create_react_agent. but the error you would see if you use non-AgentState is a ValueError, not a KeyError, e.g.

ValueError: Missing required key(s) {'messages', 'is_last_step'} in state_schema

also responded on this discussion here: #1247 (reply in thread)

[AR-BABER]

this is good, but how can i person update the graph state but doesn't let the agent know? will it also receive the updated state? as it usually receive the output of what happened in the tool, for the agent can be effective. i want the agent to recive a msg but also update the state.

[AR-BABER]

sorry, i understand now, tool message

[bryce-zhd]

Where is Command defined? I don't see how to import it.

Never mind, I found it. For anyone else wondering: from langgraph.types import Command

From: https://langchain-ai.github.io/langgraph/how-tos/update-state-from-tools/#define-tool

[AR-BABER]

yesssssssssssssss

…

On Tue, Jan 14, 2025 at 9:42 PM Corey ***@***.***> wrote: Is it possible to get state from within a tool using Langgraph JS? In Python, it looks like there's an InjectedState value but not seeing such a thing in JS. If not, i'll have to rethink how i'm utilizing the graph. — Reply to this email directly, view it on GitHub <#2806 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AUCPIWC2P45NDXCTZQ6U3RL2KU47JAVCNFSM6AAAAABTZVROVSVHI2DSMVQWIX3LMV43URDJONRXK43TNFXW4Q3PNVWWK3TUHMYTCOBTGM3TOMQ> . You are receiving this because you commented.Message ID: ***@***.*** com>

[kev52]

I use :

'''
@tool
def process_qualifications(collected_data: Dict[str, Any], tool_call_id: Annotated[str, InjectedToolCallId]):
"""Process the collected qualification checks."""

print(collected_data)

return Command(
    update={
        "step": StepEnum.EVALUATE,
        "messages": [ToolMessage(tool_call_id=tool_call_id, content=f"Interview completed.")]
    },
)

''' but get an error 1 validation error for process_qualifications
collected_data
Field required [type=missing, input_value={'tool_call_id': 'call_kZzh8C1b9GcNDrbDU7UZ51iG'}, input_type=dict]
For further information visit https://errors.pydantic.dev/2.10/v/missingTraceback (most recent call last):

[hadijaveed]

I am getting following error on multiple parallel tool calls of the same tool

{'error': {'message': "An assistant message with 'tool_calls' must be followed by tool messages responding to each 'tool_call_id'. The following tool_call_ids did not have response messages: call_jqk1z01rCP2y66a2RDlzp0wp, call_6HQn9srMsYpFIe77680aXuor", 'type': 'invalid_request_error', 'param': 'messages.[3].role', 'code': None}}

following is the way I have structured the tool

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    last_tool_call: Annotated[Optional[str], operator.add] = None


@tool
async def meeting_slots_finder(
    params: MeetingSlotFindInput,
    tool_call_id: Annotated[str, InjectedToolCallId],
    config: RunnableConfig
):

   return Command(
                update={
                    "last_tool_call": "meeting_slots_finder",
                    "messages": [
                        ToolMessage(
                            tool_call_id=tool_call_id,
                            content=', '.join(timeslots)
                        )
                    ]
                },
            )

how can I fix this above error?

[dheerajiiitv]

Following is the fix:

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
     tool_call_id: Annotated[str, InjectedToolCallId],
    last_tool_call: Annotated[Optional[str], operator.add] = None


@tool
async def meeting_slots_finder(
    params: MeetingSlotFindInput,
    tool_call_id: Annotated[str, InjectedToolCallId],
    config: RunnableConfig
):

   return Command(
                update={
                    "last_tool_call": "meeting_slots_finder",
                    "messages": [
                        ToolMessage(
                            tool_call_id=tool_call_id,
                            content=', '.join(timeslots)
                        )
                    ]
                },
            )

[eswar-sai-123]

were you able to solve in this way?
if not what was the approach you followed to solve it ?

[Simon-U]

This guide shows how you can do this using LangGraph's prebuilt components (create_react_agent / ToolNode).

Before we added a router function after the ToolNode to go to the llm or depending on the tool.
With the new Command, I would just not add a function?
And how would I use the command to indicate with the ToolNode what are the next possible nodes:

    @staticmethod
    def create_tool_node_with_fallback(tools: list) -> dict:
        return ToolNode(tools).with_fallbacks(
            [RunnableLambda(Pilot.handle_tool_error)], exception_key="error"
        )

[IllyaBrod]

I create the following function that returns Command with some state updates. When I use the async streaming like agent.astream(...) (with debug mode on and stream mode set to "messages"), I can clearly see that the ToolMessage is printed in the terminal. However, the same ToolMessage does not seem to be streamed, since I am not receiving a chunk for it. And when the retrieve the message history after this call, I also do see the ToolMessage there. This is rather a weird behavior, since I'd expect to not see the ToolMessage in the state history if it was not working correctly.

Can someone explain if I am doing something wrong?

Tool implementation

@tool("compile_landing_page")
def compile_landing_page(
  hero_section: HeaderSection,
  content_section: ContentSection,
  goals_section: GoalsSection,
  tool_call_id: Annotated[str, InjectedToolCallId],
  state: Annotated[dict, InjectedState],
):
  """
  Compile a landing page payload given various text components.
  """
  generated_page_sections: PageSections = [
    {
      "type": f"HEADER ${tool_call_id}",
      "is_hidden": hero_section.is_hidden,
      "title": hero_section.section_title,
      "description": hero_section.description,
    },
    {
      "type": "CONTENT",
      "is_hidden": content_section.is_hidden,
      "title": content_section.section_title,
      "description": content_section.description,
      # "image": content_section.image,
    },
    {
      "type": "GOALS",
      "is_hidden": goals_section.is_hidden,
      "title": goals_section.section_title,
      "list": [
        {
          # "icon": goal.icon,
          "title": goal.title,
          "description": goal.description,
        } for goal in goals_section.goals
      ],
    },
  ]

  # Get the version of the new page
  version = state.get("current_page_version", 0) + 1

  compiled_pages = state.get("compiled_pages", {})
  compiled_pages[version] = generated_page_sections

  # Then update the state and return the Command
  return Command(
    update={
       "compiled_pages": compiled_pages,
       "current_page_version": version,
       "messages": [
          ToolMessage(
            content=json.dumps(generated_page_sections),
            tool_call_id=tool_call_id,
            status="success",
          ),
        ],
    }
  )

Graph compilation

tools = [compile_landing_page]

# Initialize the language model and bind the tools
llm = ChatOpenAI(
    model="gpt-4o-mini", 
    streaming=True,
).bind_tools(tools)

memory = MemorySaver()

# Compile the agent with the checkpointer
self.graph = create_react_agent(
    model=llm,
    tools=tools,
    state_schema=State,
    checkpointer=memory,
)

Streaming (adjusted for AI SDK on FE)

async def stream_text(user_message: List[HumanMessage], config: dict):
        """
        Async generator yielding partial assistant messages and a final `done` event.
        """

        draft_tool_calls_index = -1
        draft_tool_calls = []

        async for chunk, _ in serey_agent.graph.astream(
            {"messages": user_message},
            stream_mode="messages",
            config=config,
            debug=True
        ):
            finish_reason = chunk.response_metadata.get("finish_reason")
            print(type(chunk))

            if finish_reason == "stop":
                continue
            
            if finish_reason == "tool_calls":
                for tool_call in draft_tool_calls:
                    id = tool_call["id"]
                    name = tool_call["name"]
                    args = tool_call["arguments"]

                    yield f'9:{{"toolCallId":"{id}","toolName":"{name}","args":{args}}}\n'


            if isinstance(chunk, AIMessageChunk):
                tool_calls = chunk.additional_kwargs.get("tool_calls")

                if tool_calls:
                    for tool_call in tool_calls:
                        id = tool_call.get("id")
                        function = tool_call.get("function")
                        args = function.get("arguments")
                        name = function.get("name")

                        if (id is not None):
                            draft_tool_calls_index += 1
                            draft_tool_calls.append(
                                {"id": id, "name": name, "arguments": ""})

                        else:
                            draft_tool_calls[draft_tool_calls_index]["arguments"] += args

                else:
                    yield f"0: {json.dumps(chunk.content)}\n"

            elif isinstance(chunk, ToolMessage):
                yield f'a:{{"toolCallId":"{chunk.tool_call_id}","result":{chunk.content}}}\n'

        # Some placeholder usage values
        yield 'd:{"finishReason":"stop","usage":{"promptTokens":10,"completionTokens":20}}\n'

[sharyar]

How would you go about writing tests for a tool based on this?
With non-command type tools, you would simply call .invoke with the right args. In this case, it looks like there is a lot more info that must be provided or should be done directly from a compiled graph.

[pedrorfig]

Could you show an example of updating state through a tool without using create_react_agent and ToolNode? Also, is it possible without Command?

[hinthornw]

If you're not using create_react_agent or toolnode then you're just using regular nodes, so state updates would behave the same way they do in the majoritt of our existing docs.

[pedrorfig]

Maybe I'm too new on this.
I don't know how to update state directly from a tool, without needing to send what I want to update to the tool node before. It works, but it doesn't seem right.
As an example, here is a custom tool node that I built, which gets an artifact from a tool call and updates state with it:

def sql_tool_node(state: AgentState, config: RunnableConfig):
        logger.debug("Executing SQL tools")
        outputs = []
        for tool_call in state["messages"][-1].tool_calls:
            tool_call["args"]["user_question"] = [
                msg for msg in state["messages"] if isinstance(msg, HumanMessage)
            ][-1].content
            tool_call["args"]["conversation_id"] = config.get("configurable", {}).get("conversation_id")            
            logger.debug(f"Executing tool: {tool_call['name']}")
            try:
                tool_result = sql_tools_by_name[tool_call["name"]].invoke(tool_call, config=config)
            except Exception as e:
                logger.error(f"Failed to execute tool {tool_call['name']}: {str(e)}")
                tool_result = ToolMessage(
                    content="",
                    name=tool_call["name"],
                    tool_call_id=tool_call["id"],
                    additional_kwargs={"error": e},
                )

            outputs.append(tool_result)

            sql_messages = [
                msg for msg in outputs
                if isinstance(msg, ToolMessage) and msg.name == 'db_sql_query'
            ]
            if sql_messages:
                logger.info(f"SQL query from db_sql_query retrieved")
                raw_data = sql_messages[-1].artifact
            else:
                logger.debug("No db_sql_query results found in messages")
                raw_data = []
               
        return Command(update={"messages": outputs, "raw_data": raw_data}, goto="sql agent")

[SuveenE-TF]

@hinthornw If I have a tool which a child class of BaseTool, how can I do the update?
I can't use @tool decorator as shown above. I am not sure how to get the tool_id

Any help is really appreciated!

[Arunprakaash]

is there any way to update the state inside the tool without returning

def create_candidate_interview(
    state: Annotated[dict, InjectedState]
) -> str:
    """Create an AI interview for the candidate."""
    if not state["is_interview_created"]:
        result = {'data': {'interview_link': "dummy_link"}}
        Command(update={"is_interview_created": True})
        RF_Services().addSimpleNote(
            API_KEY=state["user_key"],
            candidate=state["candidate"],
            note_content=result['data']['interview_link']
        )
    analysis = interrupt(value='waiting for interview completion.')
    if analysis:
        return f"Interview completed. Results: {analysis['insights']}"```