Streaming Agent Not Really Work.

[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

from langchain import hub
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.tools import tool
from langchain_core.callbacks import Callbacks
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

model = ChatOpenAI(temperature=0, streaming=True)

import random


@tool
async def where_cat_is_hiding() -> str:
    """Where is the cat hiding right now?"""
    return random.choice(["under the bed", "on the shelf"])


@tool
async def get_items(place: str) -> str:
    """Use this tool to look up which items are in the given place."""
    if "bed" in place:  # For under the bed
        return "socks, shoes and dust bunnies"
    if "shelf" in place:  # For 'shelf'
        return "books, penciles and pictures"
    else:  # if the agent decides to ask about a different place
        return "cat snacks"

# Get the prompt to use - you can modify this!
prompt = hub.pull("hwchase17/openai-tools-agent")
# print(prompt.messages) -- to see the prompt
tools = [get_items, where_cat_is_hiding]
agent = create_openai_tools_agent(
    model.with_config({"tags": ["agent_llm"]}), tools, prompt
)
agent_executor = AgentExecutor(agent=agent, tools=tools).with_config(
    {"run_name": "Agent"}
)

async def test(query):
    async for chunk in agent_executor.astream(
        {
            "input": query,
            "tools": tools,
            "tool_names": tool_names,
            "agent_scratchpad": lambda x: format_to_openai_tool_messages(
                x["intermediate_steps"]
            ),
        }
    ): 
        if "actions" in chunk:
            for action in chunk["actions"]:
                print(f"Calling Tool: `{action.tool}` with input `{action.tool_input}`")
        elif "steps" in chunk:
            for step in chunk["steps"]:
                print(f"Tool Result: `{step.observation}`")
        elif "output" in chunk:
            print(f'Final Output: {chunk["output"]}')
        else:
            raise ValueError()
        print("---")

asyncio.run(test("what's items are located where the cat is hiding?"))

Description

I just followed the document.

The entire process will output in a streaming manner, but if you focus on a single step, such as "Output," you will notice that the Output is not streamed.

For example:

asyncio.run(test("what's ai?"))

Get result without streaming:

Ai is...

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 address the issue of the output not being streamed as expected, ensure that all components in your chain support streaming. Here are a few steps and considerations to help you troubleshoot and resolve the issue:

1. Ensure All Components Support Streaming:
   Make sure that all components in your chain, including tools and the language model, support streaming. Components that do not support streaming can interrupt the streaming process.

2. Check the format_to_openai_tool_messages Function:
   The format_to_openai_tool_messages function converts intermediate steps into messages for the language model. If this function introduces delays or errors, it could affect the streaming output. Ensure that this function is efficient and correctly implemented.

3. Use the Correct Streaming Method:
   LangChain provides several methods for streaming, including stream, astream, astream_log, and astream_events. Make sure you are using the appropriate method for your use case. For example, astream is the async method for streaming back chunks of the response.

Here is an example of how to set up a chain that supports streaming:

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template("tell me a joke about {topic}")
parser = StrOutputParser()
chain = prompt | model | parser

async for chunk in chain.astream({"topic": "parrot"}):
    print(chunk, end="|", flush=True)

4. Example with AgentExecutor:
   Ensure that the AgentExecutor and tools are correctly configured for streaming. Here is an example of how to use AgentExecutor with streaming:

from langchain import hub
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.tools import tool
from langchain_core.callbacks import Callbacks
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import asyncio
import random

model = ChatOpenAI(temperature=0, streaming=True)

@tool
async def where_cat_is_hiding() -> str:
    return random.choice(["under the bed", "on the shelf"])

@tool
async def get_items(place: str) -> str:
    if "bed" in place:
        return "socks, shoes and dust bunnies"
    if "shelf" in place:
        return "books, pencils and pictures"
    else:
        return "cat snacks"

prompt = hub.pull("hwchase17/openai-tools-agent")
tools = [get_items, where_cat_is_hiding]
agent = create_openai_tools_agent(model.with_config({"tags": ["agent_llm"]}), tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools).with_config({"run_name": "Agent"})

async def test(query):
    async for chunk in agent_executor.astream({"input": query}):
        if "actions" in chunk:
            for action in chunk["actions"]:
                print(f"Calling Tool: `{action.tool}` with input `{action.tool_input}`")
        elif "steps" in chunk:
            for step in chunk["steps"]:
                print(f"Tool Result: `{step.observation}`")
        elif "output" in chunk:
            print(f'Final Output: {chunk["output"]}')
        else:
            raise ValueError()
        print("---")

asyncio.run(test("what's items are located where the cat is hiding?"))

Ensure that all components in your chain are compatible with streaming to achieve the desired behavior. If the issue persists, you may need to review the implementation of each component to ensure they support streaming correctly ^[1][2].

To continue talking to Dosu, mention @dosu.