How to stream token of agent response in agent supervisor? #137
Comments
|
It should be astream_log() but it is broken in some agents. I have an issue filed for the break. |
|
try this: graph = workflow.compile()
async def main():
async for output in graph.astream_log(
{
"messages": [
HumanMessage(content="Code hello world and print it to the terminal")
]
}, include_types=["llm"]
):
for op in output.ops:
if op["path"] == "/streamed_output/-":
# this is the output from .stream()
...
elif op["path"].startswith("/logs/") and op["path"].endswith(
"/streamed_output/-"
):
# because we chose to only include LLMs, these are LLM tokens
print(op["value"])
if __name__ == "__main__":
import asyncio
asyncio.run(main()) |
|
Thank you for your reply, but I am facing this issue. And also can you please advice how to add converational memory there after we fix streaming issue? |
|
That is the issue I have filed. It is not fixed yet. It appears in many graphs, but not all. |
|
Thank you for your response, have you ever implemented conversational memory in agent supervisor? |
|
I have faced the same issue this week - any luck guys? The streaming doesnt seem to work at all in the langgraph library :( |
|
Not yet. (
… Message ID: ***@***.***>
|
|
It seems I have figured it out how to fix tokens streaming. I am not sure about your code, because it is not async, but I was using this notebook and the graph was not streaming tokens as expected. After debugging internals of langgraph, I have figured out that you need to create extra parameter for your llm calling function and pass it to llm: That way langgraph is able to pass callbacks to the llm to handle its stream and I got my tokens stream. Again, I am not sure about your code, because tutorials suggest to use |
Thanks this worked for me with AzureChatOpenAI |
|
@dmitryrPlanner5D - I spent a few hours trying to solve this and you figured it out for me! Passing the config to the LLM enables chunk streaming inside graphs!!! Wonderful!!! If this were nostr I would zap you! |
|
I have the same problem with this setup # List of members participating in the conversation
members = ["MongoDBAgent"]
# Setup the nodes
supervisor_node = supervisor_factory.setup_node(members)
mongo_node = mongo_factory.setup_node(config)
# Setup the graph and add the nodes
workflow = StateGraph(AgentState)
workflow.add_node("Supervisor", supervisor_node)
workflow.add_node("MongoDBAgent", mongo_node)
# Define the edges
for member in members:
workflow.add_edge(member, "Supervisor")
# The supervisor populates the "next" state, hence routing the conversation
# A conditional map is a dict that maps the output of the supervisor to the next node
# e.g. {'MongoDBAgent': 'MongoDBAgent', 'FINISH': END}
conditional_map = {member: member for member in members}
conditional_map["FINISH"] = END
print(conditional_map)
workflow.add_conditional_edges("Supervisor", lambda state: state["next"], conditional_map)
# Entry point
workflow.set_entry_point("Supervisor")
# Compile
graph = workflow.compile()with the following def create_agent(
llm: ChatOpenAI, tools: Sequence[BaseTool], system_prompt: str
) -> AgentExecutor:
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
system_prompt,
),
MessagesPlaceholder(variable_name="messages"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
]
)
agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools) # type: ignore https://github.com/langchain-ai/langchain/issues/13075
return executor
def setup_agent(config: Config) -> AgentExecutor:
llm = ChatOpenAI(model=config.model, streaming=config.streaming)
return create_agent(llm, _get_tools(config), system_message)
def setup_node(config: Config) -> functools.partial[dict[str, list[HumanMessage]]]:
mongo_agent = setup_agent(config)
mongo_node = functools.partial(agent_node, agent=mongo_agent, name=NODE_NAME)
return mongo_nodeCalling my graph with async for output in graph.astream_log(inputs, include_types=["llm"]):
# astream_log() yields the requested logs (here LLMs) in JSONPatch format
for op in output.ops:
if op["path"] == "/streamed_output/-":
# this is the output from .stream()
...
elif op["path"].startswith("/logs/") and op["path"].endswith(
"/streamed_output/-"
):
# because we chose to only include LLMs, these are LLM tokens
print(op["value"])Gives the following error
I can't see where I can add the config
|
|
Hi, @usersina And regarding config: in my solution, I suggest to add config parameter to function that you pass to Also I see that you already have parameter with name |
|
@dmitryrPlanner5D , thanks for the suggestion! The MongoDB setup is just a simple tool to know the names of the collections in a mongodb database, so nothing crazy. I did however follow what you said and streaming is working now. However, I still see the inputs = {"messages": [HumanMessage(content="How many collections do we have?")]}
async for output in graph.astream_log(inputs, include_types=["llm"]):
for op in output.ops:
print(op)
The reason there is a This is most definitely because my class AgentState(TypedDict):
"""
The agent state is the input to each node in the graph
"""
messages: Annotated[Sequence[BaseMessage], operator.add]
"""
The annotation tells the graph that new messages will always be added
to the current state
"""
next: str
"""
The next node to execute
"""
|
|
I found the issue and the solution. |
Checked other resources
Example Code
import os
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langchain.output_parsers.openai_functions import JsonOutputFunctionsParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
import operator
from typing import Annotated, Any, Dict, List, Optional, Sequence, TypedDict
import functools
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langgraph.graph import StateGraph, END
from langchain_community.utilities import SerpAPIWrapper
from langchain.agents import Tool
from utils import toggle_case,sort_string
os.environ["OPENAI_API_KEY"] = "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"
os.environ["SERPER_API_KEY"] = "c3b73653f4256d5f2b4b5cf4e6fa438d736de7a4717b0fe06d92df0f30fbd3ce"
class AgentSupervisor:
def init(self, llm):
self.llm = llm
agent_supervisor = AgentSupervisor.getAgentSupervisor()
agent_name = ''
for s in agent_supervisor.stream(
{
"messages": [
HumanMessage(content=question)
]
},
{
"recursion_limit": 100
}
):
if "end" not in s:
if 'supervisor' in s:
agent_name = s['supervisor']['next']
if agent_name != "FINISH":
await websocket.send_text(json.dumps({"token":"AgentName:"+agent_name+"\n"}))
print(agent_name)
if agent_name in s:
content = s[agent_name]['messages'][0].content
await websocket.send_text(json.dumps({"token":"Response:"+content+"\n"}))
print(content)
print("----")
Error Message and Stack Trace (if applicable)
No Error, it is outputing properly, but I need a way to stream tokens of agent response, it is outputing full agent response now.
Description
I am trying to stream tokens of agent response in agent super visor.
Right now, it is outputing agent name and full agent response, Here I want to stream tokens of agent response.
System Info
platform: windows
python version: 3.11.2
langchain version: latest version
The text was updated successfully, but these errors were encountered: