diff --git a/src/google/adk/flows/llm_flows/contents.py b/src/google/adk/flows/llm_flows/contents.py index 425ed9b3..2ce917ba 100644 --- a/src/google/adk/flows/llm_flows/contents.py +++ b/src/google/adk/flows/llm_flows/contents.py @@ -41,6 +41,10 @@ class _ContentLlmRequestProcessor(BaseLlmRequestProcessor): agent = invocation_context.agent + # Preserve all contents that were added by instruction processor + # (since llm_request.contents will be completely reassigned below) + instruction_related_contents = llm_request.contents + if agent.include_contents == 'default': # Include full conversation history llm_request.contents = _get_contents( @@ -56,9 +60,9 @@ class _ContentLlmRequestProcessor(BaseLlmRequestProcessor): agent.name, ) - # Add dynamic instructions to the last user content if static instructions exist - await _add_dynamic_instructions_to_user_content( - invocation_context, llm_request + # Add instruction-related contents to proper position in conversation + await _add_instructions_to_user_content( + invocation_context, llm_request, instruction_related_contents ) # Maintain async generator behavior @@ -562,47 +566,41 @@ def _is_live_model_audio_event(event: Event) -> bool: return False -async def _add_dynamic_instructions_to_user_content( - invocation_context: InvocationContext, llm_request: LlmRequest +async def _add_instructions_to_user_content( + invocation_context: InvocationContext, + llm_request: LlmRequest, + instruction_contents: list, ) -> None: - """Add dynamic instructions to the last user content when static instructions exist.""" - from ...agents.readonly_context import ReadonlyContext - from ...utils import instructions_utils + """Insert instruction-related contents at proper position in conversation. - agent = invocation_context.agent + This function inserts instruction-related contents (passed as parameter) at the + proper position in the conversation flow, specifically before the last continuous + batch of user content to maintain conversation context. - dynamic_instructions = [] - - # Handle agent dynamic instructions if static instruction exists - if agent.static_instruction and agent.instruction: - # Static instruction exists, so add dynamic instruction to content - raw_si, bypass_state_injection = await agent.canonical_instruction( - ReadonlyContext(invocation_context) - ) - si = raw_si - if not bypass_state_injection: - si = await instructions_utils.inject_session_state( - raw_si, ReadonlyContext(invocation_context) - ) - if si: # Only add if not empty - dynamic_instructions.append(si) - - if not dynamic_instructions: + Args: + invocation_context: The invocation context + llm_request: The LLM request to modify + instruction_contents: List of instruction-related contents to insert + """ + if not instruction_contents: return - # Find the start of the last continuous batch of user content - # Walk backwards to find the first non-user content, then insert before next user content + # Find the insertion point: before the last continuous batch of user content + # Walk backwards to find the first non-user content, then insert after it insert_index = len(llm_request.contents) - for i in range(len(llm_request.contents) - 1, -1, -1): - if llm_request.contents[i].role != 'user': - insert_index = i + 1 - break - elif i == 0: - # All content from start is user content - insert_index = 0 - break - # Create new user content with dynamic instructions - instruction_parts = [types.Part(text=instr) for instr in dynamic_instructions] - new_content = types.Content(role='user', parts=instruction_parts) - llm_request.contents.insert(insert_index, new_content) + if llm_request.contents: + for i in range(len(llm_request.contents) - 1, -1, -1): + if llm_request.contents[i].role != 'user': + insert_index = i + 1 + break + elif i == 0: + # All content from start is user content + insert_index = 0 + break + else: + # No contents remaining, just append at the end + insert_index = 0 + + # Insert all instruction contents at the proper position using efficient slicing + llm_request.contents[insert_index:insert_index] = instruction_contents diff --git a/src/google/adk/flows/llm_flows/instructions.py b/src/google/adk/flows/llm_flows/instructions.py index 70c307d6..bcdc6f56 100644 --- a/src/google/adk/flows/llm_flows/instructions.py +++ b/src/google/adk/flows/llm_flows/instructions.py @@ -34,6 +34,28 @@ if TYPE_CHECKING: class _InstructionsLlmRequestProcessor(BaseLlmRequestProcessor): """Handles instructions and global instructions for LLM flow.""" + async def _process_agent_instruction( + self, agent, invocation_context: InvocationContext + ) -> str: + """Process agent instruction with state injection. + + Args: + agent: The agent with instruction to process + invocation_context: The invocation context + + Returns: + The processed instruction text + """ + raw_si, bypass_state_injection = await agent.canonical_instruction( + ReadonlyContext(invocation_context) + ) + si = raw_si + if not bypass_state_injection: + si = await instructions_utils.inject_session_state( + raw_si, ReadonlyContext(invocation_context) + ) + return si + @override async def run_async( self, invocation_context: InvocationContext, llm_request: LlmRequest @@ -66,16 +88,16 @@ class _InstructionsLlmRequestProcessor(BaseLlmRequestProcessor): # Handle instruction based on whether static_instruction exists if agent.instruction and not agent.static_instruction: # Only add to system instructions if no static instruction exists - # If static instruction exists, content processor will handle it - raw_si, bypass_state_injection = await agent.canonical_instruction( - ReadonlyContext(invocation_context) - ) - si = raw_si - if not bypass_state_injection: - si = await instructions_utils.inject_session_state( - raw_si, ReadonlyContext(invocation_context) - ) + si = await self._process_agent_instruction(agent, invocation_context) llm_request.append_instructions([si]) + elif agent.instruction and agent.static_instruction: + # Static instruction exists, so add dynamic instruction to content + from google.genai import types + + si = await self._process_agent_instruction(agent, invocation_context) + # Create user content for dynamic instruction + dynamic_content = types.Content(role='user', parts=[types.Part(text=si)]) + llm_request.contents.append(dynamic_content) # Maintain async generator behavior return diff --git a/src/google/adk/models/llm_request.py b/src/google/adk/models/llm_request.py index aa898ef7..814e0dbe 100644 --- a/src/google/adk/models/llm_request.py +++ b/src/google/adk/models/llm_request.py @@ -90,7 +90,7 @@ class LlmRequest(BaseModel): def append_instructions( self, instructions: Union[list[str], types.Content] - ) -> None: + ) -> list[types.Content]: """Appends instructions to the system instruction. Args: @@ -98,46 +98,120 @@ class LlmRequest(BaseModel): - list[str]: Strings to append/concatenate to system instruction - types.Content: Content object to append to system instruction - Note: Only text content is supported. Model API requires system_instruction - to be a string. Non-text parts in Content will be handled differently. + Returns: + List of user contents from non-text parts (when instructions is types.Content + with non-text parts). Empty list otherwise. + + Note: Model API requires system_instruction to be a string. Non-text parts + in Content are processed with references in system_instruction and returned + as user contents. Behavior: - list[str]: concatenates with existing system_instruction using \\n\\n - - types.Content: extracts text from parts and concatenates + - types.Content: extracts text parts with references to non-text parts, + returns non-text parts as user contents """ - # Handle Content object - extract only text parts + # Handle Content object if isinstance(instructions, types.Content): - # TODO: Handle non-text contents in instruction by putting non-text parts - # into llm_request.contents and adding a reference in the system instruction - # that references the contents. + text_parts = [] + user_contents = [] - # Extract text from all text parts - text_parts = [part.text for part in instructions.parts if part.text] + # Process all parts, creating references for non-text parts + non_text_count = 0 + for part in instructions.parts: + if part.text: + # Text part - add to system instruction + text_parts.append(part.text) + elif part.inline_data: + # Inline data part - create reference and user content + reference_id = f"inline_data_{non_text_count}" + non_text_count += 1 - if not text_parts: - return # No text content to append + # Create descriptive reference based on mime_type and display_name + display_info = [] + if part.inline_data.display_name: + display_info.append(f"'{part.inline_data.display_name}'") + if part.inline_data.mime_type: + display_info.append(f"type: {part.inline_data.mime_type}") - new_text = "\n\n".join(text_parts) - if not self.config.system_instruction: - self.config.system_instruction = new_text - elif isinstance(self.config.system_instruction, str): - self.config.system_instruction += "\n\n" + new_text - else: - # Log warning for unsupported system_instruction types - logging.warning( - "Cannot append to system_instruction of unsupported type: %s. " - "Only string system_instruction is supported.", - type(self.config.system_instruction), - ) - return + display_text = f" ({', '.join(display_info)})" if display_info else "" + reference_text = ( + f"[Reference to inline binary data: {reference_id}{display_text}]" + ) + text_parts.append(reference_text) + + # Create user content with reference and data + user_content = types.Content( + role="user", + parts=[ + types.Part.from_text( + text=f"Referenced inline data: {reference_id}" + ), + types.Part(inline_data=part.inline_data), + ], + ) + user_contents.append(user_content) + + elif part.file_data: + # File data part - create reference and user content + reference_id = f"file_data_{non_text_count}" + non_text_count += 1 + + # Create descriptive reference based on file_uri and display_name + display_info = [] + if part.file_data.display_name: + display_info.append(f"'{part.file_data.display_name}'") + if part.file_data.file_uri: + display_info.append(f"URI: {part.file_data.file_uri}") + if part.file_data.mime_type: + display_info.append(f"type: {part.file_data.mime_type}") + + display_text = f" ({', '.join(display_info)})" if display_info else "" + reference_text = ( + f"[Reference to file data: {reference_id}{display_text}]" + ) + text_parts.append(reference_text) + + # Create user content with reference and file data + user_content = types.Content( + role="user", + parts=[ + types.Part.from_text( + text=f"Referenced file data: {reference_id}" + ), + types.Part(file_data=part.file_data), + ], + ) + user_contents.append(user_content) + + # Handle text parts for system instruction + if text_parts: + new_text = "\n\n".join(text_parts) + if not self.config.system_instruction: + self.config.system_instruction = new_text + elif isinstance(self.config.system_instruction, str): + self.config.system_instruction += "\n\n" + new_text + else: + # Log warning for unsupported system_instruction types + logging.warning( + "Cannot append to system_instruction of unsupported type: %s. " + "Only string system_instruction is supported.", + type(self.config.system_instruction), + ) + + # Add user contents directly to llm_request.contents + if user_contents: + self.contents.extend(user_contents) + + return user_contents # Handle list of strings if isinstance(instructions, list) and all( isinstance(inst, str) for inst in instructions ): if not instructions: # Handle empty list - return + return [] new_text = "\n\n".join(instructions) if not self.config.system_instruction: @@ -151,7 +225,7 @@ class LlmRequest(BaseModel): "Only string system_instruction is supported.", type(self.config.system_instruction), ) - return + return [] # Invalid input raise TypeError("instructions must be list[str] or types.Content") diff --git a/tests/unittests/agents/test_static_instructions.py b/tests/unittests/agents/test_static_instructions.py index 912a062b..3d20e5cf 100644 --- a/tests/unittests/agents/test_static_instructions.py +++ b/tests/unittests/agents/test_static_instructions.py @@ -17,7 +17,8 @@ from google.adk.agents.invocation_context import InvocationContext from google.adk.agents.llm_agent import LlmAgent from google.adk.agents.run_config import RunConfig -from google.adk.flows.llm_flows.contents import _add_dynamic_instructions_to_user_content +from google.adk.flows.llm_flows.contents import _add_instructions_to_user_content +from google.adk.flows.llm_flows.contents import request_processor as contents_processor from google.adk.flows.llm_flows.instructions import request_processor from google.adk.models.llm_request import LlmRequest from google.adk.sessions.in_memory_session_service import InMemorySessionService @@ -148,9 +149,17 @@ class TestStaticInstructions: pass # Static instruction should be in system instructions - assert len(llm_request.contents) == 0 + # Dynamic instruction should be added as user content by instruction processor + assert len(llm_request.contents) == 1 assert llm_request.config.system_instruction == 'Static instruction content' + # Check that dynamic instruction was added as user content + assert llm_request.contents[0].role == 'user' + assert len(llm_request.contents[0].parts) == 1 + assert ( + llm_request.contents[0].parts[0].text == 'Dynamic instruction content' + ) + @pytest.mark.asyncio async def test_dynamic_instructions_added_to_user_content(self, llm_backend): """Test that dynamic instructions are added to user content when static exists.""" @@ -166,16 +175,22 @@ class TestStaticInstructions: invocation_context = await _create_invocation_context(agent) llm_request = LlmRequest() - # Add some existing user content - llm_request.contents = [ - types.Content(role='user', parts=[types.Part(text='Hello world')]) - ] - # Run the content processor function - await _add_dynamic_instructions_to_user_content( - invocation_context, llm_request + # Run the instruction processor to add dynamic instruction + async for _ in request_processor.run_async(invocation_context, llm_request): + pass + + # Add some existing user content to simulate conversation history + llm_request.contents.append( + types.Content(role='user', parts=[types.Part(text='Hello world')]) ) + # Run the content processor to move instructions to proper position + async for _ in contents_processor.run_async( + invocation_context, llm_request + ): + pass + # Dynamic instruction should be inserted before the last continuous batch of user content assert len(llm_request.contents) == 2 assert llm_request.contents[0].role == 'user' @@ -204,10 +219,15 @@ class TestStaticInstructions: llm_request = LlmRequest() # No existing content - # Run the content processor function - await _add_dynamic_instructions_to_user_content( + # Run the instruction processor to add dynamic instruction + async for _ in request_processor.run_async(invocation_context, llm_request): + pass + + # Run the content processor to handle any positioning (no change expected for single content) + async for _ in contents_processor.run_async( invocation_context, llm_request - ) + ): + pass # Dynamic instruction should create new user content assert len(llm_request.contents) == 1 @@ -230,9 +250,7 @@ class TestStaticInstructions: llm_request.contents = [original_content] # Run the content processor function - await _add_dynamic_instructions_to_user_content( - invocation_context, llm_request - ) + await _add_instructions_to_user_content(invocation_context, llm_request, []) # Content should remain unchanged assert len(llm_request.contents) == 1 @@ -264,9 +282,216 @@ class TestStaticInstructions: async for _ in request_processor.run_async(invocation_context, llm_request): pass - # Static instruction should extract only text parts and concatenate them - assert len(llm_request.contents) == 0 + # Static instruction should contain text parts with references to non-text parts + assert len(llm_request.contents) == 1 assert ( llm_request.config.system_instruction - == 'Analyze this image:\n\nFocus on the key elements.' + == 'Analyze this image:\n\n[Reference to inline binary data:' + ' inline_data_0 (type: image/png)]\n\nFocus on the key elements.' ) + + # The non-text part should be in user content + assert llm_request.contents[0].role == 'user' + assert len(llm_request.contents[0].parts) == 2 + assert ( + llm_request.contents[0].parts[0].text + == 'Referenced inline data: inline_data_0' + ) + assert llm_request.contents[0].parts[1].inline_data + assert ( + llm_request.contents[0].parts[1].inline_data.data == b'fake_image_data' + ) + + @pytest.mark.asyncio + async def test_static_instruction_non_text_parts_moved_to_user_content( + self, llm_backend + ): + """Test that non-text parts from static instruction are moved to user content.""" + static_content = types.Content( + role='user', + parts=[ + types.Part(text='Analyze this image:'), + types.Part( + inline_data=types.Blob( + data=b'fake_image_data', + mime_type='image/png', + display_name='test_image.png', + ) + ), + types.Part( + file_data=types.FileData( + file_uri='files/test123', + mime_type='text/plain', + display_name='test_file.txt', + ) + ), + types.Part(text='Focus on the key elements.'), + ], + ) + agent = LlmAgent(name='test_agent', static_instruction=static_content) + + invocation_context = await _create_invocation_context(agent) + llm_request = LlmRequest() + + # Run the instruction processor + async for _ in request_processor.run_async(invocation_context, llm_request): + pass + + # Run the contents processor to move non-text parts + async for _ in contents_processor.run_async( + invocation_context, llm_request + ): + pass + + # System instruction should contain text with references + expected_system = ( + 'Analyze this image:\n\n[Reference to inline binary data: inline_data_0' + " ('test_image.png', type: image/png)]\n\n[Reference to file data:" + " file_data_1 ('test_file.txt', URI: files/test123, type:" + ' text/plain)]\n\nFocus on the key elements.' + ) + assert llm_request.config.system_instruction == expected_system + + # Non-text parts should be moved to user content + assert len(llm_request.contents) == 2 + + # Check first content object (inline_data) + inline_content = llm_request.contents[0] + assert inline_content.role == 'user' + assert len(inline_content.parts) == 2 + assert ( + inline_content.parts[0].text == 'Referenced inline data: inline_data_0' + ) + assert inline_content.parts[1].inline_data + assert inline_content.parts[1].inline_data.data == b'fake_image_data' + assert inline_content.parts[1].inline_data.mime_type == 'image/png' + assert inline_content.parts[1].inline_data.display_name == 'test_image.png' + + # Check second content object (file_data) + file_content = llm_request.contents[1] + assert file_content.role == 'user' + assert len(file_content.parts) == 2 + assert file_content.parts[0].text == 'Referenced file data: file_data_1' + assert file_content.parts[1].file_data + assert file_content.parts[1].file_data.file_uri == 'files/test123' + assert file_content.parts[1].file_data.mime_type == 'text/plain' + assert file_content.parts[1].file_data.display_name == 'test_file.txt' + + @pytest.mark.asyncio + async def test_static_instruction_reference_id_generation(self, llm_backend): + """Test that reference IDs are generated correctly for non-text parts.""" + static_content = types.Content( + role='user', + parts=[ + types.Part(text='Multiple files:'), + types.Part( + inline_data=types.Blob(data=b'data1', mime_type='image/png') + ), + types.Part( + file_data=types.FileData( + file_uri='files/test1', mime_type='text/plain' + ) + ), + types.Part( + inline_data=types.Blob(data=b'data2', mime_type='image/jpeg') + ), + ], + ) + agent = LlmAgent(name='test_agent', static_instruction=static_content) + + invocation_context = await _create_invocation_context(agent) + llm_request = LlmRequest() + + # Run the instruction processor + async for _ in request_processor.run_async(invocation_context, llm_request): + pass + + # Run the contents processor to move non-text parts + async for _ in contents_processor.run_async( + invocation_context, llm_request + ): + pass + + # System instruction should contain sequential reference IDs + expected_system = ( + 'Multiple files:\n\n[Reference to inline binary data: inline_data_0' + ' (type: image/png)]\n\n[Reference to file data: file_data_1 (URI:' + ' files/test1, type: text/plain)]\n\n[Reference to inline binary data:' + ' inline_data_2 (type: image/jpeg)]' + ) + assert llm_request.config.system_instruction == expected_system + + # All non-text parts should be in user content + assert len(llm_request.contents) == 3 + # Each non-text part gets its own content object with 2 parts (text description + actual part) + for content in llm_request.contents: + assert len(content.parts) == 2 + + @pytest.mark.asyncio + async def test_static_instruction_only_text_parts(self, llm_backend): + """Test that static instruction with only text parts works normally.""" + static_content = types.Content( + role='user', + parts=[ + types.Part(text='First part'), + types.Part(text='Second part'), + ], + ) + agent = LlmAgent(name='test_agent', static_instruction=static_content) + + invocation_context = await _create_invocation_context(agent) + llm_request = LlmRequest() + + # Run the instruction processor + async for _ in request_processor.run_async(invocation_context, llm_request): + pass + + # Only text should be in system instruction + assert llm_request.config.system_instruction == 'First part\n\nSecond part' + # No user content should be created + assert len(llm_request.contents) == 0 + + @pytest.mark.asyncio + async def test_static_instruction_only_non_text_parts(self, llm_backend): + """Test that static instruction with only non-text parts works correctly.""" + static_content = types.Content( + role='user', + parts=[ + types.Part( + inline_data=types.Blob(data=b'data', mime_type='image/png') + ), + types.Part( + file_data=types.FileData( + file_uri='files/test', mime_type='text/plain' + ) + ), + ], + ) + agent = LlmAgent(name='test_agent', static_instruction=static_content) + + invocation_context = await _create_invocation_context(agent) + llm_request = LlmRequest() + + # Run the instruction processor + async for _ in request_processor.run_async(invocation_context, llm_request): + pass + + # Run the contents processor to move non-text parts + async for _ in contents_processor.run_async( + invocation_context, llm_request + ): + pass + + # System instruction should contain only references + expected_system = ( + '[Reference to inline binary data: inline_data_0 (type:' + ' image/png)]\n\n[Reference to file data: file_data_1 (URI: files/test,' + ' type: text/plain)]' + ) + assert llm_request.config.system_instruction == expected_system + + # All parts should be in user content + assert len(llm_request.contents) == 2 + # Each non-text part gets its own content object with 2 parts (text description + actual part) + for content in llm_request.contents: + assert len(content.parts) == 2 diff --git a/tests/unittests/models/test_llm_request.py b/tests/unittests/models/test_llm_request.py index b0722291..2c2f6a0a 100644 --- a/tests/unittests/models/test_llm_request.py +++ b/tests/unittests/models/test_llm_request.py @@ -625,7 +625,7 @@ def test_append_instructions_content_extracts_text_only(): def test_append_instructions_content_with_non_text_parts(): - """Test that non-text parts in Content are ignored.""" + """Test that non-text parts in Content are processed with references.""" request = LlmRequest() # Create Content with text and non-text parts @@ -640,14 +640,28 @@ def test_append_instructions_content_with_non_text_parts(): ], ) - request.append_instructions(content) + user_contents = request.append_instructions(content) - # Only text parts should be extracted - assert request.config.system_instruction == 'Text instruction\n\nMore text' + # Text parts should be extracted with references to non-text parts + expected_system = ( + 'Text instruction\n\n' + '[Reference to inline binary data: inline_data_0 (type: text/plain)]\n\n' + 'More text' + ) + assert request.config.system_instruction == expected_system + + # Should return user content for the non-text part + assert len(user_contents) == 1 + assert user_contents[0].role == 'user' + assert len(user_contents[0].parts) == 2 + assert ( + user_contents[0].parts[0].text == 'Referenced inline data: inline_data_0' + ) + assert user_contents[0].parts[1].inline_data.data == b'file_data' def test_append_instructions_content_no_text_parts(): - """Test that Content with no text parts does nothing.""" + """Test that Content with no text parts processes non-text parts with references.""" request = LlmRequest() # Set initial system instruction @@ -663,10 +677,23 @@ def test_append_instructions_content_no_text_parts(): ], ) - request.append_instructions(content) + user_contents = request.append_instructions(content) - # Should remain unchanged since no text to extract - assert request.config.system_instruction == 'Initial' + # Should add reference to non-text part to system instruction + expected_system = ( + 'Initial\n\n[Reference to inline binary data: inline_data_0 (type:' + ' text/plain)]' + ) + assert request.config.system_instruction == expected_system + + # Should return user content for the non-text part + assert len(user_contents) == 1 + assert user_contents[0].role == 'user' + assert len(user_contents[0].parts) == 2 + assert ( + user_contents[0].parts[0].text == 'Referenced inline data: inline_data_0' + ) + assert user_contents[0].parts[1].inline_data.data == b'file_data' def test_append_instructions_content_empty_text_parts(): @@ -725,3 +752,87 @@ def test_append_instructions_warning_unsupported_system_instruction_type( assert ( 'Cannot append to system_instruction of unsupported type' in caplog.text ) + + +@pytest.mark.parametrize('llm_backend', ['GOOGLE_AI', 'VERTEX']) +def test_append_instructions_with_mixed_content(llm_backend): + """Test append_instructions with mixed text and non-text content.""" + request = LlmRequest() + + # Create static instruction with mixed content + static_content = types.Content( + role='user', + parts=[ + types.Part(text='Analyze this:'), + types.Part( + inline_data=types.Blob( + data=b'test_data', + mime_type='image/png', + display_name='test.png', + ) + ), + types.Part(text='Focus on details.'), + types.Part( + file_data=types.FileData( + file_uri='files/doc123', + mime_type='text/plain', + display_name='document.txt', + ) + ), + ], + ) + + user_contents = request.append_instructions(static_content) + + # System instruction should contain text with references + expected_system = ( + 'Analyze this:\n\n[Reference to inline binary data: inline_data_0' + " ('test.png', type: image/png)]\n\nFocus on details.\n\n[Reference to" + " file data: file_data_1 ('document.txt', URI: files/doc123, type:" + ' text/plain)]' + ) + assert request.config.system_instruction == expected_system + + # Should return user contents for non-text parts + assert len(user_contents) == 2 + + # Check inline_data content + assert user_contents[0].role == 'user' + assert len(user_contents[0].parts) == 2 + assert ( + user_contents[0].parts[0].text == 'Referenced inline data: inline_data_0' + ) + assert user_contents[0].parts[1].inline_data.data == b'test_data' + assert user_contents[0].parts[1].inline_data.display_name == 'test.png' + + # Check file_data content + assert user_contents[1].role == 'user' + assert len(user_contents[1].parts) == 2 + assert user_contents[1].parts[0].text == 'Referenced file data: file_data_1' + assert user_contents[1].parts[1].file_data.file_uri == 'files/doc123' + assert user_contents[1].parts[1].file_data.display_name == 'document.txt' + + +@pytest.mark.parametrize('llm_backend', ['GOOGLE_AI', 'VERTEX']) +def test_append_instructions_with_only_text_parts(llm_backend): + """Test append_instructions with only text parts.""" + request = LlmRequest() + + static_content = types.Content( + role='user', + parts=[ + types.Part(text='First instruction'), + types.Part(text='Second instruction'), + ], + ) + + user_contents = request.append_instructions(static_content) + + # Should only have text in system instruction + assert ( + request.config.system_instruction + == 'First instruction\n\nSecond instruction' + ) + + # Should return empty list since no non-text parts + assert user_contents == []