Files
adk-python/contributing/samples/adk_stale_agent/utils.py
T
Rohit YanamadalaandCopybara-Service cb19d0714c fix: Optimize Stale Agent with GraphQL and Search API to resolve 429 Quota errors
Merge https://github.com/google/adk-python/pull/3700

### Description
This PR refactors the `adk_stale_agent` to address `429 RESOURCE_EXHAUSTED` errors encountered during workflow execution. The previous implementation was inefficient in fetching issue history (using pagination over the REST API) and lacked server-side filtering, causing excessive API calls and huge token consumption that breached Gemini API quotas.

The new implementation switches to a **GraphQL-first approach**, implements server-side filtering via the Search API, adds robust concurrency controls, and significantly improves code maintainability through modular refactoring.

### Root Cause of Failure
The previous workflow failed with the following error due to passing too much context to the LLM and processing too many irrelevant issues:
```text
google.genai.errors.ClientError: 429 RESOURCE_EXHAUSTED.
Quota exceeded for metric: generativelanguage.googleapis.com/generate_content_paid_tier_input_token_count
```
### Key Changes

#### 1. Optimization: REST → GraphQL (`agent.py`)
*   **Old:** Fetched issue comments and timeline events using multiple paginated REST API calls (`/timeline`).
*   **New:** Implemented `get_issue_state` using a single **GraphQL** query. This fetches comments, `userContentEdits`, and specific timeline events (Labels, Renames) in one network request.
*   **Refactoring:** The complex analysis logic has been decomposed into focused helper functions (_fetch_graphql_data, _build_history_timeline, _replay_history_to_find_state) for better readability and testing.
*   **Configurable:** Added GRAPHQL_COMMENT_LIMIT and GRAPHQL_TIMELINE_LIMIT settings to tune context depth
*   **Impact:** Drastically reduces the data payload size and eliminates multiple API round-trips, significantly lowering the token count sent to the LLM.

#### 2. Optimization: Server-Side Filtering (`utils.py`)
*   **Old:** Fetched *all* open issues via REST and filtered them in Python memory.
*   **New:** Uses the GitHub Search API (`get_old_open_issue_numbers`) with `created:<DATE` syntax.
*   **Impact:** Only fetches issue numbers that actually meet the age threshold, preventing the agent from wasting cycles and tokens on brand-new issues.

#### 3. Concurrency & Rate Limiting (`main.py` & `settings.py`)
*   **Old:** Sequential execution loop.
*   **New:** Implemented `asyncio.gather` with a configurable `CONCURRENCY_LIMIT` (set to 3).
*   **New:** Added `urllib3` retry strategies (exponential backoff) in `utils.py` to handle GitHub API rate limits (HTTP 429) gracefully.

#### 4. Logic Improvements ("Ghost Edits")
*   **New Feature:** The agent now detects "Ghost Edits" (where an author updates the issue description without posting a new comment).
*   **Action:** If a silent edit is detected on a stale candidate, the agent now alerts maintainers instead of marking it stale, preventing false positives.

### File Comparison Summary

| File | Change |
| :--- | :--- |
| `main.py` | Switched from `InMemoryRunner` loop to `asyncio` chunked processing. Added execution timing and API usage logging. |
| `agent.py` | Replaced REST logic with GraphQL query. Added logic to handle silent body edits. Decomposed giant get_issue_state into helper functions with docstrings. Added _format_days helper. |
| `utils.py` | Added `HTTPAdapter` with Retries. Added `get_old_open_issue_numbers` using Search API. |
| `settings.py` | Removed `ISSUES_PER_RUN`; added configuration for CONCURRENCY_LIMIT, SLEEP_BETWEEN_CHUNKS, and GraphQL limits. |
| `PROMPT_INSTRUCTIONS.txt` | Simplified decision tree; removed date calculation responsibility from LLM. |

### Verification
The new logic minimizes token usage by offloading date calculations to Python and strictly limiting the context passed to the LLM to semantic intent analysis (e.g., "Is this a question?").

*   **Metric Check:** The workflow now tracks API calls per issue to ensure we stay within limits.
*   **Safety:** Silent edits by users now correctly reset the "Stale" timer.
*   **Maintainability:** All complex logic is now isolated in typed helper functions with comprehensive docstrings.

Co-authored-by: Xuan Yang <xygoogle@google.com>
COPYBARA_INTEGRATE_REVIEW=https://github.com/google/adk-python/pull/3700 from ryanaiagent:feat/improve-stale-agent 888064eff125ae74f7c3a9ad6c74f98de80243a2
PiperOrigin-RevId: 838885530
2025-12-01 12:25:51 -08:00

261 lines
6.7 KiB
Python

# 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 datetime import datetime
from datetime import timedelta
from datetime import timezone
import logging
import threading
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from adk_stale_agent.settings import GITHUB_TOKEN
from adk_stale_agent.settings import STALE_HOURS_THRESHOLD
import dateutil.parser
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
logger = logging.getLogger("google_adk." + __name__)
# --- API Call Counter for Monitoring ---
_api_call_count = 0
_counter_lock = threading.Lock()
def get_api_call_count() -> int:
"""
Returns the total number of API calls made since the last reset.
Returns:
int: The global count of API calls.
"""
with _counter_lock:
return _api_call_count
def reset_api_call_count() -> None:
"""Resets the global API call counter to zero."""
global _api_call_count
with _counter_lock:
_api_call_count = 0
def _increment_api_call_count() -> None:
"""
Atomically increments the global API call counter.
Required because the agent may run tools in parallel threads.
"""
global _api_call_count
with _counter_lock:
_api_call_count += 1
# --- Production-Ready HTTP Session with Exponential Backoff ---
# Configure the retry strategy:
retry_strategy = Retry(
total=6,
backoff_factor=2,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=[
"HEAD",
"GET",
"POST",
"PUT",
"DELETE",
"OPTIONS",
"TRACE",
"PATCH",
],
)
adapter = HTTPAdapter(max_retries=retry_strategy)
# Create a single, reusable Session object for connection pooling
_session = requests.Session()
_session.mount("https://", adapter)
_session.mount("http://", adapter)
_session.headers.update({
"Authorization": f"token {GITHUB_TOKEN}",
"Accept": "application/vnd.github.v3+json",
})
def get_request(url: str, params: Optional[Dict[str, Any]] = None) -> Any:
"""
Sends a GET request to the GitHub API with automatic retries.
Args:
url (str): The URL endpoint.
params (Optional[Dict[str, Any]]): Query parameters.
Returns:
Any: The JSON response parsed into a dict or list.
Raises:
requests.exceptions.RequestException: If retries are exhausted.
"""
_increment_api_call_count()
try:
response = _session.get(url, params=params or {}, timeout=60)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
logger.error(f"GET request failed for {url}: {e}")
raise
def post_request(url: str, payload: Any) -> Any:
"""
Sends a POST request to the GitHub API with automatic retries.
Args:
url (str): The URL endpoint.
payload (Any): The JSON payload.
Returns:
Any: The JSON response.
"""
_increment_api_call_count()
try:
response = _session.post(url, json=payload, timeout=60)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
logger.error(f"POST request failed for {url}: {e}")
raise
def patch_request(url: str, payload: Any) -> Any:
"""
Sends a PATCH request to the GitHub API with automatic retries.
Args:
url (str): The URL endpoint.
payload (Any): The JSON payload.
Returns:
Any: The JSON response.
"""
_increment_api_call_count()
try:
response = _session.patch(url, json=payload, timeout=60)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
logger.error(f"PATCH request failed for {url}: {e}")
raise
def delete_request(url: str) -> Any:
"""
Sends a DELETE request to the GitHub API with automatic retries.
Args:
url (str): The URL endpoint.
Returns:
Any: A success dict if 204, else the JSON response.
"""
_increment_api_call_count()
try:
response = _session.delete(url, timeout=60)
response.raise_for_status()
if response.status_code == 204:
return {"status": "success", "message": "Deletion successful."}
return response.json()
except requests.exceptions.RequestException as e:
logger.error(f"DELETE request failed for {url}: {e}")
raise
def error_response(error_message: str) -> Dict[str, Any]:
"""
Creates a standardized error response dictionary for tool outputs.
Args:
error_message (str): The error details.
Returns:
Dict[str, Any]: Standardized error object.
"""
return {"status": "error", "message": error_message}
def get_old_open_issue_numbers(
owner: str, repo: str, days_old: Optional[float] = None
) -> List[int]:
"""
Finds open issues older than the specified threshold using server-side filtering.
OPTIMIZATION:
Instead of fetching ALL issues and filtering in Python (which wastes API calls),
this uses the GitHub Search API `created:<DATE` syntax.
Args:
owner (str): Repository owner.
repo (str): Repository name.
days_old (Optional[float]): Filter issues older than this many days.
Defaults to STALE_HOURS_THRESHOLD / 24.
Returns:
List[int]: A list of issue numbers matching the criteria.
"""
if days_old is None:
days_old = STALE_HOURS_THRESHOLD / 24
now_utc = datetime.now(timezone.utc)
cutoff_dt = now_utc - timedelta(days=days_old)
cutoff_str = cutoff_dt.strftime("%Y-%m-%dT%H:%M:%SZ")
query = f"repo:{owner}/{repo} is:issue state:open created:<{cutoff_str}"
logger.info(
f"Searching for issues in '{owner}/{repo}' created before {cutoff_str}..."
)
issue_numbers = []
page = 1
url = "https://api.github.com/search/issues"
while True:
params = {"q": query, "per_page": 100, "page": page}
try:
data = get_request(url, params=params)
items = data.get("items", [])
if not items:
break
for item in items:
if "pull_request" not in item:
issue_numbers.append(item["number"])
if len(items) < 100:
break
page += 1
except requests.exceptions.RequestException as e:
logger.error(f"GitHub search failed on page {page}: {e}")
break
logger.info(f"Found {len(issue_numbers)} stale issues.")
return issue_numbers