Some tool function should be wait when using astream_event

[Lin-jun-xiang]

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 👆

Description

I have used astream_event to completed my agent.

If I have tools = [t1, t2, t3], and the agent flow like t3->t1->t2.

The t2 should be waiting for t1 completed, or the t2 will not execute correctly.

Could I use some condition to identity when should wait?

[dosubot[bot]]

Yes, you can use a condition to ensure that t2 waits for t1 to complete when using astream_event. Here is an example of how you might achieve this:

1. Define your tools:

   from langchain_core.tools import tool
   
   @tool
   def t1(input: Any) -> Any:
       # Implementation of t1
       pass
   
   @tool
   def t2(input: Any) -> Any:
       # Implementation of t2
       pass
   
   @tool
   def t3(input: Any) -> Any:
       # Implementation of t3
       pass
   
   tools = [t1, t2, t3]

2. Create an agent and use astream_event:

   from langchain import hub
   from langchain.agents import AgentExecutor, create_tool_calling_agent
   from langchain_openai.chat_models import ChatOpenAI
   
   llm = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0)
   prompt = hub.pull("hwchase17/openai-tools-agent")
   agent = create_tool_calling_agent(llm, tools, prompt)
   agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

3. Use astream_event to manage dependencies:

   async def run_agent():
       async for event in agent_executor.astream_event(
           "Your input query here",
           version="v2",
           include_names=["t1", "t2", "t3"]
       ):
           if event.name == "t1" and event.status == "completed":
               # t1 has completed, now t2 can proceed
               await agent_executor.invoke({"input": "Input for t2"})
           elif event.name == "t2" and event.status == "completed":
               # t2 has completed, now t3 can proceed
               await agent_executor.invoke({"input": "Input for t3"})
           print(event)
   
   import asyncio
   asyncio.run(run_agent())

In this example, astream_event is used to monitor the completion of each tool. The if conditions ensure that t2 only starts after t1 has completed, and t3 starts after t2 has completed. This way, you can manage the dependencies between your tools effectively ^[1][2].

To continue talking to Dosu, mention @dosu.

[Lin-jun-xiang]

   tools = [t1, t2, t3] # each tool have no input parameter.

    agent = create_openai_tools_agent(langchain_llm_client, tools, AGENT_PROMPT)
    agent_executor = AgentExecutor(
        agent=agent,
        tools=tools,
        verbose=False,
        return_intermediate_steps=True
    )


    async for event in agent_executor.astream_events(
        {
            "input": formatted_history_query,
            "tools": tools,
            "tool_names": tool_names,
            "agent_scratchpad": lambda x: format_to_openai_tool_messages(x["intermediate_steps"]),
        },
        version='v2'
    ):
        kind = event['event']
        if kind == "on_chain_start":
            if (
                event["name"] == "Agent"
            ):
                yield(
                    f"\n### 執行代理: `{event['name']}`，代理輸入: `{event['data'].get('input')}`\n"
                )
        elif kind == "on_chat_model_stream":
            content = event["data"]["chunk"].content
            if content:
                yield content
        elif kind == "on_tool_start":
            yield(
                f"\n### 執行任務: `{event['name']}`，任務輸入: `{event['data'].get('input')}`\n"
            )
        elif kind == "on_tool_end":
            yield(
                f"\n### 任務完成: `{event['name']}`，任務結果: \n"
            )
            if isinstance(event['data'].get('output'), AsyncGenerator):
                async for event_chunk in event['data'].get('output'):
                    yield event_chunk
            else:
                yield(
                    f"`{event['data'].get('output')}`\n"
                )
        elif kind == "on_chain_end":
            if (
                event["name"] == "Agent"
            ):
                yield(
                    f"\n### 代理完成: `{event['name']}`，代理結果: \n"
                )
                yield(
                    f"{event['data'].get('output')['output']}\n"
                )

How should I fix?

[dosubot[bot]]

To ensure that t2 waits for t1 to complete before executing, and t3 waits for t2 to complete, you can modify your code to include conditions that check the completion status of each tool. Here's how you can adjust your code:

1. Define a state to track the completion of each tool:

   tool_status = {
       "t1": False,
       "t2": False,
       "t3": False
   }

2. Modify the astream_events loop to handle dependencies:

   async for event in agent_executor.astream_events(
       {
           "input": formatted_history_query,
           "tools": tools,
           "tool_names": tool_names,
           "agent_scratchpad": lambda x: format_to_openai_tool_messages(x["intermediate_steps"]),
       },
       version='v2'
   ):
       kind = event['event']
       if kind == "on_chain_start":
           if event["name"] == "Agent":
               yield f"\n### 執行代理: `{event['name']}`，代理輸入: `{event['data'].get('input')}`\n"
       elif kind == "on_chat_model_stream":
           content = event["data"]["chunk"].content
           if content:
               yield content
       elif kind == "on_tool_start":
           yield f"\n### 執行任務: `{event['name']}`，任務輸入: `{event['data'].get('input')}`\n"
       elif kind == "on_tool_end":
           yield f"\n### 任務完成: `{event['name']}`，任務結果: \n"
           if isinstance(event['data'].get('output'), AsyncGenerator):
               async for event_chunk in event['data'].get('output'):
                   yield event_chunk
           else:
               yield f"`{event['data'].get('output')}`\n"
           
           # Update tool status
           tool_status[event["name"]] = True
   
           # Check dependencies and proceed accordingly
           if event["name"] == "t1" and tool_status["t1"]:
               # t1 has completed, now t2 can proceed
               await agent_executor.invoke({"input": "Input for t2"})
           elif event["name"] == "t2" and tool_status["t2"]:
               # t2 has completed, now t3 can proceed
               await agent_executor.invoke({"input": "Input for t3"})
       elif kind == "on_chain_end":
           if event["name"] == "Agent":
               yield f"\n### 代理完成: `{event['name']}`，代理結果: \n"
               yield f"{event['data'].get('output')['output']}\n"

In this modified code:

• tool_status is used to track whether each tool (t1, t2, t3) has completed.
• The on_tool_end event updates the status of the completed tool.
• Conditions check the status of each tool and invoke the next tool in the sequence once the previous tool has completed.

This ensures that t2 waits for t1 to complete, and t3 waits for t2 to complete before executing.

To continue talking to Dosu, mention @dosu.