Невозможно ответить как поток по токену, когда агент является частью рабочего процесса в Langgraph. ⇐ Python

[Anonymous]

Я разрабатываю сложный рабочий процесс, в котором будет несколько агентов. Некоторые из агентов будут промежуточными, а некоторые будут ориентированы на пользователя (то есть я покажу пользователю выходные данные этих агентов). Поскольку задействовано несколько агентов, мне приходится использовать рабочий процесс.
Но проблема в том, что когда я добавляю любого агента в рабочий процесс (

Код: Выделить всё

StateGraph

), он теряет возможность потоковой передачи по токену. Для простоты я показываю это поведение с одним узлом агента в рабочем процессе:

Код: Выделить всё

from langgraph.graph import MessagesState
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START
from langgraph.checkpoint.memory import MemorySaver
from langchain.agents import create_agent
from dotenv import load_dotenv

load_dotenv()
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.1, max_tokens=1000)

agent = create_agent(
model=model,
tools=[],
)

def agent_node(state: MessagesState):
return agent.invoke({"messages": state["messages"]})

checkpointer = MemorySaver()
workflow = StateGraph(MessagesState)
workflow.add_node("agent", agent_node) # Adding a node containing agent
# workflow.add_node("agent", agent) # Or, we could have also added the agent directly, but it will not help in token-by-token streaming
workflow.add_edge(START, "agent")
compiled_workflow = workflow.compile(checkpointer=checkpointer)

def response(user_input, agent_or_workflow, config):
user_input = {"messages": [("human", user_input)]}
for mode_chunk in agent_or_workflow.stream(user_input, config, stream_mode=["messages"]):
mode, chunk = mode_chunk

if mode == "messages":
token, metadata = chunk
if token.content:
print(token.content, end="\n", flush=True)

if __name__ == "__main__":
config = {"configurable": {"thread_id": "chat-1"}}

# The following will stream token-by-token
response("Hi", agent, config)
print("-" * 30)
# The following will NOT stream token-by-token
response("Hi", compiled_workflow, config)

Здесь вы можете видеть, что когда я вызываю агент как отдельный объект, он может прекрасно передавать поток токен за токеном. Но как только он становится частью рабочего процесса, он больше не может этого делать. Сначала я подумал, что, поскольку я инкапсулирую агент с помощью узла Agent_node, это может сделать этот шаг операцией блокировки, поэтому я попробовал напрямую добавить агента в качестве узла. Но это тоже не помогло (см. строку с комментариями в процессе построения рабочего процесса.
Как я решу эту проблему?
В моем случае этот рабочий процесс будет серверной частью голосового бота. Таким образом, пропускная способность имеет решающее значение. И меня интересует только потоковая передача узлов, которые обращены к пользователю, т. е. выходные данные должны быть отправлены пользователю. Любой промежуточный узел, выходные данные которого не предназначены для просмотра или прослушивания пользователем - Меня не волнует, транслируются они в потоковом режиме или нет, поскольку для них нам в любом случае нужно получить общий результат, чтобы мы могли перейти к следующему шагу рабочего процесса.
Примечание:

• Мне нужна потоковая передача в реальном времени, а не просто циклический цикл по токенам уже полученного полного сообщения.
• Я хочу использовать встроенный в Langgraph метод create_agent и НЕ хочу имитировать агент с помощью llm с помощьюbind_tools. Поскольку я уже реализовал это сам, и для этого подходит потоковая передача, но я считаю, что теряю встроенные функции (например, промежуточное программное обеспечение), которые есть в агентах, созданных с помощью встроенного метода create_agent, мне понадобятся эти дополнительные функции для других функций.

Подробнее здесь: https://stackoverflow.com/questions/798 ... workflow-i