Closed
Description
I'm trying to run a simple agent to interact with some user inputs, and then confirm that the agent's session is keeping the context.
I already tested with adk web that if I give my name, and then ask what is my name, the agent remembers the first message. However, when I run the python file directly, the agent is not able to tell me what the content of the first message was. I appreciate any help you could provide.
This is the code I'm using:
--- Scenario ---
async def main_async():
# --- Constants ---
APP_NAME = "memory_example_app"
USER_ID = "mem_user"
MODEL = "gemini-2.0-flash"
# --- Agent Definitions ---
# Agent 1: Simple agent to capture information
info_capture_agent = LlmAgent(
model=MODEL,
name="InfoCaptureAgent",
instruction="Acknowledge the user's statement.",
# output_key="captured_info" # Could optionally save to state too
)
# Agent 2: Agent that can use memory
memory_recall_agent = LlmAgent(
model=MODEL,
name="MemoryRecallAgent",
instruction="Answer the user's question. Use the 'load_memory' tool "
"if the answer might be in past conversations.",
tools=[load_memory] # Give the agent the tool
)
# --- Services and Runner ---
session_service = InMemorySessionService()
memory_service = InMemoryMemoryService() # Use in-memory for demo
runner = Runner(
# Start with the info capture agent
agent=info_capture_agent,
app_name=APP_NAME,
session_service=session_service,
memory_service=memory_service # Provide the memory service to the Runner
)
# Turn 1: Capture some information in a session
print("--- Turn 1: Capturing Information ---")
message = input() #Hi, I'm Max, and I'm 20 years old.
session1_id = "session_info"
session1 = runner.session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=session1_id)
user_input1 = Content(parts=[Part(text=message)], role="user")
# Run the agent
final_response_text = "(No final response)"
async for event in runner.run_async(user_id=USER_ID, session_id=session1_id, new_message=user_input1):
if event.is_final_response() and event.content and event.content.parts:
final_response_text = event.content.parts[0].text
print(f"Agent 1 Response: {final_response_text}")
# Get the completed session
completed_session1 = runner.session_service.get_session(app_name=APP_NAME, user_id=USER_ID, session_id=session1_id)
# Add this session's content to the Memory Service
print("\n--- Adding Session 1 to Memory ---")
memory_service = memory_service.add_session_to_memory(completed_session1)
print("Session added to memory.")
# Turn 2: In a *new* (or same) session, ask a question requiring memory
print("\n--- Turn 2: Recalling Information ---")
session2_id = "session_recall" # Can be same or different session ID
session2 = runner.session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=session2_id)
# Switch runner to the recall agent
runner.agent = memory_recall_agent
user_input2 = Content(parts=[Part(text="What is my name?")], role="user")
# Run the recall agent
print("Running MemoryRecallAgent...")
final_response_text_2 = "(No final response)"
async for event in runner.run_async(user_id=USER_ID, session_id=session2_id, new_message=user_input2):
print(f" Event: {event.author} - Type: {'Text' if event.content and event.content.parts and event.content.parts[0].text else ''}"
f"{'FuncCall' if event.get_function_calls() else ''}"
f"{'FuncResp' if event.get_function_responses() else ''}")
if event.is_final_response() and event.content and event.content.parts:
final_response_text_2 = event.content.parts[0].text
print(f"Agent 2 Final Response: {final_response_text_2}")
break # Stop after final response
def main():
"""Entry point for the application."""
asyncio.run(main_async())
if name == "main":
main()