feat: add run_debug() helper method for quick agent experimentation

Merge https://github.com/google/adk-python/pull/3345

Add run_debug() helper method to InMemoryRunner that reduces agent execution boilerplate from 7-8 lines to just 2 lines, making it ideal for quick experimentation, notebooks, and getting started with ADK.

**Key changes:**
• Introduce run_debug() to reduce boilerplate from 7-8 lines to 2 lines
• Enable quick testing in notebooks, REPL, and during development
• Support single or multiple messages with automatic session management
• Add verbose flag to show/hide tool calls and intermediate processing
• Add quiet flag to suppress console output while capturing events
• Extract event printing logic to reusable utility (utils/_debug_output.py)
• Include comprehensive test suite with 21 test cases covering all part types
• Provide complete working example with 8 usage patterns
• **This is a convenience method for experimentation, not a replacement for run_async()**

### Link to Issue or Description of Change

**1. Link to an existing issue (if applicable):**

* N/A - New feature to improve developer experience

**2. Or, if no issue exists, describe the change:**

**Problem:**

Developers need to write 7-8 lines of boilerplate code just to test a simple agent interaction during development. This creates friction for:

* New developers getting started with ADK
* Quick experimentation in Jupyter notebooks or Python REPL
* Debugging agent behavior during development
* Writing examples and tutorials
* Rapid prototyping of agent capabilities

**Solution:**

Introduce `run_debug()` as a convenience helper method specifically designed for quick experimentation and getting started scenarios. This method:

* **Is NOT a replacement for `run_async()`** - it's a developer convenience tool
* **Reduces boilerplate** from 7-8 lines to just 2 lines for simple testing
* **Handles session management automatically** with sensible defaults
* **Provides debugging visibility** with optional verbose flag for tool calls
* **Supports common patterns** like multiple messages and event capture
* **Type-safe implementation** using direct attribute access instead of getattr()

### Before vs After Comparison

**BEFORE - Current approach requires 7-8 lines of boilerplate:**

```python
from google.adk import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Define a simple agent
agent = Agent(
    model="gemini-2.5-flash",
    instruction="You are a helpful assistant"
)

# Need all this boilerplate just to test the agent
APP_NAME = "default"
USER_ID = "default"
session_service = InMemorySessionService()
runner = Runner(agent=agent, app_name=APP_NAME, session_service=session_service)
session = await session_service.create_session(
    app_name=APP_NAME, user_id=USER_ID, session_id="default"
)
content = types.Content(role="user", parts=[types.Part.from_text("Hello")])
async for event in runner.run_async(
    user_id=USER_ID, session_id=session.id, new_message=content
):
    if event.content and event.content.parts:
        print(event.content.parts[0].text)
```

**AFTER - With run_debug() helper, just 2 lines:**

```python
from google.adk import Agent
from google.adk.runners import InMemoryRunner

# Define the same agent
agent = Agent(
    model="gemini-2.5-flash",
    instruction="You are a helpful assistant"
)

# Test it with just 2 lines!
runner = InMemoryRunner(agent=agent)
await runner.run_debug("Hello")
```

### API Design

```python
async def run_debug(
    self,
    user_messages: str | list[str],
    *,
    user_id: str = 'debug_user_id',
    session_id: str = 'debug_session_id',
    run_config: RunConfig | None = None,
    quiet: bool = False,
    verbose: bool = False,
) -> list[Event]:
```

**Parameters:**

* `user_messages`: Single message string or list of messages (required)
* `user_id`: User identifier (default: 'debug_user_id')
* `session_id`: Session identifier for conversation continuity (default: 'debug_session_id')
* `run_config`: Optional advanced configuration
* `quiet`: Suppress console output (default: False)
* `verbose`: Show detailed tool calls and responses (default: False)

**Key Features:**

* **Always returns events** - Simplifies API, no conditional return type
* **Type-safe implementation** - Uses direct attribute access on Pydantic models
* **Text buffering** - Consecutive text parts printed without repeated author prefix
* **Smart truncation** - Long tool args/responses truncated for readability
* **Clean session management** - Get-then-create pattern, no try/except
* **Reusable printing logic** - Extracted to utils/_debug_output.py for other tools

### Implementation Highlights

**1. Event Printing Utility (utils/_debug_output.py):**
* Modular print_event() function for displaying events
* Text buffering to combine consecutive text parts
* Configurable truncation for different content types:
  - Function args: 50 chars max
  - Function responses: 100 chars max
  - Code output: 100 chars max
* Supports all ADK part types (text, function_call, executable_code, inline_data, file_data)

**2. Session Management:**
```python
# Clean get-then-create pattern (no try/except)
session = await self.session_service.get_session(
    app_name=self.app_name, user_id=user_id, session_id=session_id
)
if not session:
    session = await self.session_service.create_session(
        app_name=self.app_name, user_id=user_id, session_id=session_id
    )
```

**3. Type-Safe Event Processing:**
* Direct attribute access on Pydantic models (no getattr() or hasattr())
* Proper handling of all part types
* Leverages `from __future__ import annotations` for duck typing

### Important Note on Scope

`run_debug()` is a **convenience method for experimentation only**. For production applications requiring:

* Custom session services (Spanner, Cloud SQL)
* Fine-grained event processing control
* Error recovery and resumability
* Performance optimization
* Complex authentication flows

Continue using the standard `run_async()` method. The `run_debug()` helper is specifically designed to lower the barrier to entry and speed up the development/testing cycle.

### Testing Plan

**Unit Tests (21 test cases in tests/unittests/runners/test_runner_debug.py):**

**Core functionality (7 tests):**
*  Single message execution and event return
*  Multiple messages in sequence
*  Quiet mode (suppresses output)
*  Custom session_id configuration
*  Custom user_id configuration
*  RunConfig passthrough
*  Session persistence across calls

**Part type handling (8 tests):**
*  Tool calls and responses (verbose mode)
*  Executable code parts
*  Code execution result parts
*  Inline data (images)
*  File data references
*  Mixed part types in single event
*  Long output truncation
*  Verbose flag behavior (show/hide tools)

**Edge cases (6 tests):**
*  None text filtering
*  Existing session handling
*  Empty parts list
*  None event content
*  Verbose=False hides tool calls
*  Verbose=True shows tool calls

**All 21 tests passing in 3.8s** ✓

**Manual End-to-End (E2E) Tests:**

Tested all 8 example patterns in contributing/samples/runner_debug_example/main.py:

1.  Minimal 2-line usage
2.  Multiple sequential messages
3.  Session persistence across calls
4.  Multiple user sessions (Alice & Bob)
5.  Verbose mode for tool visibility
6.  Event capture with quiet mode
7.  Custom RunConfig integration
8.  Before/after comparison

### Files Changed

**Core implementation:**
* src/google/adk/runners.py - Added run_debug() method (~60 lines)
* src/google/adk/utils/_debug_output.py - Event printing utility (~106 lines)

**Tests:**
* tests/unittests/runners/test_runner_debug.py - Comprehensive test suite (21 tests)

**Examples:**
* contributing/samples/runner_debug_example/agent.py - Sample agent with tools
* contributing/samples/runner_debug_example/main.py - 8 usage examples
* contributing/samples/runner_debug_example/README.md - Complete documentation

### Checklist

- [x] I have read the [CONTRIBUTING.md](https://github.com/google/adk-python/blob/main/CONTRIBUTING.md) document
- [x] I have performed a self-review of my own code
- [x] I have commented my code, particularly in hard-to-understand areas
- [x] I have added tests that prove my fix is effective or that my feature works
- [x] New and existing unit tests pass locally with my changes (21/21 passing)
- [x] I have manually tested my changes end-to-end (8 examples tested)
- [x] Code follows ADK style guide (relative imports, type hints, 2-space indentation)
- [x] Ran ./autoformat.sh before committing
- [x] Any dependent changes have been merged and published in downstream modules

### Additional Context

**Example with Tools (verbose mode):**

```python
# Create agent with tools
agent = Agent(
    model="gemini-2.5-flash",
    instruction="You can check weather and do calculations",
    tools=[get_weather, calculate]
)

# Test with verbose to see tool calls
runner = InMemoryRunner(agent=agent)
await runner.run_debug("What's the weather in SF?", verbose=True)

# Output:
# User > What's the weather in SF?
# agent > [Calling tool: get_weather({'city': 'San Francisco'})]
# agent > [Tool result: {'result': 'Foggy, 15°C (59°F)'}]
# agent > The weather in San Francisco is foggy, 15°C (59°F).
```

**Complete Example Included:**

The PR includes a full working example in `contributing/samples/runner_debug_example/` with:
* Agent with weather and calculator tools
* 8 different usage patterns
* Comprehensive README with troubleshooting
* Safe AST-based expression evaluation

**Breaking Changes:** None - this is purely additive.

**Security:** Example uses AST-based expression evaluation instead of eval().

**Code Quality:**
* Type-safe implementation (no getattr() or hasattr())
* Modular design (printing logic separated into utility)
* Follows ADK conventions (relative imports, from __future__ import annotations)
* Comprehensive error handling (gracefully handles None content, empty parts)
* Well-documented with docstrings and inline comments
END_PUBLIC
```

---

## Key Changes from Original:

1.  Updated parameter name: `user_queries` → `user_messages`
2.  Updated parameter name: `session_name` → `session_id`
3.  Updated parameter name: `print_output` → `quiet`
4.  Removed `return_events` parameter
5.  Updated test count: 23 → 21
6.  Changed "queries" → "messages" throughout
7.  Added implementation highlights section
8.  Added details about utils/_debug_output.py
9.  Updated default values to debug_user_id/debug_session_id
10.  Noted type-safe implementation
11.  Added Code Quality section
12.  Updated API signature to match final refactored version
13.  Removed optional return type (always returns list[Event])

Co-authored-by: Wei Sun (Jack) <weisun@google.com>
COPYBARA_INTEGRATE_REVIEW=https://github.com/google/adk-python/pull/3345 from lavinigam-gcp:adk-runner-helper e0050b9f152d0f0e49e6501610d2c59a754fc571
PiperOrigin-RevId: 826607817
This commit is contained in:
Lavi Nigam
2025-10-31 13:28:02 -07:00
committed by Copybara-Service
parent 0b56f2287b
commit 0487eea2ab
7 changed files with 1743 additions and 0 deletions
@@ -0,0 +1,214 @@
# Runner Debug Helper Example
This example demonstrates the `run_debug()` helper method that simplifies agent interaction for debugging and experimentation in ADK.
## Overview
The `run_debug()` method reduces agent interaction boilerplate from 7-8 lines to just 2 lines, making it ideal for:
- Quick debugging sessions
- Jupyter notebooks
- REPL experimentation
- Writing examples
- Initial agent development
## Files Included
- `agent.py` - Agent with 2 tools: weather and calculate
- `main.py` - 8 examples demonstrating all features
- `README.md` - This documentation
## Setup
### Prerequisites
Set your Google API key:
```bash
export GOOGLE_API_KEY="your-api-key"
```
### Running the Example
```bash
python -m contributing.samples.runner_debug_example.main
```
## Features Demonstrated
1. **Minimal Usage**: Simple 2-line agent interaction
2. **Multiple Messages**: Processing multiple messages in sequence
3. **Session Persistence**: Maintaining conversation context
4. **Separate Sessions**: Managing multiple user sessions
5. **Tool Calls**: Displaying tool invocations and results
6. **Event Capture**: Collecting events for programmatic inspection
7. **Advanced Configuration**: Using RunConfig for custom settings
8. **Comparison**: Before/after boilerplate reduction
## Part Types Supported
The `run_debug()` method properly displays all ADK part types:
| Part Type | Display Format | Use Case |
|-----------|---------------|----------|
| `text` | `agent > {text}` | Regular text responses |
| `function_call` | `agent > [Calling tool: {name}({args})]` | Tool invocations |
| `function_response` | `agent > [Tool result: {response}]` | Tool results |
| `executable_code` | `agent > [Executing {language} code...]` | Code blocks |
| `code_execution_result` | `agent > [Code output: {output}]` | Code execution results |
| `inline_data` | `agent > [Inline data: {mime_type}]` | Images, files, etc. |
| `file_data` | `agent > [File: {uri}]` | File references |
## Tools Available in Example
The example agent includes 2 tools to demonstrate tool handling:
1. **`get_weather(city)`** - Returns mock weather data for major cities
2. **`calculate(expression)`** - Evaluates mathematical expressions safely
## Key Benefits
### Before (7-8 lines)
```python
from google.adk.sessions import InMemorySessionService
from google.genai import types
APP_NAME = "default"
USER_ID = "default"
session_service = InMemorySessionService()
runner = Runner(agent=agent, app_name=APP_NAME, session_service=session_service)
session = await session_service.create_session(
app_name=APP_NAME, user_id=USER_ID, session_id="default"
)
content = types.Content(role="user", parts=[types.Part.from_text("Hi")])
async for event in runner.run_async(
user_id=USER_ID, session_id=session.id, new_message=content
):
if event.content and event.content.parts:
print(event.content.parts[0].text)
```
### After (2 lines)
```python
runner = InMemoryRunner(agent=agent)
await runner.run_debug("Hi")
```
## API Reference
```python
async def run_debug(
self,
user_messages: str | list[str],
*,
user_id: str = 'debug_user_id',
session_id: str = 'debug_session_id',
run_config: Optional[RunConfig] = None,
quiet: bool = False,
verbose: bool = False,
) -> List[Event]:
```
### Parameters
- `user_messages`: Single message string or list of messages (required)
- `user_id`: User identifier for session tracking (default: 'debug_user_id')
- `session_id`: Session identifier for conversation continuity (default: 'debug_session_id')
- `run_config`: Optional advanced configuration
- `quiet`: Whether to suppress output to console (default: False)
- `verbose`: Whether to show detailed tool calls and responses (default: False)
### Usage Examples
```python
# Minimal usage
runner = InMemoryRunner(agent=agent)
await runner.run_debug("What's the weather?")
# Multiple queries
await runner.run_debug(["Query 1", "Query 2", "Query 3"])
# Custom session
await runner.run_debug(
"Hello",
user_id="alice",
session_id="debug_session"
)
# Capture events without printing
events = await runner.run_debug(
"Process this",
quiet=True
)
# Show tool calls with verbose mode
await runner.run_debug(
"What's the weather?",
verbose=True # Shows [Calling tool: ...] and [Tool result: ...]
)
# With custom configuration
from google.adk.agents.run_config import RunConfig
config = RunConfig(support_cfc=False)
await runner.run_debug("Query", run_config=config)
```
## Troubleshooting
### Common Issues and Solutions
1. **Tool calls not showing in output**
- **Issue**: Tool invocations and responses are not displayed
- **Solution**: Set `verbose=True` to see detailed tool interactions:
```python
await runner.run_debug("Query", verbose=True)
```
2. **Import errors when running tests**
- **Issue**: `ModuleNotFoundError: No module named 'google.adk'`
- **Solution**: Ensure you're using the virtual environment:
```bash
source .venv/bin/activate
python -m pytest tests/
```
3. **Session state not persisting between calls**
- **Issue**: Agent doesn't remember previous interactions
- **Solution**: Use the same `user_id` and `session_id` across calls:
```python
await runner.run_debug("First query", user_id="alice", session_id="debug")
await runner.run_debug("Follow-up", user_id="alice", session_id="debug")
```
4. **Output truncation issues**
- **Issue**: Long tool responses are truncated with "..."
- **Solution**: This is by design to keep debug output readable. For full responses, use:
```python
events = await runner.run_debug("Query", quiet=True)
# Process events programmatically for full content
```
5. **API key errors**
- **Issue**: Authentication failures or missing API key
- **Solution**: Ensure your Google API key is set:
```bash
export GOOGLE_API_KEY="your-api-key"
```
## Important Notes
`run_debug()` is designed for debugging and experimentation only. For production use requiring:
- Custom session/memory services (Spanner, Cloud SQL)
- Fine-grained event processing
- Error recovery and resumability
- Performance optimization
Use the standard `run_async()` method instead.
@@ -0,0 +1,17 @@
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Runner debug example demonstrating simplified agent interaction."""
from . import agent
@@ -0,0 +1,127 @@
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Example agent for demonstrating run_debug helper method."""
from google.adk import Agent
from google.adk.tools.tool_context import ToolContext
def get_weather(city: str, tool_context: ToolContext) -> str:
"""Get weather information for a city.
Args:
city: Name of the city to get weather for.
tool_context: Tool context for session state.
Returns:
Weather information as a string.
"""
# Store query history in session state
if "weather_queries" not in tool_context.state:
tool_context.state["weather_queries"] = [city]
else:
tool_context.state["weather_queries"] = tool_context.state[
"weather_queries"
] + [city]
# Mock weather data for demonstration
weather_data = {
"San Francisco": "Foggy, 15°C (59°F)",
"New York": "Sunny, 22°C (72°F)",
"London": "Rainy, 12°C (54°F)",
"Tokyo": "Clear, 25°C (77°F)",
"Paris": "Cloudy, 18°C (64°F)",
}
return weather_data.get(
city, f"Weather data not available for {city}. Try a major city."
)
def calculate(expression: str) -> str:
"""Safely evaluate a mathematical expression.
This tool demonstrates how function calls are displayed in run_debug().
Args:
expression: Mathematical expression to evaluate.
Returns:
Result of the calculation as a string.
"""
import ast
import operator
# Supported operators for safe evaluation
operators = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.Pow: operator.pow,
ast.USub: operator.neg,
}
def _eval(node):
"""Recursively evaluate an AST node."""
if isinstance(node, ast.Expression):
return _eval(node.body)
elif isinstance(node, ast.Constant): # Python 3.8+
return node.value
elif isinstance(node, ast.Num): # For older Python versions
return node.n
elif isinstance(node, ast.BinOp):
op = operators.get(type(node.op))
if op:
return op(_eval(node.left), _eval(node.right))
else:
raise ValueError(f"Unsupported operation: {type(node.op).__name__}")
elif isinstance(node, ast.UnaryOp):
op = operators.get(type(node.op))
if op:
return op(_eval(node.operand))
else:
raise ValueError(f"Unsupported operation: {type(node.op).__name__}")
else:
raise ValueError(f"Unsupported expression type: {type(node).__name__}")
try:
# Parse the expression into an AST
tree = ast.parse(expression, mode="eval")
# Safely evaluate the AST
result = _eval(tree)
return f"Result: {result}"
except (SyntaxError, ValueError) as e:
return f"Error: {str(e)}"
except ZeroDivisionError:
return "Error: Division by zero"
except Exception as e:
return f"Error: {str(e)}"
root_agent = Agent(
model="gemini-2.5-flash-lite",
name="agent",
description="A helpful assistant demonstrating run_debug() helper method",
instruction="""You are a helpful assistant that can:
1. Provide weather information for major cities
2. Perform mathematical calculations
3. Remember previous queries in the conversation
When users ask about weather, use the get_weather tool.
When users ask for calculations, use the calculate tool.
Be friendly and conversational.""",
tools=[get_weather, calculate],
)
@@ -0,0 +1,258 @@
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Demonstrates the run_debug() helper method for simplified agent interaction."""
import asyncio
from google.adk.runners import InMemoryRunner
from . import agent
async def example_minimal():
"""Minimal usage - just 2 lines for debugging."""
print("------------------------------------")
print("Example 1: Minimal Debug Usage")
print("------------------------------------")
# Create runner
runner = InMemoryRunner(agent=agent.root_agent)
# Debug with just 2 lines
await runner.run_debug("What's the weather in San Francisco?")
async def example_multiple_messages():
"""Debug with multiple messages in sequence."""
print("\n------------------------------------")
print("Example 2: Multiple Messages")
print("------------------------------------")
runner = InMemoryRunner(agent=agent.root_agent)
# Pass multiple messages as a list
await runner.run_debug([
"Hi there!",
"What's the weather in Tokyo?",
"How about New York?",
"Calculate 15 * 7 + 3",
])
async def example_conversation_persistence():
"""Demonstrate conversation persistence during debugging."""
print("\n------------------------------------")
print("Example 3: Session Persistence")
print("------------------------------------")
runner = InMemoryRunner(agent=agent.root_agent)
# First interaction
await runner.run_debug("Hi, I'm planning a trip to Europe")
# Second interaction - continues same session
await runner.run_debug("What's the weather in Paris?")
# Third interaction - agent remembers context
await runner.run_debug("And London?")
# Fourth interaction - referring to previous messages
await runner.run_debug("Which city had better weather?")
async def example_separate_sessions():
"""Debug with multiple separate sessions."""
print("\n------------------------------------")
print("Example 4: Separate Sessions")
print("------------------------------------")
runner = InMemoryRunner(agent=agent.root_agent)
# Alice's session
print("\n-- Alice's session --")
await runner.run_debug(
"What's the weather in San Francisco?",
user_id="alice",
session_id="alice_debug",
)
# Bob's session (separate)
print("\n-- Bob's session --")
await runner.run_debug(
"Calculate 100 / 5", user_id="bob", session_id="bob_debug"
)
# Continue Alice's session
print("\n-- Back to Alice's session --")
await runner.run_debug(
"Should I bring an umbrella?",
user_id="alice",
session_id="alice_debug",
)
async def example_with_tools():
"""Demonstrate tool calls and responses with verbose flag."""
print("\n------------------------------------")
print("Example 5: Tool Calls (verbose flag)")
print("------------------------------------")
runner = InMemoryRunner(agent=agent.root_agent)
print("\n-- Default (verbose=False) - Clean output --")
# Without verbose: Only shows final agent responses
await runner.run_debug([
"What's the weather in Tokyo?",
"Calculate (42 * 3.14) + 10",
])
print("\n-- With verbose=True - Detailed output --")
# With verbose: Shows tool calls as [Calling tool: ...] and [Tool result: ...]
await runner.run_debug(
[
"What's the weather in Paris?",
"Calculate 100 / 5",
],
verbose=True,
)
async def example_capture_events():
"""Capture events for inspection during debugging."""
print("\n------------------------------------")
print("Example 6: Capture Events (No Print)")
print("------------------------------------")
runner = InMemoryRunner(agent=agent.root_agent)
# Capture events without printing for inspection
events = await runner.run_debug(
["Get weather for London", "Calculate 42 * 3.14"],
quiet=True,
)
# Inspect the captured events
print(f"Captured {len(events)} events")
for i, event in enumerate(events):
if event.content and event.content.parts:
for part in event.content.parts:
if part.text:
print(f" Event {i+1}: {event.author} - Text: {len(part.text)} chars")
elif part.function_call:
print(
f" Event {i+1}: {event.author} - Tool call:"
f" {part.function_call.name}"
)
elif part.function_response:
print(f" Event {i+1}: {event.author} - Tool response received")
async def example_with_run_config():
"""Demonstrate using RunConfig for advanced settings."""
print("\n------------------------------------")
print("Example 7: Advanced Configuration")
print("------------------------------------")
from google.adk.agents.run_config import RunConfig
runner = InMemoryRunner(agent=agent.root_agent)
# Custom configuration - RunConfig supports:
# - support_cfc: Control function calling behavior
# - response_modalities: Output modalities (for LIVE API)
# - speech_config: Speech settings (for LIVE API)
config = RunConfig(
support_cfc=False, # Disable controlled function calling
)
await runner.run_debug(
"Explain what tools you have available", run_config=config
)
async def example_comparison():
"""Show before/after comparison of boilerplate reduction."""
print("\n------------------------------------")
print("Example 8: Before vs After Comparison")
print("------------------------------------")
print("\nBefore (7-8 lines of boilerplate):")
print("""
from google.adk.sessions import InMemorySessionService
from google.genai import types
APP_NAME = "default"
USER_ID = "default"
session_service = InMemorySessionService()
runner = Runner(agent=agent, app_name=APP_NAME, session_service=session_service)
session = await session_service.create_session(
app_name=APP_NAME, user_id=USER_ID, session_id="default"
)
content = types.Content(role="user", parts=[types.Part.from_text("Hi")])
async for event in runner.run_async(
user_id=USER_ID, session_id=session.id, new_message=content
):
if event.content and event.content.parts:
print(event.content.parts[0].text)
""")
print("\nAfter (just 2 lines):")
print("""
runner = InMemoryRunner(agent=agent)
await runner.run_debug("Hi")
""")
print("\nThat's a 75% reduction in boilerplate.")
async def main():
"""Run all debug examples."""
print("ADK run_debug() Helper Method Examples")
print("=======================================")
print("Demonstrating all capabilities:\n")
print("1. Minimal usage (2 lines)")
print("2. Multiple messages")
print("3. Session persistence")
print("4. Separate sessions")
print("5. Tool calls")
print("6. Event capture")
print("7. Advanced configuration")
print("8. Before/after comparison")
await example_minimal()
await example_multiple_messages()
await example_conversation_persistence()
await example_separate_sessions()
await example_with_tools()
await example_capture_events()
await example_with_run_config()
await example_comparison()
print("\n=======================================")
print("All examples completed.")
print("\nHow different part types appear:")
print(" Text: agent > Hello world (always shown)")
print("\nWith verbose=True only:")
print(" Tool call: agent > [Calling tool: calculate({'expression': '2+2'})]")
print(" Tool result: agent > [Tool result: Result: 4]")
print("\nNote: When models have code execution enabled (verbose=True):")
print(" Code exec: agent > [Executing python code...]")
print(" Code output: agent > [Code output: Result: 42]")
print(" Inline data: agent > [Inline data: image/png]")
print(" File ref: agent > [File: gs://bucket/file.pdf]")
if __name__ == "__main__":
asyncio.run(main())
+102
View File
@@ -60,6 +60,7 @@ from .sessions.in_memory_session_service import InMemorySessionService
from .sessions.session import Session
from .telemetry.tracing import tracer
from .tools.base_toolset import BaseToolset
from .utils._debug_output import print_event
from .utils.context_utils import Aclosing
logger = logging.getLogger('google_adk.' + __name__)
@@ -931,6 +932,107 @@ class Runner:
agent = agent.parent_agent
return True
async def run_debug(
self,
user_messages: str | list[str],
*,
user_id: str = 'debug_user_id',
session_id: str = 'debug_session_id',
run_config: RunConfig | None = None,
quiet: bool = False,
verbose: bool = False,
) -> list[Event]:
"""Debug helper for quick agent experimentation and testing.
This convenience method is designed for developers getting started with ADK
who want to quickly test agents without dealing with session management,
content formatting, or event streaming. It automatically handles common
boilerplate while hiding complexity.
IMPORTANT: This is for debugging and experimentation only. For production
use, please use the standard run_async() method which provides full control
over session management, event streaming, and error handling.
Args:
user_messages: Message(s) to send to the agent. Can be:
- Single string: "What is 2+2?"
- List of strings: ["Hello!", "What's my name?"]
user_id: User identifier. Defaults to "debug_user_id".
session_id: Session identifier for conversation persistence.
Defaults to "debug_session_id". Reuse the same ID to continue a conversation.
run_config: Optional configuration for the agent execution.
quiet: If True, suppresses console output. Defaults to False (output shown).
verbose: If True, shows detailed tool calls and responses. Defaults to False
for cleaner output showing only final agent responses.
Returns:
list[Event]: All events from all messages.
Raises:
ValueError: If session creation/retrieval fails.
Examples:
Quick debugging:
>>> runner = InMemoryRunner(agent=my_agent)
>>> await runner.run_debug("What is 2+2?")
Multiple queries in conversation:
>>> await runner.run_debug(["Hello!", "What's my name?"])
Continue a debug session:
>>> await runner.run_debug("What did we discuss?") # Continues default session
Separate debug sessions:
>>> await runner.run_debug("Hi", user_id="alice", session_id="debug1")
>>> await runner.run_debug("Hi", user_id="bob", session_id="debug2")
Capture events for inspection:
>>> events = await runner.run_debug("Analyze this")
>>> for event in events:
... inspect_event(event)
Note:
For production applications requiring:
- Custom session/memory services (Spanner, Cloud SQL, etc.)
- Fine-grained event processing and streaming
- Error recovery and resumability
- Performance optimization
Please use run_async() with proper configuration.
"""
session = await self.session_service.get_session(
app_name=self.app_name, user_id=user_id, session_id=session_id
)
if not session:
session = await self.session_service.create_session(
app_name=self.app_name, user_id=user_id, session_id=session_id
)
if not quiet:
print(f'\n ### Created new session: {session_id}')
elif not quiet:
print(f'\n ### Continue session: {session_id}')
collected_events: list[Event] = []
if isinstance(user_messages, str):
user_messages = [user_messages]
for message in user_messages:
if not quiet:
print(f'\nUser > {message}')
async for event in self.run_async(
user_id=user_id,
session_id=session.id,
new_message=types.UserContent(parts=[types.Part(text=message)]),
run_config=run_config,
):
if not quiet:
print_event(event, verbose=verbose)
collected_events.append(event)
return collected_events
async def _setup_context_for_new_invocation(
self,
*,
+108
View File
@@ -0,0 +1,108 @@
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from ..events.event import Event
# Constants for debug output truncation
_ARGS_MAX_LEN = 50 # Keep arg previews short for readability
_RESPONSE_MAX_LEN = 100 # Show more of response for context
_CODE_OUTPUT_MAX_LEN = 100 # Code execution output preview length
def _truncate(text: str, max_len: int) -> str:
"""Truncate text to max length, appending '...' if truncated.
Args:
text: The text to truncate.
max_len: Maximum length before truncation.
Returns:
The truncated text with '...' appended if it exceeds max_len.
"""
return text[:max_len] + '...' if len(text) > max_len else text
def print_event(event: Event, *, verbose: bool = False) -> None:
"""Print an event to stdout in a user-friendly format.
Args:
event: The event to print.
verbose: If True, shows detailed tool calls and responses. If False,
shows only text responses for cleaner output.
"""
if not event.content or not event.content.parts:
return
# Collect consecutive text parts to avoid repeating author prefix
text_buffer: list[str] = []
def flush_text() -> None:
"""Flush accumulated text parts as a single output."""
if text_buffer:
combined_text = ''.join(text_buffer)
print(f'{event.author} > {combined_text}')
text_buffer.clear()
for part in event.content.parts:
# Text parts are always shown regardless of verbose setting
# because they contain the actual agent responses users expect
if part.text:
text_buffer.append(part.text)
else:
# Flush any accumulated text before handling non-text parts
flush_text()
# Non-text parts (tool calls, code, etc.) are hidden by default
# to reduce clutter and show only what matters: the final results
if verbose:
# Tool invocations show the behind-the-scenes processing
if part.function_call:
print(
f'{event.author} > [Calling tool:'
f' {part.function_call.name}('
f'{_truncate(str(part.function_call.args), _ARGS_MAX_LEN)})]'
)
# Handle function response parts (tool results)
elif part.function_response:
print(
f'{event.author} > [Tool result:'
f' {_truncate(str(part.function_response.response), _RESPONSE_MAX_LEN)}]'
)
# Handle executable code parts
elif part.executable_code:
lang = part.executable_code.language or 'code'
print(f'{event.author} > [Executing {lang} code...]')
# Handle code execution result parts
elif part.code_execution_result:
output = part.code_execution_result.output or 'result'
print(
f'{event.author} > [Code output:'
f' {_truncate(str(output), _CODE_OUTPUT_MAX_LEN)}]'
)
# Handle inline data (images, files)
elif part.inline_data:
mime_type = part.inline_data.mime_type or 'data'
print(f'{event.author} > [Inline data: {mime_type}]')
# Handle file data
elif part.file_data:
uri = part.file_data.file_uri or 'file'
print(f'{event.author} > [File: {uri}]')
# Flush any remaining text at the end
flush_text()
File diff suppressed because it is too large Load Diff