Artifact stored as text is not retrieved as text #3157
Description
** Description **
When you store an artifact Part.from_text() with GcsArtifactService, the artifact.text attribute is None after loading the artifact again. The data is however present in artifact.inline_data.
I would have expected the text attribute to contain the text.
To Reproduce
See this adapted weather sample to demonstrate the issue.
import os
import asyncio
from google.adk.agents import Agent
from google.adk.sessions import InMemorySessionService, Session
from google.adk.artifacts import GcsArtifactService
from google.adk.runners import Runner
from google.adk.tools import BaseTool
from google.adk.tools.tool_context import ToolContext
from google.genai import types
import logging
import json
logging.basicConfig(level=logging.ERROR)
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "True"
os.environ["GOOGLE_CLOUD_PROJECT"] = "your-favourite-project"
os.environ["GOOGLE_CLOUD_LOCATION"] = "your-region"
ARTIFACTS_BUCKET = "your-favourite-bucket"
AGENT_MODEL = "gemini-2.5-flash-lite"
APP_NAME = "weather_tutorial_app"
USER_ID = "user_1"
SESSION_ID = "session_001" # Using a fixed ID for simplicity
def get_weather(city: str) -> dict:
"""Retrieves the current weather report for a specified city.
Args:
city (str): The name of the city (e.g., "New York", "London", "Tokyo").
Returns:
dict: A dictionary containing the weather information.
Includes a 'status' key ('success' or 'error').
If 'success', includes a 'report' key with weather details.
If 'error', includes an 'error_message' key.
"""
city_normalized = city.lower().replace(" ", "")
mock_weather_db = {
"newyork": {"status": "success", "report": "The weather in New York is sunny with a temperature of 25°C."},
"london": {"status": "success", "report": "It's cloudy in London with a temperature of 15°C."},
"tokyo": {"status": "success", "report": "Tokyo is experiencing light rain and a temperature of 18°C."},
}
if city_normalized in mock_weather_db:
return mock_weather_db[city_normalized]
else:
return {"status": "error", "error_message": f"Sorry, I don't have weather information for '{city}'."}
async def collect_artifacts(tool:BaseTool, args, tool_context:ToolContext, tool_response):
if tool.name == 'transfer_to_agent':
return
await tool_context.save_artifact('weather', types.Part.from_text(text=json.dumps(tool_response)))
root_agent = Agent(
name="weather_agent",
model=AGENT_MODEL,
description="The weather agent. Answers questions about the weather in certain places.",
instruction="You are the Weather Agent. Your primary responsibility is to provide weather information. "
"Use the 'get_weather' tool for specific weather requests (e.g., 'weather in London'). "
"For anything else, respond appropriately or state you cannot handle it.",
tools=[get_weather],
after_tool_callback=collect_artifacts
)
async def run_agent(runner: Runner, session: Session, content):
print(" ================================= ")
final_response_text = ''
def from_content(content: types.Part):
if not content:
return '<None>'
else:
if content.text:
return content.text.replace('\n', '\\n')
else:
return '<data structure>'
async for event in runner.run_async(user_id=session.user_id, session_id=session.id, new_message=content):
if event and event.content and event.content.parts:
for part in event.content.parts:
if part.thought:
event_type = 'Thought'
elif part.text:
event_type = 'Text'
elif part.function_call:
event_type = 'Function Call'
elif part.function_response:
event_type = 'Function Response'
else:
event_type = 'Other event'
print(f"--> [Event] Author: {event.author}, Type: {event_type}, Final: {event.is_final_response()}, Actions: {event.actions is not None}, Content: {from_content(part)}")
print()
if event_type == 'Text' and event.is_final_response() and part.text:
final_response_text += f'{event.author}: {part.text} '
if event.is_final_response() and event.actions and event.actions.escalate:
final_response_text = f"Agent escalated: {event.error_message or 'No specific message.'}"
print(f"<<< Agent Response: {final_response_text if len(final_response_text) else 'Could not identify final response.'}")
if runner.artifact_service:
if artifact := await runner.artifact_service.load_artifact(app_name=session.app_name, user_id=session.user_id, session_id=session.id, filename='weather'):
print(f'Weather artifact text contains: {artifact.text}')
print(f'inline data: {artifact.inline_data}')
async def main():
session_service = InMemorySessionService()
session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
runner = Runner(agent=root_agent, app_name=session.app_name, session_service=session_service, artifact_service=GcsArtifactService(bucket_name=ARTIFACTS_BUCKET))
content = types.Content(role='user', parts=[types.Part(text='Tell me about the weather in New York')])
await run_agent(runner, session, content)
if __name__ == "__main__":
try:
asyncio.run(main())
except Exception as e:
print(f"An error occurred: {e}")Expected behavior
Weather artifact text should be json string of function calling result instead of None.
Desktop (please complete the following information):
- Mac OS
- Python 3.11.9
- ADK 1.16.0 (also tested in 1.15.1)
Model Information:
- gemini-2.5-flash-lite (tested with different models, not relevant)
- Vertex AI mode