Как агент langchain мог шаг за шагом выполнять `astream_event()`? ⇐ Python

[Anonymous]

Пример кода

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

@tool
async def tool1():
"""tool1"""
print('start...')
await asyncio.sleep(3)
print('time....')

@tool
async def tool2():
"""tool2"""
print('start...')
await asyncio.sleep(3)
print('time....')

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)

Результаты

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

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

start...
start...

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

time....
time....

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

`None`

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

`None`

Описание
Если некоторые из моих инструментов имеют зависимости, например: инструменту 3 необходимо дождаться завершения работы инструмента 2, прежде чем его можно будет выполнить. Есть ли способ добиться этого? С помощью astream_event я заметил, что разные инструменты выполняются практически одновременно и асинхронно.
Я также пробовал использовать метод .stream. Хотя этот метод позволяет выполнять инструменты шаг за шагом, процесс задачи не выводится в потоковом виде, а это не то, что мне нужно.

Подробнее здесь: https://stackoverflow.com/questions/787 ... ream-event