Description
I am running the following code in Colab when I get the "Session not found: session_vsearch_1" error.
YOUR_ENGINE_ID = "projects/322843095383/locations/global/collections/default_collection/engines/3gpp_1684856522257"
Constants
APP_NAME_VSEARCH = "vertex_search_app"
USER_ID_VSEARCH = "user_vsearch_1"
SESSION_ID_VSEARCH = "session_vsearch_1"
AGENT_NAME_VSEARCH = "root_agent"
GEMINI_2_FLASH = "gemini-2.0-flash-001"
Tool Instantiation
You MUST provide your datastore ID here.
vertex_search_tool = VertexAiSearchTool(search_engine_id=YOUR_ENGINE_ID)
Agent Definition
root_agent = LlmAgent(
name=AGENT_NAME_VSEARCH,
model=GEMINI_2_FLASH, # Requires Gemini model
tools=[vertex_search_tool],
instruction=f"""You are a helpful assistant that answers questions based on information found through the tool provided to you: {YOUR_ENGINE_ID}.
Use the tool to find relevant information before answering.
If the answer isn't in the response from the tool, say that you couldn't find the information. Do not use any other source to answer questions.
""",
description="Answers questions using a specific Vertex AI Search engine.",
)
Session and Runner Setup
session_service_vsearch = InMemorySessionService()
runner_vsearch = Runner(
agent=root_agent, app_name=APP_NAME_VSEARCH, session_service=session_service_vsearch
)
session_vsearch = session_service_vsearch.create_session(
app_name=APP_NAME_VSEARCH, user_id=USER_ID_VSEARCH, session_id=SESSION_ID_VSEARCH
)
Agent Interaction Function
async def call_vsearch_agent_async(query):
print("\n--- Running Vertex AI Search Agent ---")
print(f"Query: {query}")
if "YOUR_DATASTORE_ID_HERE" in YOUR_ENGINE_ID:
print("Skipping execution: Please replace YOUR_DATASTORE_ID_HERE with your actual datastore ID.")
print("-" * 30)
return
content = types.Content(role='user', parts=[types.Part(text=query)])
final_response_text = "No response received."
try:
async for event in runner_vsearch.run_async(
user_id=USER_ID_VSEARCH, session_id=SESSION_ID_VSEARCH, new_message=content
):
# Like Google Search, results are often embedded in the model's response.
if event.is_final_response() and event.content and event.content.parts:
final_response_text = event.content.parts[0].text.strip()
print(f"Agent Response: {final_response_text}")
# You can inspect event.grounding_metadata for source citations
if event.grounding_metadata:
print(f" (Grounding metadata found with {len(event.grounding_metadata.grounding_attributions)} attributions)")
except Exception as e:
print(f"An error occurred: {e}")
print("Ensure your datastore ID is correct and the service account has permissions.")
print("-" * 30)
--- Run Example ---
async def run_vsearch_example():
# Replace with a question relevant to YOUR datastore content
await call_vsearch_agent_async("Describe ODA")
#await call_vsearch_agent_async("Generate Python Flask Code for ")
Execute the example
await run_vsearch_example()
Running locally due to potential colab asyncio issues with multiple awaits
asyncio.run(run_vsearch_example())
"""
try:
asyncio.run(run_vsearch_example())
except RuntimeError as e:
if "cannot be called from a running event loop" in str(e):
print("Skipping execution in running event loop (like Colab/Jupyter). Run locally.")
else:
raise e
"""