Python使用google.adk实现有状态会话时遇异步相关错误

问题根源 #

• InMemorySessionService的create_session、get_session是异步方法，直接同步调用会返回协程对象而非实际会话实例，触发AttributeError: 'coroutine' object has no attribute 'state'
• Runner.run()是异步生成器，普通for循环无法遍历，未await导致会话未正确创建，触发ValueError: Session not found
• 未构建异步执行上下文，导致所有异步操作未被调度执行，出现RuntimeWarning: coroutine '...' was never awaited

修正后的完整代码 #

basic_stateful_session.py

import uuid
import asyncio
from dotenv import load_dotenv
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from question_answering_agent import question_answering_agent

load_dotenv()


async def main():
    # 初始化会话存储服务
    session_service_stateful = InMemorySessionService()

    initial_state = {
        "user_name": "John Doe",
        "user_preferences": """
            I like to play Pickleball, Disc Golf, and Tennis.
            My favorite food is Mexican.
            My favorite TV show is Game of Thrones.
            Loves it when people like and subscribe to his YouTube channel.
        """,
    }

    # 创建会话 - 必须await异步方法
    APP_NAME = "John Doe Bot"
    USER_ID = "john_doe"
    SESSION_ID = str(uuid.uuid4())
    stateful_session = await session_service_stateful.create_session(
        app_name=APP_NAME,
        user_id=USER_ID,
        session_id=SESSION_ID,
        state=initial_state,
    )
    print("CREATED NEW SESSION:")
    print(f"\tSession ID: {SESSION_ID}")

    runner = Runner(
        agent=question_answering_agent,
        app_name=APP_NAME,
        session_service=session_service_stateful,
    )

    new_message = types.Content(
        role="user", parts=[types.Part(text="What is Johns favorite TV show?")]
    )

    # 用async for遍历异步生成器
    async for event in runner.run(
        user_id=USER_ID,
        session_id=SESSION_ID,
        new_message=new_message,
    ):
        if event.is_final_response():
            if event.content and event.content.parts:
                print(f"Final Response: {event.content.parts[0].text}")

    print("==== Session Event Exploration ====")
    # 获取会话 - 必须await异步方法
    session = await session_service_stateful.get_session(
        app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID
    )

    # 打印最终会话状态
    print("=== Final Session State ===")
    for key, value in session.state.items():
        print(f"{key}: {value}")


if __name__ == "__main__":
    asyncio.run(main())

question_answering_agent.py（无需修改）

from google.adk.agents import Agent

# 创建问答Agent
question_answering_agent = Agent(
    name="question_answering_agent",
    model="gemini-2.0-flash",
    description="Question answering agent",
    instruction="""
    You are a helpful assistant that answers questions about the user's preferences.

    Here is some information about the user:
    Name: 
    {user_name}
    Preferences: 
    {user_preferences}
    """,
)

关键修正点 #

1. 新增asyncio依赖，将核心逻辑封装到async def main()函数中，通过asyncio.run(main())启动异步执行上下文
2. 所有异步方法调用前添加await关键字，包括create_session、get_session
3. 使用async for替代普通for循环，遍历runner.run()返回的异步生成器
4. 确保.env文件中配置了GOOGLE_API_KEY，load_dotenv()会自动加载该密钥供Google AI SDK使用

内容的提问来源于stack exchange，提问作者PinkBanter