refactor(agent/bedrock_adapter): table the discovery step/log pairs, walrus early-returns, comprehension for system blocks

This commit is contained in:
Teknium
2026-09-02 22:51:19 -07:00
parent 93af152fdb
commit fd28dc7d26
+26 -41
View File
@@ -228,13 +228,11 @@ def is_streaming_access_denied_error(exc: BaseException) -> bool:
msg = str(exc).lower()
if "invokemodelwithresponsestream" not in msg:
return False
try:
with suppress(ImportError): # pragma: no cover — botocore always present with boto3
from botocore.exceptions import ClientError
except ImportError: # pragma: no cover — botocore always present with boto3
ClientError = None # type: ignore[assignment]
if ClientError is not None and isinstance(exc, ClientError):
code = (getattr(exc, "response", None) or {}).get("Error", {}).get("Code", "")
return code in ("AccessDeniedException", "UnauthorizedException")
if isinstance(exc, ClientError):
code = (getattr(exc, "response", None) or {}).get("Error", {}).get("Code", "")
return code in ("AccessDeniedException", "UnauthorizedException")
return "not authorized" in msg or "accessdenied" in msg
@@ -294,11 +292,9 @@ def resolve_bedrock_runtime_region(config: Optional[Dict[str, Any]] = None) -> s
non-runtime Bedrock endpoint must use this so auxiliary calls never leave the primary
runtime's region when config and ambient AWS env disagree. Pass *config* to skip disk."""
if config is None:
try:
with suppress(Exception):
from hermes_cli.config import load_config_readonly
config = load_config_readonly()
except Exception:
config = {}
cfg_region = str(((config or {}).get("bedrock") or {}).get("region") or "").strip()
return cfg_region or resolve_bedrock_region()
@@ -401,9 +397,9 @@ def strip_cache_points(kwargs: Dict[str, Any], placement: str) -> Dict[str, Any]
return {**kwargs, "system": cleaned} if placement == "system" else {**kwargs, "toolConfig": {**tool_config, "tools": cleaned}}
if placement == "messages":
messages = kwargs.get("messages")
if not isinstance(messages, list):
return kwargs
cleaned_contents = [_without_cache_points(msg.get("content") if isinstance(msg, dict) else None) for msg in messages]
cleaned_contents = [
_without_cache_points(msg.get("content") if isinstance(msg, dict) else None) for msg in messages
] if isinstance(messages, list) else []
if all(content is None for content in cleaned_contents):
return kwargs
return {**kwargs, "messages": [
@@ -473,8 +469,7 @@ def _image_block_from_data_url(url: str) -> Dict:
raw_bytes = base64.b64decode(data)
except Exception:
raw_bytes = data.encode("utf-8")
image_format = media_type.split("/")[-1] if "/" in media_type else "jpeg"
return {"image": {"format": image_format, "source": {"bytes": raw_bytes}}}
return {"image": {"format": media_type.split("/")[-1] if "/" in media_type else "jpeg", "source": {"bytes": raw_bytes}}}
def _convert_content_to_converse(content) -> List[Dict]:
@@ -500,12 +495,8 @@ def _convert_content_to_converse(content) -> List[Dict]:
def _system_blocks(content) -> List[Dict]:
"""System content → text blocks; blank parts are dropped, not placeholder-filled."""
parts = [content] if isinstance(content, str) else content if isinstance(content, list) else []
blocks: List[Dict] = []
for part in parts:
text = part.get("text", "") if isinstance(part, dict) and part.get("type") == "text" else part
if isinstance(text, str) and text.strip():
blocks.append({"text": text})
return blocks
texts = [part.get("text", "") if isinstance(part, dict) and part.get("type") == "text" else part for part in parts]
return [{"text": text} for text in texts if isinstance(text, str) and text.strip()]
def _tool_use_block(tool_use_id, name, input_dict) -> Dict:
@@ -566,10 +557,8 @@ def _assistant_blocks(msg: Dict, content) -> List[Dict]:
authoritative; otherwise redacted thinking from ``reasoning_details`` (byte-for-byte
round-trip), then text, then tool calls."""
ordered_blocks = msg.get("bedrock_content_blocks")
if isinstance(ordered_blocks, list) and ordered_blocks:
content_blocks = _replay_ordered_blocks(ordered_blocks)
if content_blocks:
return content_blocks
if isinstance(ordered_blocks, list) and (content_blocks := _replay_ordered_blocks(ordered_blocks)):
return content_blocks
content_blocks = []
for detail in (msg.get("reasoning_details") or []):
if isinstance(detail, dict) and detail.get("type") == "redacted_thinking":
@@ -788,7 +777,7 @@ def stream_converse_with_callbacks(
on_reasoning(delta["reasoningContent"])
elif "contentBlockStop" in event:
if current_tool is not None:
input_dict = _parse_tool_args(current_tool["input_json"]) if current_tool["input_json"] else {}
input_dict = _parse_tool_args(current_tool["input_json"]) # "" → {} via the JSON-error path
parts.tool_calls.append(_tool_call_ns(current_tool["toolUseId"], current_tool["name"], input_dict))
if current_block_index is not None and current_block_index in stream_blocks:
stream_blocks[current_block_index]["toolUse"]["input"] = input_dict
@@ -919,13 +908,11 @@ def _list_foundation_models(client, filter_set: set, models: List[Dict[str, Any]
def _list_inference_profiles(client, filter_set: set, models: List[Dict[str, Any]]) -> None:
"""Append active cross-region inference profiles whose IDs are not already present (paginated)."""
profiles = []
next_token = None
profiles, next_token = [], None
while True:
response = client.list_inference_profiles(**({"nextToken": next_token} if next_token else {}))
profiles.extend(response.get("inferenceProfileSummaries", []))
next_token = response.get("nextToken")
if not next_token:
if not (next_token := response.get("nextToken")):
break
seen_ids = {m["id"].lower() for m in models}
for profile in profiles:
@@ -955,14 +942,14 @@ def discover_bedrock_models(region: str, provider_filter: Optional[List[str]] =
return []
models: List[Dict[str, Any]] = []
filter_set = {f.lower() for f in (provider_filter or [])}
try:
_list_foundation_models(client, filter_set, models)
except Exception as e:
logger.warning("Failed to list Bedrock foundation models: %s", e)
try:
_list_inference_profiles(client, filter_set, models)
except Exception as e:
logger.debug("Skipping inference profile discovery: %s", e)
for step, log, message in (
(_list_foundation_models, logger.warning, "Failed to list Bedrock foundation models: %s"),
(_list_inference_profiles, logger.debug, "Skipping inference profile discovery: %s"),
):
try:
step(client, filter_set, models)
except Exception as e:
log(message, e)
models.sort(key=lambda m: (0 if m["id"].startswith("global.") else 1, m["name"].lower()))
_discovery_cache[cache_key] = {"timestamp": time.time(), "models": models}
return models
@@ -1050,10 +1037,8 @@ def get_bedrock_context_length(model_id: str, region: str = "", probe: bool = Tr
"""Context window: live probe (if ``probe`` and ``region``) → static table → default. The
table is fallback only: a stale substring match silently caps the window (a 1M Opus
pinned to 200K via "opus-4"). ``probe=False`` / empty region skips the network call."""
if probe and region:
probed = probe_bedrock_context_length(model_id, region)
if probed:
return probed
if probe and region and (probed := probe_bedrock_context_length(model_id, region)):
return probed
model_lower = model_id.lower()
matches = [key for key in BEDROCK_CONTEXT_LENGTHS if key in model_lower]
return BEDROCK_CONTEXT_LENGTHS[max(matches, key=len)] if matches else BEDROCK_DEFAULT_CONTEXT_LENGTH