mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-15 14:22:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

perf(yuanbao): bounded-concurrency inbound media resolve

Browse source
This commit is contained in:
loongfay 2026-06-16 22:41:40 +08:00 • committed by Teknium
parent 58e1647b49
commit 63c4100fe6
2 changed files with 409 additions and 45 deletions
Download patch file Download diff file Expand all files Collapse all files

212
gateway/platforms/yuanbao.py
View file

@ -179,6 +179,16 @@ OBSERVED_MEDIA_BACKFILL_LOOKBACK = 50
# Max number of resource references to resolve per inbound turn
OBSERVED_MEDIA_BACKFILL_MAX_RESOLVE_PER_TURN = 12
# Bounded concurrency for inbound media resolve/download.
# - 1 = sequential (legacy behavior, safe rollback knob)
# - 6 = default; aligns with the per-origin HTTP/1.1 ceiling browsers use,
# balances first-token latency vs. backend pressure
# - 12 = upper clamp; matches OBSERVED_MEDIA_BACKFILL_MAX_RESOLVE_PER_TURN
# Configured via config.yaml: platforms.yuanbao.extra.media_resolve_concurrency.
_DEFAULT_RESOLVE_CONCURRENCY = 6
_MIN_RESOLVE_CONCURRENCY = 1
_MAX_RESOLVE_CONCURRENCY = 12
class MarkdownProcessor:
"""Encapsulates all Markdown-related utilities for the Yuanbao platform.
@ -2813,45 +2823,94 @@ class MediaResolveMiddleware(InboundMiddleware):
Yuanbao COS hostnames resolve to private IPs, tripping the SSRF guard
in vision_tools. We download ourselves and return local cache paths.
"""
media_urls: List[str] = []
media_types: List[str] = []
Resolution runs with bounded concurrency
(``adapter.media_resolve_concurrency``); see :meth:`_resolve_ybres_refs`
for the same order-preserving / exception-isolated contract.
"""
# Pre-filter resolvable refs, preserving input order.
active: List[Tuple[str, str, str, str]] = []
for ref in media_refs:
kind = str(ref.get("kind") or "").strip().lower()
url = str(ref.get("url") or "").strip()
filename = str(ref.get("name") or "").strip()
if kind not in _RESOLVABLE_MEDIA_KINDS or not url:
continue
# Extract resourceId from the placeholder URL for cache dedup.
rid = ExtractContentMiddleware._parse_resource_id(url)
if rid and cls._append_cached_resource(adapter, rid, media_urls, media_types):
continue
active.append((kind, url, filename, rid))
try:
fetch_url = await cls._resolve_download_url(adapter, url)
except Exception as exc:
logger.warning(
"[%s] inbound media resolve failed: kind=%s url=%s err=%s",
adapter.name, kind, url, exc,
if not active:
return [], []
semaphore = asyncio.Semaphore(adapter.media_resolve_concurrency)
async def _resolve_one(
kind: str, url: str, filename: str, rid: str,
) -> Optional[Tuple[str, str]]:
# Cache dedup first — avoids the RPC + download entirely on a hit.
if rid:
hit = cls._get_cached_resource(rid)
if hit is not None:
logger.debug(
"[%s] resource cache hit: rid=%s path=%s",
adapter.name, rid, hit[0],
)
return hit
async with semaphore:
try:
fetch_url = await cls._resolve_download_url(adapter, url)
except Exception as exc:
logger.warning(
"[%s] inbound media resolve failed: kind=%s url=%s err=%s",
adapter.name, kind, url, exc,
)
return None
return await cls._download_and_cache(
adapter,
fetch_url=fetch_url,
kind=kind,
file_name=filename or None,
log_tag=f"placeholder_url={url[:80]}",
resource_id=rid,
)
continue
cached = await cls._download_and_cache(
adapter,
fetch_url=fetch_url,
kind=kind,
file_name=filename or None,
log_tag=f"placeholder_url={url[:80]}",
resource_id=rid,
)
if cached is None:
_t0 = time.monotonic()
results = await asyncio.gather(
*(_resolve_one(kind, url, filename, rid)
for kind, url, filename, rid in active),
return_exceptions=True,
)
_elapsed_ms = int((time.monotonic() - _t0) * 1000)
media_urls: List[str] = []
media_types: List[str] = []
_failed = 0
for (kind, url, _filename, _rid), result in zip(active, results):
if isinstance(result, BaseException):
logger.warning(
"[%s] inbound media resolve crashed: kind=%s url=%s err=%s",
adapter.name, kind, url[:80], result,
)
_failed += 1
continue
local_path, mime = cached
if result is None:
_failed += 1
continue
local_path, mime = result
media_urls.append(local_path)
media_types.append(mime)
# Batch summary: keep fields stable for offline aggregation
# (concurrency vs elapsed_ms is the core knob-tuning view).
logger.info(
"[%s] media resolve batch: scope=media concurrency=%d total=%d ok=%d failed=%d elapsed_ms=%d",
adapter.name,
adapter.media_resolve_concurrency,
len(active),
len(media_urls),
_failed,
_elapsed_ms,
)
return media_urls, media_types
@classmethod
@ -2862,36 +2921,86 @@ class MediaResolveMiddleware(InboundMiddleware):
*,
log_prefix: str,
) -> Tuple[List[str], List[str]]:
"""Resolve a list of ``(rid, kind, filename)`` ybres tuples to local paths.
"""Resolve ``(rid, kind, filename)`` ybres tuples to local paths.
Runs with bounded concurrency (``adapter.media_resolve_concurrency``)
so cold-start turns with many anchors don't pay ``N × (RPC + download)``
sequentially. Output order matches input; per-rid failures are isolated.
"""
# Pre-filter resolvable kinds, preserving input order.
active: List[Tuple[str, str, str]] = [
(rid, kind, filename)
for rid, kind, filename in refs
if kind in _RESOLVABLE_MEDIA_KINDS
]
if not active:
return [], []
semaphore = asyncio.Semaphore(adapter.media_resolve_concurrency)
async def _resolve_one(rid: str, kind: str, filename: str) -> Optional[Tuple[str, str]]:
# Cache dedup first — avoids the RPC + download entirely on a hit.
hit = cls._get_cached_resource(rid)
if hit is not None:
logger.debug(
"[%s] resource cache hit: rid=%s path=%s",
adapter.name, rid, hit[0],
)
return hit
async with semaphore:
try:
fresh_url = await cls._fetch_resource_url(adapter, rid)
except Exception as exc:
logger.warning(
"[%s] %s resolve failed: rid=%s kind=%s err=%s",
adapter.name, log_prefix, rid, kind, exc,
)
return None
return await cls._download_and_cache(
adapter,
fetch_url=fresh_url,
kind=kind,
file_name=filename or None,
log_tag=f"{log_prefix} rid={rid}",
resource_id=rid,
)
# return_exceptions=True isolates per-coroutine failures.
_t0 = time.monotonic()
results = await asyncio.gather(
*(_resolve_one(rid, kind, filename) for rid, kind, filename in active),
return_exceptions=True,
)
_elapsed_ms = int((time.monotonic() - _t0) * 1000)
media_paths: List[str] = []
mimes: List[str] = []
for rid, kind, filename in refs:
if kind not in _RESOLVABLE_MEDIA_KINDS:
continue
if cls._append_cached_resource(adapter, rid, media_paths, mimes):
continue
try:
fresh_url = await cls._fetch_resource_url(adapter, rid)
except Exception as exc:
_failed = 0
for (rid, kind, _filename), result in zip(active, results):
if isinstance(result, BaseException):
logger.warning(
"[%s] %s resolve failed: rid=%s kind=%s err=%s",
adapter.name, log_prefix, rid, kind, exc,
"[%s] %s resolve crashed: rid=%s kind=%s err=%s",
adapter.name, log_prefix, rid, kind, result,
)
_failed += 1
continue
cached = await cls._download_and_cache(
adapter,
fetch_url=fresh_url,
kind=kind,
file_name=filename or None,
log_tag=f"{log_prefix} rid={rid}",
resource_id=rid,
)
if cached is None:
if result is None:
_failed += 1
continue
path, mime = cached
path, mime = result
media_paths.append(path)
mimes.append(mime)
# Batch summary: stable fields for offline aggregation.
logger.info(
"[%s] media resolve batch: scope=ybres concurrency=%d total=%d ok=%d failed=%d elapsed_ms=%d",
adapter.name,
adapter.media_resolve_concurrency,
len(active),
len(media_paths),
_failed,
_elapsed_ms,
)
return media_paths, mimes
@classmethod
@ -5071,6 +5180,19 @@ class YuanbaoAdapter(BasePlatformAdapter):
self._api_domain: str = (_extra.get("api_domain") or DEFAULT_API_DOMAIN).rstrip("/")
self._route_env: str = (_extra.get("route_env") or "").strip()
# Bounded concurrency for inbound media resolve/download.
# See _DEFAULT_RESOLVE_CONCURRENCY for rationale; clamped to [min, max]
# so a misconfigured value cannot hammer the resource backend nor
# accidentally drop below sequential behavior.
try:
_raw_concurrency = int(_extra.get("media_resolve_concurrency", _DEFAULT_RESOLVE_CONCURRENCY))
except (TypeError, ValueError):
_raw_concurrency = _DEFAULT_RESOLVE_CONCURRENCY
self.media_resolve_concurrency: int = max(
_MIN_RESOLVE_CONCURRENCY,
min(_MAX_RESOLVE_CONCURRENCY, _raw_concurrency),
)
# Core managers (UML composition)
self._connection: ConnectionManager = ConnectionManager(self)
self._outbound: OutboundManager = OutboundManager(self)

242
tests/test_yuanbao_pipeline.py
View file

@ -10,6 +10,7 @@ Tests cover:
6. OOP middleware ABC and class tests
"""
import asyncio
import sys
import os
import json
@ -45,6 +46,8 @@ from gateway.platforms.yuanbao import (
DispatchMiddleware,
InboundPipelineBuilder,
YuanbaoAdapter,
_MIN_RESOLVE_CONCURRENCY,
_MAX_RESOLVE_CONCURRENCY,
)
from gateway.config import PlatformConfig
@ -1600,6 +1603,245 @@ class TestResolveMediaUrlsCacheHit:
MediaResolveMiddleware._resource_cache.clear()
class TestResolveYbresRefsConcurrency:
"""Bounded-concurrency contracts for ``_resolve_ybres_refs``."""
# ------------------------------------------------------------------
# Bounded-concurrency contracts (issue 3 in
# yuanbao-media-pipeline-optimizations.md). These are behavior
# contracts, not implementation snapshots — they assert the
# invariants the new gather()-based path must hold, not how it's
# wired internally.
# ------------------------------------------------------------------
@pytest.mark.asyncio
async def test_concurrent_resolve_preserves_input_order(self):
"""Order of returned (paths, mimes) must match input ``refs`` order
even when later refs finish downloading first.
"""
adapter = make_adapter(extra={"media_resolve_concurrency": 4})
refs = [
("rid-A", "image", ""),
("rid-B", "image", ""),
("rid-C", "image", ""),
]
# _fetch is fast and uniform; the interesting variation is in
# _download_and_cache, where rid-A is the slowest. If results
# were assembled by completion order, rid-A would land last.
async def slow_fetch(_adapter, rid):
return f"https://fresh/{rid}"
delays = {"rid-A": 0.06, "rid-B": 0.02, "rid-C": 0.0}
results_by_rid = {
"rid-A": ("/cache/A.jpg", "image/jpeg"),
"rid-B": ("/cache/B.jpg", "image/jpeg"),
"rid-C": ("/cache/C.jpg", "image/jpeg"),
}
async def slow_download(_adapter, *, fetch_url, kind, file_name, log_tag, resource_id):
await asyncio.sleep(delays[resource_id])
return results_by_rid[resource_id]
with patch.object(
MediaResolveMiddleware, "_fetch_resource_url",
new=AsyncMock(side_effect=slow_fetch),
), patch.object(
MediaResolveMiddleware, "_download_and_cache",
new=AsyncMock(side_effect=slow_download),
):
paths, mimes = await MediaResolveMiddleware._resolve_ybres_refs(
adapter, refs, log_prefix="test",
)
assert paths == ["/cache/A.jpg", "/cache/B.jpg", "/cache/C.jpg"]
assert mimes == ["image/jpeg", "image/jpeg", "image/jpeg"]
@pytest.mark.asyncio
async def test_concurrency_one_equivalent_to_sequential(self):
"""``media_resolve_concurrency = 1`` must behave like the legacy
sequential path — at any moment at most one ``_download_and_cache``
is in flight.
"""
adapter = make_adapter(extra={"media_resolve_concurrency": 1})
refs = [("rid-A", "image", ""), ("rid-B", "image", ""), ("rid-C", "image", "")]
in_flight = 0
max_in_flight = 0
async def tracked_download(_adapter, *, fetch_url, kind, file_name, log_tag, resource_id):
nonlocal in_flight, max_in_flight
in_flight += 1
max_in_flight = max(max_in_flight, in_flight)
try:
# Yield to the event loop so any concurrent coroutine
# would have a chance to also enter the critical section.
await asyncio.sleep(0.01)
return (f"/cache/{resource_id}.jpg", "image/jpeg")
finally:
in_flight -= 1
with patch.object(
MediaResolveMiddleware, "_fetch_resource_url",
new=AsyncMock(side_effect=lambda _a, rid: f"https://fresh/{rid}"),
), patch.object(
MediaResolveMiddleware, "_download_and_cache",
new=AsyncMock(side_effect=tracked_download),
):
paths, _ = await MediaResolveMiddleware._resolve_ybres_refs(
adapter, refs, log_prefix="test",
)
assert max_in_flight == 1
assert paths == ["/cache/rid-A.jpg", "/cache/rid-B.jpg", "/cache/rid-C.jpg"]
@pytest.mark.asyncio
async def test_concurrency_caps_inflight_downloads(self):
"""Configured concurrency bounds the number of in-flight downloads."""
adapter = make_adapter(extra={"media_resolve_concurrency": 2})
refs = [(f"rid-{i}", "image", "") for i in range(6)]
in_flight = 0
max_in_flight = 0
async def tracked_download(_adapter, *, fetch_url, kind, file_name, log_tag, resource_id):
nonlocal in_flight, max_in_flight
in_flight += 1
max_in_flight = max(max_in_flight, in_flight)
try:
await asyncio.sleep(0.01)
return (f"/cache/{resource_id}.jpg", "image/jpeg")
finally:
in_flight -= 1
with patch.object(
MediaResolveMiddleware, "_fetch_resource_url",
new=AsyncMock(side_effect=lambda _a, rid: f"https://fresh/{rid}"),
), patch.object(
MediaResolveMiddleware, "_download_and_cache",
new=AsyncMock(side_effect=tracked_download),
):
paths, _ = await MediaResolveMiddleware._resolve_ybres_refs(
adapter, refs, log_prefix="test",
)
assert max_in_flight == 2
assert paths == [f"/cache/rid-{i}.jpg" for i in range(6)]
@pytest.mark.asyncio
async def test_download_exception_isolated_to_single_ref(self):
"""An exception raised inside ``_download_and_cache`` for one rid
must not poison the whole batch; surviving refs still resolve.
"""
adapter = make_adapter(extra={"media_resolve_concurrency": 4})
refs = [
("rid-ok-1", "image", ""),
("rid-boom", "image", ""),
("rid-ok-2", "image", ""),
]
async def maybe_boom(_adapter, *, fetch_url, kind, file_name, log_tag, resource_id):
if resource_id == "rid-boom":
raise RuntimeError("download crashed")
return (f"/cache/{resource_id}.jpg", "image/jpeg")
with patch.object(
MediaResolveMiddleware, "_fetch_resource_url",
new=AsyncMock(side_effect=lambda _a, rid: f"https://fresh/{rid}"),
), patch.object(
MediaResolveMiddleware, "_download_and_cache",
new=AsyncMock(side_effect=maybe_boom),
):
paths, mimes = await MediaResolveMiddleware._resolve_ybres_refs(
adapter, refs, log_prefix="test",
)
assert paths == ["/cache/rid-ok-1.jpg", "/cache/rid-ok-2.jpg"]
assert mimes == ["image/jpeg", "image/jpeg"]
@pytest.mark.asyncio
async def test_misconfigured_concurrency_clamped(self):
"""Out-of-range or non-int concurrency values are clamped, not crashed."""
# Negative -> clamped up to MIN
adapter_low = make_adapter(extra={"media_resolve_concurrency": -3})
assert adapter_low.media_resolve_concurrency >= _MIN_RESOLVE_CONCURRENCY
# Huge -> clamped down to MAX
adapter_high = make_adapter(extra={"media_resolve_concurrency": 9999})
assert adapter_high.media_resolve_concurrency <= _MAX_RESOLVE_CONCURRENCY
# Non-int garbage -> falls back to default, doesn't raise
adapter_garbage = make_adapter(extra={"media_resolve_concurrency": "fast"})
assert (
_MIN_RESOLVE_CONCURRENCY
<= adapter_garbage.media_resolve_concurrency
<= _MAX_RESOLVE_CONCURRENCY
)
class TestResolveMediaUrlsConcurrency:
"""Bounded-concurrency contracts for ``_resolve_media_urls`` (own-turn
media). Same invariants as ``_resolve_ybres_refs`` — order preserved,
failures isolated, concurrency clamped.
"""
@pytest.mark.asyncio
async def test_preserves_input_order(self):
adapter = make_adapter(extra={"media_resolve_concurrency": 4})
media_refs = [
{"kind": "image", "url": "https://hunyuan.tencent.com/api/resource/download?resourceId=rid-A", "name": ""},
{"kind": "image", "url": "https://hunyuan.tencent.com/api/resource/download?resourceId=rid-B", "name": ""},
{"kind": "image", "url": "https://hunyuan.tencent.com/api/resource/download?resourceId=rid-C", "name": ""},
]
delays = {"rid-A": 0.05, "rid-B": 0.02, "rid-C": 0.0}
async def slow_download(_adapter, *, fetch_url, kind, file_name, log_tag, resource_id):
await asyncio.sleep(delays[resource_id])
return (f"/cache/{resource_id}.jpg", "image/jpeg")
with patch.object(
MediaResolveMiddleware, "_resolve_download_url",
new=AsyncMock(side_effect=lambda _a, url: url),
), patch.object(
MediaResolveMiddleware, "_download_and_cache",
new=AsyncMock(side_effect=slow_download),
):
paths, mimes = await MediaResolveMiddleware._resolve_media_urls(
adapter, media_refs,
)
assert paths == ["/cache/rid-A.jpg", "/cache/rid-B.jpg", "/cache/rid-C.jpg"]
assert mimes == ["image/jpeg"] * 3
@pytest.mark.asyncio
async def test_failure_isolated(self):
adapter = make_adapter(extra={"media_resolve_concurrency": 4})
media_refs = [
{"kind": "image", "url": "https://x/api/resource/download?resourceId=ok", "name": ""},
{"kind": "image", "url": "https://x/api/resource/download?resourceId=boom", "name": ""},
]
async def maybe_boom(_adapter, *, fetch_url, kind, file_name, log_tag, resource_id):
if resource_id == "boom":
raise RuntimeError("download crashed")
return ("/cache/ok.jpg", "image/jpeg")
with patch.object(
MediaResolveMiddleware, "_resolve_download_url",
new=AsyncMock(side_effect=lambda _a, url: url),
), patch.object(
MediaResolveMiddleware, "_download_and_cache",
new=AsyncMock(side_effect=maybe_boom),
):
paths, mimes = await MediaResolveMiddleware._resolve_media_urls(
adapter, media_refs,
)
assert paths == ["/cache/ok.jpg"]
assert mimes == ["image/jpeg"]
class TestMediaResolveMiddlewareRouting:
"""Branch-routing tests for MediaResolveMiddleware.handle()."""