how could agent step by step with streaming?

[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 👆

Example Code

@tool
async def tool1():
    """tool1"""
    print('start1...')
    cmd = {
        'cmd': 'Scrape',
    }
    await asyncio.sleep(5) # do something need long time.
    go2_command_queue.put(cmd)
    print('time1....')


@tool
async def tool2():
    """tool2"""
    print('start2...')
    print(go2_command_queue.qsize())
    print('time2....')


async def agent_completion(
    agent_executor,
    message: str,
    tools: List = None,
) -> AsyncGenerator:
    """Base on query to decide the tool which should use.
    Response with `async` and `streaming`.
    """
    tool_names = [tool.name for tool in tools]


    async for event in agent_executor.astream_events(
        {
            "input": message,
            "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### Agent: `{event['name']}`，Agent Input: `{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### Tool: `{event['name']}`，Tool Input: `{event['data'].get('input')}`\n"
            )
        elif kind == "on_tool_end":
            yield(
                f"\n### Tool Finished: `{event['name']}`，Tool Results: \n"
            )
            yield(
                f"`{event['data'].get('output')}`\n"
            )
        elif kind == "on_chain_end":
            if (
                event["name"] == "Agent"
            ):
                yield(
                    f"\n### Agent Finished: `{event['name']}`，Agent Results: \n"
                )
                yield(
                    f"{event['data'].get('output')['output']}\n"
                )


if __name__ == '__main__':
    llm = ChatOpenAI(
        model='gpt-4o',
        temperature=0.,
        api_key=OPENAI_API_KEY,
        streaming=True,
        max_tokens=None,
    )
    tools = [
        too1,
        tool2,
        tool3,
        ...
    ]
    agent = create_openai_tools_agent(llm, tools, AGENT_PROMPT)
    agent_executor = AgentExecutor(
        agent=agent,
        tools=tools,
        verbose=False,
        return_intermediate_steps=True
    )

    while True:
        user_message = input('Query: ') # tool1 then tool2
        responses = agent_completion(agent_executor, user_message, tools)
        async for response in responses:
            print(response, flush=True)

Results

### Tool: `tool1`，Tool Input: `{}`

start1...
start2...
0
time2...

### Tool: `tool2`，Tool Input: `{}`

time1....

### Tool Finished: `tool1`，Tool Results: 

`None`


### Tool Finished: `tool2`，Tool Results: 

`None`

Description

If some of my tools have dependencies, for example: tool3 needs to wait for tool2 to complete before it can be executed. Is there a way to achieve this? Through astream_event, I noticed that different tools are executed almost concurrently in an asynchronous manner.

I also tried using the .stream method. Although this method can execute tools step by step, the task process is not output in a streaming manner, which is not what I want.

System Info

System Information

OS: Windows
OS Version: 10.0.19045
Python Version: 3.11.8 (tags/v3.11.8:db85d51, Feb 6 2024, 22:03:32) [MSC v.1937 64 bit (AMD64)]

Package Information

langchain_core: 0.2.10
langchain: 0.2.6
langchain_community: 0.2.1
langsmith: 0.1.82
langchain_text_splitters: 0.2.0
langchainhub: 0.1.20

Packages not installed (Not Necessarily a Problem)

The following packages were not found:

langgraph
langserve

[dosubot[bot]]

To ensure that tools with dependencies are executed in the correct order using astream_events in an asynchronous manner, you can use the RunnableSequence to chain the tools together. The RunnableSequence ensures that each tool in the sequence is executed in the correct order, and the events are collected accordingly.

Here is an example of how you can achieve this:

async def test_event_stream_with_simple_function_tool() -> None:
    """Test the event stream with a function and tool"""

    def foo(x: int) -> dict:
        """Foo"""
        return {"x": 5}

    @tool
    def get_docs(x: int) -> List[Document]:
        """Hello Doc"""
        return [Document(page_content="hello")]

    chain = RunnableLambda(foo) | get_docs
    events = await _collect_events(chain.astream_events({}, version="v2"))
    assert events == [
        {
            "event": "on_chain_start",
            "run_id": "",
            "parent_ids": [],
            "name": "RunnableSequence",
            "tags": [],
            "metadata": {},
            "data": {"input": {}},
        },
        {
            "event": "on_chain_start",
            "name": "foo",
            "run_id": "",
            "parent_ids": [],
            "tags": ["seq:step:1"],
            "metadata": {},
            "data": {},
        },
        {
            "event": "on_chain_stream",
            "name": "foo",
            "run_id": "",
            "parent_ids": [],
            "tags": ["seq:step:1"],
            "metadata": {},
            "data": {"chunk": {"x": 5}},
        },
        {
            "event": "on_chain_end",
            "name": "foo",
            "run_id": "",
            "parent_ids": [],
            "tags": ["seq:step:1"],
            "metadata": {},
            "data": {"input": {}, "output": {"x": 5}},
        },
        {
            "event": "on_tool_start",
            "name": "get_docs",
            "run_id": "",
            "parent_ids": [],
            "tags": ["seq:step:2"],
            "metadata": {},
            "data": {"input": {"x": 5}},
        },
        {
            "event": "on_tool_end",
            "name": "get_docs",
            "run_id": "",
            "parent_ids": [],
            "tags": ["seq:step:2"],
            "metadata": {},
            "data": {"input": {"x": 5}, "output": [Document(page_content="hello")]},
        },
        {
            "event": "on_chain_stream",
            "run_id": "",
            "parent_ids": [],
            "tags": [],
            "metadata": {},
            "name": "RunnableSequence",
            "data": {"chunk": [Document(page_content="hello")]},
        },
        {
            "event": "on_chain_end",
            "name": "RunnableSequence",
            "run_id": "",
            "parent_ids": [],
            "tags": [],
            "metadata": {},
            "data": {"output": [Document(page_content="hello")]},
        },
    ]

In this example, the foo function and the get_docs tool are chained together using RunnableLambda(foo) | get_docs. The astream_events method is then used to execute the chain asynchronously and collect the events in the correct order. The events are asserted to ensure that they are executed in the expected sequence ^[1][2].

To continue talking to Dosu, mention @dosu.