Gemini causes 'Event loop is closed' when running inside an async context #748
Comments
|
@asaf, it looks like the event loop in GeminiAgentModel, which is responsible for managing and running asynchronous tasks, has been shut down, and the program is trying to reuse it. Once closed, the event loop cannot execute any more tasks. Please could you share the full stack trace of the error so that we can narrow down which lines are the culprit of this problem. Thanks in advance! |
|
Unfortunately, I'm not familiar enough with streamlit to be able to reproduce the issue on my own, I tried and just get this: If you can provide a more detailed traceback or a self-contained way to reproduce this (i.e., more information about how to execute the script to cause the problem, any environment settings that need to be set, etc.) then we can continue to look into this more. |
|
Hey guys, |
|
@asaf I just did some research and this is what I found. I hope the explanation makes sense and the suggestions work for you. The RuntimeError: Event loop is closed error in Streamlit when running code in an async context occurs due to event loop conflicts. This happens because Streamlit itself uses asynchronous programming internally, and improper handling of the event loop can lead to the error. This is very similiar to the issues with Jupyter notebooks addressed already in the docs [1]
Possible FixesTry to call your main() function directly # Call main() directly, without the need for wrapping it with asyncio.run()
# Let me know if this works for you
import asyncio
import streamlit as st
from pydantic_ai import Agent
from pydantic_ai.messages import ModelRequest, UserPromptPart
async def main():
st.title("Simple chat")
# Initialize chat history
if "messages" not in st.session_state:
st.session_state.messages = []
# Display chat messages from history on app rerun
for message in st.session_state.messages:
part1 = message.parts[0]
role = ""
if isinstance(part1, UserPromptPart):
role = "user"
with st.chat_message("human" if role == "user" else "ai"):
st.markdown(message.parts[0].content)
# Accept user input
if prompt := st.chat_input("Say something..."):
# Display user message in chat message container
st.chat_message("user").markdown(prompt)
# Add user message to chat history
st.session_state.messages.append(
ModelRequest(parts=[UserPromptPart(content=prompt)])
)
# Display assistant response in chat message container
with st.chat_message("assistant"):
response_container = st.empty() # Placeholder for streaming response
agent = Agent(
"google-gla:gemini-2.0-flash-exp",
# "openai:gpt-4o", # works!
# https://ai.pydantic.dev/results/#structured-result-validation
result_type=str, # type: ignore
system_prompt="you are a nice bot",
)
result = await agent.run(prompt)
response_container.markdown(result.data)
if __name__ == "__main__":
main()Alternatively, call nest_asyncio.apply() at the beginning of your code right after the imports as shown in [1] import asyncio
import nest_asyncio
import streamlit as st
from pydantic_ai import Agent
from pydantic_ai.messages import ModelRequest, UserPromptPart
nest_asyncio.apply() # Allows nesting of event loops
async def main():
st.title("Simple chat")
# Initialize chat history
if "messages" not in st.session_state:
st.session_state.messages = []
# Display chat messages from history on app rerun
for message in st.session_state.messages:
part1 = message.parts[0]
role = ""
if isinstance(part1, UserPromptPart):
role = "user"
with st.chat_message("human" if role == "user" else "ai"):
st.markdown(message.parts[0].content)
# Accept user input
if prompt := st.chat_input("Say something..."):
# Display user message in chat message container
st.chat_message("user").markdown(prompt)
# Add user message to chat history
st.session_state.messages.append(
ModelRequest(parts=[UserPromptPart(content=prompt)])
)
# Display assistant response in chat message container
with st.chat_message("assistant"):
response_container = st.empty() # Placeholder for streaming response
agent = Agent(
"google-gla:gemini-2.0-flash-exp",
# "openai:gpt-4o", # works!
# https://ai.pydantic.dev/results/#structured-result-validation
result_type=str, # type: ignore
system_prompt="you are a nice bot",
)
result = await agent.run(prompt)
response_container.markdown(result.data)
if __name__ == "__main__":
asyncio.run(main())References[1] https://ai.pydantic.dev/troubleshooting/#runtimeerror-this-event-loop-is-already-running |
|
Hey @izzyacademy thanks for answering,
Best, |
|
This issue is stale, and will be closed in 3 days if no reply is received. |
|
Closing this issue as it has been inactive for 10 days. |
|
Just wanted to confirm, nest_asyncio.apply() does not work with Gemini models such as 1.5-flash and 2.0-flash, but even without nest_asyncio.apply() it works with other model providers like groq. Here is a different script to see the problem as well. Even though this was closed, it seems like a bug since it only seems to effect Gemini models. The error for the script below ends with: Here is the script: import streamlit as st
import asyncio
import nest_asyncio
from pydantic_ai import Agent
from pydantic_ai.messages import ModelResponse, TextPart, ModelRequest, UserPromptPart
# nest_asyncio.apply()
# Does not work with or without the above uncommented
#agent = Agent("google-gla:gemini-1.5-flash", system_prompt="Be a helpful assistant who responds like a pirate.")
# Works!
agent = Agent('groq:llama-3.3-70b-versatile', system_prompt="Be a helpful assistant who responds like a pirate.")
def display_message_part(part):
"""
Display a single part of a message in the Streamlit UI.
Customize how you display system prompts, user prompts,
tool calls, tool returns, etc.
"""
# system-prompt
if part.part_kind == 'system-prompt':
with st.chat_message("system"):
st.markdown(f"**System**: {part.content}")
# user-prompt
elif part.part_kind == 'user-prompt':
with st.chat_message("user"):
st.markdown(part.content)
# text
elif part.part_kind == 'text':
with st.chat_message("assistant"):
st.markdown(part.content)
# Streamlit UI
async def run_agent_with_streaming(user_input: str):
"""
Run the agent with streaming text for the user_input prompt,
while maintaining the entire conversation in `st.session_state.messages`.
"""
# Run the agent in a stream
async with agent.run_stream(
user_input,
message_history= st.session_state.messages[:-1], # pass entire conversation so far
) as result:
# We'll gather partial text to show incrementally
partial_text = ""
message_placeholder = st.empty()
# Render partial text as it arrives
async for chunk in result.stream_text(delta=True):
partial_text += chunk
message_placeholder.markdown(partial_text)
# Now that the stream is finished, we have a final result.
# Add new messages from this run, excluding user-prompt messages
filtered_messages = [msg for msg in result.new_messages()
if not (hasattr(msg, 'parts') and
any(part.part_kind == 'user-prompt' for part in msg.parts))]
st.session_state.messages.extend(filtered_messages)
# Add the final response to the messages
st.session_state.messages.append(
ModelResponse(parts=[TextPart(content=partial_text)])
)
async def main():
st.title("Archon - Agent Builder")
st.write("Describe to me an AI agent you want to build and I'll code it for you with Pydantic AI.")
# Initialize chat history in session state if not present
if "messages" not in st.session_state:
st.session_state.messages = []
# Display all messages from the conversation so far
# Each message is either a ModelRequest or ModelResponse.
# We iterate over their parts to decide how to display them.
for msg in st.session_state.messages:
if isinstance(msg, ModelRequest) or isinstance(msg, ModelResponse):
for part in msg.parts:
display_message_part(part)
# Chat input for the user
user_input = st.chat_input("What do you want to build today?")
if user_input:
# We append a new request to the conversation explicitly
st.session_state.messages.append(
ModelRequest(parts=[UserPromptPart(content=user_input)])
)
# Display user prompt in the UI
with st.chat_message("user"):
st.markdown(user_input)
# Display the assistant's partial response while streaming
with st.chat_message("assistant"):
# Actually run the agent now, streaming the text
await run_agent_with_streaming(user_input)
if __name__ == "__main__":
asyncio.run(main()) |
|
Also having a hard time mixing streamlit with pydantic-ai + Gemini. |
|
Did you try the last release? @rubentorresbonet |
|
I am on "0.0.30"; I think that is pretty much the latest one? Thanks |
|
Do you have an MRE I can reproduce the issue? |
|
Hi, I have the same issue. MRE without streamlit here: https://github.com/oscar-broman/pydantic-ai-gemini-issue-mre |
|
This issue is stale, and will be closed in 3 days if no reply is received. |
|
Got same issue when using pytest to test. Add the following can work sometimes but can't guarantee.
|
|
Got the same issue. |
|
This issue is stale, and will be closed in 3 days if no reply is received. |
|
Got the same issue when using gemini 2.0-flash inside a flask[async] endpoint |
|
Hello, getting the same issue even when using Reproducible example: Gives me the following error (traceback here): Anyone else experiencing issues with Gemini synchronous agent runs? Thanks in advance! |
|
I get it with an even simpler reproduction from pydantic_ai.models.gemini import GeminiModel
from pydantic_ai import Agent
GEMINI_PROVIDER = "google-vertex"
model_name: str = "gemini-2.0-flash-001"
model = GeminiModel(model_name, provider=GEMINI_PROVIDER)
llm = Agent(model)
print(llm.run_sync("What is the capital of pennsylvania?"))
print(llm.run_sync("What is the capital of pennsylvania?")) |
|
It seems the problem is related to the fact that we cache the |
|
I was able to reproduce a similar error with this code |
|
I was able to fix this by creating a global loop and using it for all streams. Add this globally: import threading
import asyncio
global_loop = asyncio.new_event_loop()
def start_global_loop(loop):
asyncio.set_event_loop(loop)
loop.run_forever()
# Start the global event loop in a background thread.
threading.Thread(target=start_global_loop, args=(global_loop,), daemon=True).start()And use it like this wherever you want to do the async generate: async def async_generate():
async for chunk in agent.astream(chat_input):
yield json.dumps(chunk)
def generate():
agen = async_generate()
try:
while True:
future = asyncio.run_coroutine_threadsafe(agen.__anext__(), global_loop)
chunk = future.result()
yield chunk
except StopAsyncIteration:
pass |
|
@teshnizi I tried your approach it fixed the event loop closed issue but I got an error related here is a snippet code to reporduce: import threading
import asyncio
from pydantic_ai.models.gemini import GeminiModel
from pydantic_ai.providers.google_vertex import GoogleVertexProvider
from pydantic_ai import Agent
global_loop = asyncio.new_event_loop()
def start_global_loop(loop):
asyncio.set_event_loop(loop)
loop.run_forever()
# Start the global event loop in a background thread.
threading.Thread(target=start_global_loop, args=(global_loop,), daemon=True).start()
agent = Agent(model=GeminiModel("gemini-2.0-flash", provider=GoogleVertexProvider()))
async def async_generate(chat_input):
async with agent.run_stream(chat_input) as result:
async for chunk in result.stream_text():
yield chunk
async def generate(chat_input):
agen = async_generate(chat_input)
try:
while True:
future = asyncio.run_coroutine_threadsafe(agen.__anext__(), global_loop)
chunk = future.result()
yield chunk
except StopAsyncIteration:
pass
async def main():
print("Response for 'Testing':")
async for chunk in generate("Testing"):
print(chunk)
#Consume the second generator
print("Response for 'Testing 2':")
async for chunk in generate("Testing 2"):
print(chunk)
# Run everything in the same event loop
if __name__ == "__main__":
asyncio.run(main())when you run the script you generate the response then you get the error: |
|
i am getting the same issue using Gemini models in async contexts even without using pydantic-ai. Are there any fixes planned for this? |
What do you mean without even pydantic-ai? |
I did mention this, and what is causing it about 3 weeks ago 😂 |
Unfortunately not sure. |
I mean I am not using the PydanticAI framework. I had the same issue using async calls with langchain. |
|
This issue is stale, and will be closed in 3 days if no reply is received. |
|
Closing this issue as it has been inactive for 10 days. |
|
This issue is stale, and will be closed in 3 days if no reply is received. |
|
Same issue with "Event loop is closed" when using Streamlit, Pydantic AI and Gemini for testing. |
|
Working on a tentative PR to fix that, but as far as I can tell, the issue is in To be precise, if I've made a first call to Gemini in a given even loop, then: GeminiModel(model_name="gemini-2.0-flash-lite", provider="google-vertex").client._transport._pool._connections[0]._connection._network_stream._stream.transport_stream._transport._loopwill be a pointer to that event loop even if it's closed, and if we try to use the GeminiModel class again (even with a new instance), it will try to reuse that loop. I think if we make the client/transport cache smarter (e.g. loop dependent), the problem will disappear. |
ewjoachim
added a commit
to ewjoachim/pydantic-ai
that referenced
this issue
May 12, 2025
…t loops # Conflicts: # pydantic_ai_slim/pydantic_ai/models/__init__.py
|
Just for completeness sake, this issue in httpcore is likely around the same issue: encode/httpcore#659 (comment) And so is agronholm/anyio#743
If anyone in this list didn't do that, the problem wouldn't not exist :D |
|
In both issues, anyio and httpx maintainers advised towards improving connection (& connection pool) management, strongly hinting towards using async context manager. That said, I don't know how that might work with PydanticAI's API. But it would make sense to consider that defining things like the HTTP connection object & transport shouldn't happen before we're in an event loop. |
|
This issue is stale, and will be closed in 3 days if no reply is received. |
|
(I think this issue is still relevant. Especially since there's a PR) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Hi,
This is an attempt to use
google-gla:gemini-2.0-flash-expvia streamit in async,Code below causes
RuntimeError: Event loop is closed, same code works just fine foropenai:gpt-4oThe text was updated successfully, but these errors were encountered: