Lesson4 - Persistence and Streaming - Last part gives error "SyntaxError: 'async for' outside async function"

Short Course Q&A AI Agents in LangGraph
1

I am trying to reproduce lesson code on a local machine, ouside Jupyter notebook, and I found that when I try to run the last part of Lesson4 - Persistence and Streaming, the part related with “Steraming tokens” section, gives this error: “SyntaxError: ‘async for’ outside async function”.

If I understand correctly, async must be inside a function, so is it a valid error expected? This only works ok in Jupyter and will fail on common environments?, or maybe is failing because my local environment conditions?

I have created a function with async, imported asyncio and implemented asyncio.run to run that function and works. I just want to let here the comment for if it is useful.

2

I’m following along in Google colab. I’ve got the async loop running by getting a handle on an event loop and by using nesting_asyncio. It runs without error, but the streaming tokens never appear.

import asyncio
import nest_asyncio

async def main():
    messages = [HumanMessage(content="What is the weather in Huntsville, Alabama?")]
    thread = {"configurable": {"thread_id": "4"}}
    async for event in abot.graph.astream_events({"messages": messages}, thread, version="v1"):
        kind = event["event"]
        if kind == "on_chat_model_stream":
            content = event["data"]["chunk"].content
            if content:
                # Empty content in the context of OpenAI means
                # that the model is asking for a tool to be invoked.
                # So we only print non-empty content
                print(content, end="|")

#loop = asyncio.new_event_loop()
nest_asyncio.apply()
# asyncio.set_event_loop(loop)
# await loop.create_task(main())
await main()

Output:

Calling: {'name': 'tavily_search_results_json', 'args': {'query': 'current weather in Huntsville, Alabama'}, 'id': 'call_1Zp7mLJYzbU93YohO1p9qcVq'}
Back to the model!
3

Hi cigotete, I have the error you. Can you please share your solution?

4

Hello oaimnil. Here is the function.

from langgraph.checkpoint.aiosqlite import AsyncSqliteSaver

async def stream_tokens():
    print("="*50)
    print("Streaming tokens")
    print("="*50)

    memory = AsyncSqliteSaver.from_conn_string(":memory:")
    abot = Agent(model, [tool], system=prompt, checkpointer=memory)

    messages = [HumanMessage(content="What is the weather in SF?")]
    thread = {"configurable": {"thread_id": "4"}}
    async for event in abot.graph.astream_events({"messages": messages}, thread, version="v1"):
        kind = event["event"]
        if kind == "on_chat_model_stream":
            content = event["data"]["chunk"].content
            if content:
                print(content, end="|")

import asyncio

asyncio.run(stream_tokens())
5

Thanks cigotete, I change the way of using memory as the context. but still didn’t get the streaming output in terms of token.
I guess it is the because the version of the packages.

6

Hello oaimnil, this is the Langgraph version used for the course on that date: 0.0.69 - 14 jun 2024