How to use agent executor astream_events?

[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 describe_image(query: str):
    """Describe the image which is looking(focus)."""
    global snapshot_session
    prompt = f"""
    Describe the image.
    """
    if selected_api == 'OpenAI':
        message=[
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt},
                    {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{snapshot_session}"}}
                ]
            }
        ]
        response = chat_completion(openai_client, message, [], client_name=selected_api)
        print('\033[36mCalled gpt4o api to check image successfully.\033[0m')

    # return response

langchain_llm_client = ChatOpenAI(
    model='gpt-4o',
    temperature=0.,
    api_key=OPENAI_API_KEY,
    streaming=True,
    max_tokens=None,
)
tools = [describe_image, ...]
agent = create_tool_calling_agent(langchain_llm_client, tools, AGENT_PROMPT)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=False,
    return_intermediate_steps=True
)

async def chat(messages):
    tool_names = [tool.name for tool in tools]
    async for event in agent_executor.astream_events(
        {
            "input": messages,
            "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### Start Agent: `{event['name']}`, Agent Input: `{event['data'].get('input')}`\n"
                )
        elif kind == "on_chat_model_stream":
            # llm model response
            content = event["data"]["chunk"].content
            if content:
                yield content
        elif kind == "on_tool_start":
            yield(
                f"\n### Start Tool: `{event['name']}`, Tool Input: `{event['data'].get('input')}`\n"
            )
        elif kind == "on_tool_end":
            yield(
                f"\n### Finished Tool: `{event['name']}`, Tool Results: \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### Finished Agent: `{event['name']}`, Agent Results: \n"
                )
                yield(
                    f"{event['data'].get('output')['output']}\n"
                )

if __name__ == '__main__':
    asyncio.run(chat(['describe'])

Description

When I use astream_event, the tool I've defined calls the OpenAI API to describe an image.

Therefore, I want the description process of this tool to output in a streaming manner.

However, I encountered the following two issues:

• The describe_image tool is called repeatedly, leading to duplicate descriptions.
• The next tool runs before the describe_image tool completes its description, but I want the next tool to wait until the image description is finished before executing.

Due to confidentiality reasons, only pseudocode can be displayed.

System Info

System Information

OS: Windows
OS Version: 10.0.22631
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.23
langchain: 0.2.11
langchain_community: 0.2.10
langsmith: 0.1.86
langchain_anthropic: 0.1.20
langchain_openai: 0.1.17
langchain_text_splitters: 0.2.2

Packages not installed (Not Necessarily a Problem)

The following packages were not found:

langgraph
langserve

[dosubot[bot]]

To use the astream_events method in the AgentExecutor class to ensure that the describe_image tool outputs in a streaming manner without being called repeatedly and that the next tool waits until the image description is finished before executing, you can follow the example provided below:

async for event in agent_executor.astream_events(
    {"messages": [HumanMessage(content="describe this image: [image_url]")]}, version="v1"
):
    kind = event["event"]
    if kind == "on_chain_start":
        if event["name"] == "Agent":
            print(f"Starting agent: {event['name']} with input: {event['data'].get('input')}")
    elif kind == "on_chain_end":
        if event["name"] == "Agent":
            print()
            print("--")
            print(f"Done agent: {event['name']} with output: {event['data'].get('output')['output']}")
    if kind == "on_chain_stream":
        content = event["data"]["chunk"].content
        if content:
            print(content, end="|")
    elif kind == "on_tool_start":
        print("--")
        print(f"Starting tool: {event['name']} with inputs: {event['data'].get('input')}")
    elif kind == "on_tool_end":
        print(f"Done tool: {event['name']}")
        print(f"Tool output was: {event['data'].get('output')}")
        print("--")

This code ensures that the describe_image tool outputs in a streaming manner and that the next tool waits until the image description is finished before executing. The AgentExecutor class uses the astream_events method to handle streaming responses, ensuring that the underlying language model is invoked in a streaming fashion, allowing access to individual tokens as they are generated. This approach accumulates the output into a final result, which is then used to determine the next steps in the execution flow.

The astream_events method in the AgentExecutor class is designed to stream all output from a runnable, including intermediate results, which ensures that the next tool in the sequence waits until the current tool has finished its execution. This is achieved by leveraging the astream_log method, which streams output and selected intermediate results from an input, ensuring that the execution flow is properly managed.

For example, in the provided document, the StreamingRunnable class demonstrates how to implement a custom runnable that streams output. The astream method in this class yields elements from an iterable, and the on_chain_start and on_chain_end events are used to manage the execution flow:

class StreamingRunnable(Runnable[Input, Output]):
    """A custom runnable used for testing purposes"""

    iterable: Iterable[Any]

    def __init__(self, iterable: Iterable[Any]) -> None:
        """Initialize the runnable."""
        self.iterable = iterable

    async def astream(
        self,
        input: Input,
        config: Optional[RunnableConfig] = None,
        **kwargs: Optional[Any],
    ) -> AsyncIterator[Output]:
        config = ensure_config(config)
        callback_manager = get_callback_manager_for_config(config)
        run_manager = callback_manager.on_chain_start(
            dumpd(self),
            input,
            name=config.get("run_name", self.get_name()),
            run_id=config.get("run_id"),
        )

        try:
            final_output = None
            for element in self.iterable:
                if isinstance(element, BaseException):
                    raise element
                yield element

                if final_output is None:
                    final_output = element
                else:
                    try:
                        final_output = final_output + element
                    except TypeError:
                        final_output = element

            # set final channel values as run output
            run_manager.on_chain_end(final_output)
        except BaseException as e:
            run_manager.on_chain_error(e)
            raise

This example shows how to handle streaming output and manage the execution flow using events, ensuring that the next tool waits until the current tool has finished its execution ^[1][2][3][4].

To continue talking to Dosu, mention @dosu.

[dosubot[bot]]

To use the astream_events method in the AgentExecutor class to ensure that the describe_image tool outputs in a streaming manner without being called repeatedly and that the next tool waits until the image description is finished before executing, you can follow the example provided below:

async for event in agent_executor.astream_events(
    {"messages": [HumanMessage(content="describe this image: [image_url]")]}, version="v1"
):
    kind = event["event"]
    if kind == "on_chain_start":
        if event["name"] == "Agent":
            print(f"Starting agent: {event['name']} with input: {event['data'].get('input')}")
    elif kind == "on_chain_end":
        if event["name"] == "Agent":
            print()
            print("--")
            print(f"Done agent: {event['name']} with output: {event['data'].get('output')['output']}")
    if kind == "on_chain_stream":
        content = event["data"]["chunk"].content
        if content:
            print(content, end="|")
    elif kind == "on_tool_start":
        print("--")
        print(f"Starting tool: {event['name']} with inputs: {event['data'].get('input')}")
    elif kind == "on_tool_end":
        print(f"Done tool: {event['name']}")
        print(f"Tool output was: {event['data'].get('output')}")
        print("--")

This code ensures that the describe_image tool outputs in a streaming manner and that the next tool waits until the image description is finished before executing. The AgentExecutor class uses the astream_events method to handle streaming responses, ensuring that the underlying language model is invoked in a streaming fashion, allowing access to individual tokens as they are generated. This approach accumulates the output into a final result, which is then used to determine the next steps in the execution flow.

The astream_events method in the AgentExecutor class is designed to stream all output from a runnable, including intermediate results, which ensures that the next tool in the sequence waits until the current tool has finished its execution. This is achieved by leveraging the astream_log method, which streams output and selected intermediate results from an input, ensuring that the execution flow is properly managed.

For example, in the provided document, the StreamingRunnable class demonstrates how to implement a custom runnable that streams output. The astream method in this class yields elements from an iterable, and the on_chain_start and on_chain_end events are used to manage the execution flow:

class StreamingRunnable(Runnable[Input, Output]):
    """A custom runnable used for testing purposes"""

    iterable: Iterable[Any]

    def __init__(self, iterable: Iterable[Any]) -> None:
        """Initialize the runnable."""
        self.iterable = iterable

    async def astream(
        self,
        input: Input,
        config: Optional[RunnableConfig] = None,
        **kwargs: Optional[Any],
    ) -> AsyncIterator[Output]:
        config = ensure_config(config)
        callback_manager = get_callback_manager_for_config(config)
        run_manager = callback_manager.on_chain_start(
            dumpd(self),
            input,
            name=config.get("run_name", self.get_name()),
            run_id=config.get("run_id"),
        )

        try:
            final_output = None
            for element in self.iterable:
                if isinstance(element, BaseException):
                    raise element
                yield element

                if final_output is None:
                    final_output = element
                else:
                    try:
                        final_output = final_output + element
                    except TypeError:
                        final_output = element

            # set final channel values as run output
            run_manager.on_chain_end(final_output)
        except BaseException as e:
            run_manager.on_chain_error(e)
            raise

This example shows how to handle streaming output and manage the execution flow using events, ensuring that the next tool waits until the current tool has finished its execution ^[1][2][3][4].

To continue talking to Dosu, mention @dosu.

[MrCsabaToth]

@dosu why did you comment the same thing twice?