hermes: bound local routing context
All checks were successful
Tests / Declarative: Post Actions passed: 255
All checks were successful
Tests / Declarative: Post Actions passed: 255
This commit is contained in:
parent
0307323dc6
commit
cd06751b66
@ -60,6 +60,7 @@ configMapGenerator:
|
||||
- codex=scripts/codex
|
||||
- configure_agent_clients.py=scripts/configure_agent_clients.py
|
||||
- codex_broker.py=scripts/codex_broker.py
|
||||
- classifier_broker.py=scripts/classifier_broker.py
|
||||
- claude_oauth_broker.py=scripts/claude_oauth_broker.py
|
||||
- worker_route_broker.py=scripts/worker_route_broker.py
|
||||
- gitea_askpass.sh=scripts/gitea_askpass.sh
|
||||
|
||||
240
services/hermes/scripts/classifier_broker.py
Normal file
240
services/hermes/scripts/classifier_broker.py
Normal file
@ -0,0 +1,240 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Bound Switchyard classifier context before forwarding it to local Ollama."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from typing import Any, Final
|
||||
|
||||
import httpx
|
||||
|
||||
|
||||
HOST: Final = os.environ.get("HERMES_CLASSIFIER_BROKER_HOST", "0.0.0.0")
|
||||
PORT: Final = int(os.environ.get("HERMES_CLASSIFIER_BROKER_PORT", "9008"))
|
||||
UPSTREAM: Final = os.environ.get(
|
||||
"HERMES_CLASSIFIER_BROKER_UPSTREAM",
|
||||
"http://ollama.ai.svc.cluster.local:11434",
|
||||
).rstrip("/")
|
||||
MAX_BODY_BYTES: Final = int(
|
||||
os.environ.get("HERMES_CLASSIFIER_BROKER_MAX_BODY", str(32 << 20))
|
||||
)
|
||||
MAX_SYSTEM_CHARS: Final = int(
|
||||
os.environ.get("HERMES_CLASSIFIER_MAX_SYSTEM_CHARS", "6500")
|
||||
)
|
||||
MAX_CONTEXT_CHARS: Final = int(
|
||||
os.environ.get("HERMES_CLASSIFIER_MAX_CONTEXT_CHARS", "7000")
|
||||
)
|
||||
READ_TIMEOUT_SECONDS: Final = float(
|
||||
os.environ.get("HERMES_CLASSIFIER_BROKER_READ_TIMEOUT", "60")
|
||||
)
|
||||
ALLOWED_PATHS: Final = {"/v1/chat/completions", "/v1/models"}
|
||||
|
||||
|
||||
def _bounded_text(value: str, limit: int) -> str:
|
||||
"""Keep both ends of text because intent and current status often sit apart."""
|
||||
if len(value) <= limit:
|
||||
return value
|
||||
if limit < 80:
|
||||
return value[:limit]
|
||||
marker = "\n...[classifier context compacted]...\n"
|
||||
remaining = limit - len(marker)
|
||||
head = remaining // 2
|
||||
return f"{value[:head]}{marker}{value[-(remaining - head):]}"
|
||||
|
||||
|
||||
def _compact_content(content: Any, limit: int) -> Any:
|
||||
"""Remove binary/multimodal payloads and bound text sent to the judge."""
|
||||
if isinstance(content, str):
|
||||
return _bounded_text(content, limit)
|
||||
if not isinstance(content, list):
|
||||
return content
|
||||
|
||||
compacted: list[Any] = []
|
||||
remaining = limit
|
||||
for block in content:
|
||||
if not isinstance(block, dict):
|
||||
continue
|
||||
kind = str(block.get("type") or "")
|
||||
if kind in {"text", "input_text", "output_text"}:
|
||||
key = "text"
|
||||
text = str(block.get(key) or "")
|
||||
if not text or remaining <= 0:
|
||||
continue
|
||||
text = _bounded_text(text, remaining)
|
||||
compacted.append({**block, key: text})
|
||||
remaining -= len(text)
|
||||
elif kind in {"image", "image_url", "input_image"}:
|
||||
marker = "[image attachment available to the selected worker]"
|
||||
if remaining >= len(marker):
|
||||
compacted.append({"type": "text", "text": marker})
|
||||
remaining -= len(marker)
|
||||
return compacted
|
||||
|
||||
|
||||
def _compact_message(message: dict[str, Any], limit: int) -> dict[str, Any]:
|
||||
"""Copy routing-relevant message metadata while bounding large values."""
|
||||
result = copy.deepcopy(message)
|
||||
if "content" in result:
|
||||
result["content"] = _compact_content(result["content"], limit)
|
||||
tool_calls = result.get("tool_calls")
|
||||
if isinstance(tool_calls, list):
|
||||
kept: list[dict[str, Any]] = []
|
||||
for call in tool_calls[-4:]:
|
||||
if not isinstance(call, dict):
|
||||
continue
|
||||
item = copy.deepcopy(call)
|
||||
function = item.get("function")
|
||||
if isinstance(function, dict) and isinstance(function.get("arguments"), str):
|
||||
function["arguments"] = _bounded_text(function["arguments"], 600)
|
||||
kept.append(item)
|
||||
result["tool_calls"] = kept
|
||||
return result
|
||||
|
||||
|
||||
def _compact_messages(messages: list[Any]) -> list[Any]:
|
||||
"""Keep the routing contract, opening task, and latest decision context."""
|
||||
valid = [message for message in messages if isinstance(message, dict)]
|
||||
system_indices = [
|
||||
index
|
||||
for index, message in enumerate(valid)
|
||||
if message.get("role") in {"system", "developer"}
|
||||
]
|
||||
non_system = [index for index in range(len(valid)) if index not in system_indices]
|
||||
user_indices = [index for index in non_system if valid[index].get("role") == "user"]
|
||||
|
||||
opening_user = user_indices[0] if user_indices else None
|
||||
latest_user = user_indices[-1] if user_indices else None
|
||||
selected = set(system_indices)
|
||||
selected.update(non_system[-4:])
|
||||
if opening_user is not None:
|
||||
selected.add(opening_user)
|
||||
if latest_user is not None:
|
||||
selected.add(latest_user)
|
||||
|
||||
other_indices = [
|
||||
index
|
||||
for index in selected
|
||||
if index not in system_indices and index not in {opening_user, latest_user}
|
||||
]
|
||||
latest_budget = min(3500, MAX_CONTEXT_CHARS)
|
||||
opening_budget = min(1200, max(0, MAX_CONTEXT_CHARS - latest_budget))
|
||||
other_budget = max(0, MAX_CONTEXT_CHARS - latest_budget - opening_budget)
|
||||
other_limit = min(1200, other_budget // max(1, len(other_indices)))
|
||||
system_limit = MAX_SYSTEM_CHARS // max(1, len(system_indices))
|
||||
|
||||
result: list[Any] = []
|
||||
for index, message in enumerate(valid):
|
||||
if index not in selected:
|
||||
continue
|
||||
role = str(message.get("role") or "")
|
||||
if role in {"system", "developer"}:
|
||||
compacted = _compact_message(message, system_limit)
|
||||
elif index == latest_user:
|
||||
compacted = _compact_message(message, latest_budget)
|
||||
elif index == opening_user:
|
||||
compacted = _compact_message(message, opening_budget)
|
||||
else:
|
||||
if other_limit <= 0:
|
||||
continue
|
||||
compacted = _compact_message(message, other_limit)
|
||||
result.append(compacted)
|
||||
return result
|
||||
|
||||
|
||||
def compact_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Return the same OpenAI request with only classifier input compacted."""
|
||||
result = copy.deepcopy(payload)
|
||||
messages = result.get("messages")
|
||||
if isinstance(messages, list):
|
||||
result["messages"] = _compact_messages(messages)
|
||||
# The judge never needs tools or binary inputs. Switchyard supplies a
|
||||
# response schema separately, and that contract must remain untouched.
|
||||
result.pop("tools", None)
|
||||
return result
|
||||
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
"""Proxy only Switchyard's local judge calls with strict input bounds."""
|
||||
|
||||
server_version = "HermesClassifierBroker/1"
|
||||
|
||||
def log_message(self, format: str, *args: object) -> None:
|
||||
return
|
||||
|
||||
def _json(self, status: int, value: dict[str, object]) -> None:
|
||||
body = json.dumps(value, separators=(",", ":")).encode("utf-8")
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.send_header("Cache-Control", "no-store")
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def do_GET(self) -> None: # noqa: N802
|
||||
if self.path == "/health":
|
||||
self._json(200, {"ok": True, "upstream": "ollama"})
|
||||
return
|
||||
if self.path not in ALLOWED_PATHS:
|
||||
self._json(404, {"error": "not found"})
|
||||
return
|
||||
self._proxy(None)
|
||||
|
||||
def do_POST(self) -> None: # noqa: N802
|
||||
if self.path not in ALLOWED_PATHS:
|
||||
self._json(404, {"error": "not found"})
|
||||
return
|
||||
try:
|
||||
length = int(self.headers.get("Content-Length", "0"))
|
||||
except ValueError:
|
||||
length = -1
|
||||
if length <= 0 or length > MAX_BODY_BYTES:
|
||||
self._json(413, {"error": "request too large"})
|
||||
return
|
||||
try:
|
||||
payload = json.loads(self.rfile.read(length))
|
||||
if not isinstance(payload, dict):
|
||||
raise ValueError("request must be a JSON object")
|
||||
body = json.dumps(
|
||||
compact_payload(payload), separators=(",", ":")
|
||||
).encode("utf-8")
|
||||
except (ValueError, TypeError, json.JSONDecodeError) as exc:
|
||||
self._json(400, {"error": str(exc)})
|
||||
return
|
||||
print(
|
||||
f"classifier-broker request_bytes={length} compacted_bytes={len(body)}",
|
||||
flush=True,
|
||||
)
|
||||
self._proxy(body)
|
||||
|
||||
def _proxy(self, body: bytes | None) -> None:
|
||||
try:
|
||||
timeout = httpx.Timeout(10.0, read=READ_TIMEOUT_SECONDS)
|
||||
with httpx.Client(timeout=timeout) as client:
|
||||
response = client.request(
|
||||
self.command,
|
||||
f"{UPSTREAM}{self.path}",
|
||||
headers={"Content-Type": "application/json"},
|
||||
content=body,
|
||||
)
|
||||
self.send_response(response.status_code)
|
||||
self.send_header(
|
||||
"Content-Type", response.headers.get("Content-Type", "application/json")
|
||||
)
|
||||
self.send_header("Content-Length", str(len(response.content)))
|
||||
self.send_header("Cache-Control", "no-store")
|
||||
self.end_headers()
|
||||
self.wfile.write(response.content)
|
||||
except (httpx.HTTPError, OSError) as exc:
|
||||
self._json(503, {"error": f"classifier unavailable: {exc}"})
|
||||
|
||||
|
||||
def main() -> None:
|
||||
server = ThreadingHTTPServer((HOST, PORT), Handler)
|
||||
server.serve_forever()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@ -12,8 +12,11 @@ data:
|
||||
|
||||
[llm_clients.classifier]
|
||||
format = "openai_chat"
|
||||
base_url = "http://ollama.ai.svc.cluster.local:11434/v1"
|
||||
max_retries = 1
|
||||
base_url = "http://127.0.0.1:9008/v1"
|
||||
# Classification is advisory and fail-open. Do not make an interactive
|
||||
# request wait through a second slow local inference before using the
|
||||
# route's conservative hosted default.
|
||||
max_retries = 0
|
||||
|
||||
[llm_clients.local_low]
|
||||
format = "openai_chat"
|
||||
@ -191,7 +194,7 @@ data:
|
||||
targets = ["codex_sol_xhigh", "claude_opus_xhigh", "codex_sol_high", "claude_sonnet_high", "claude_opus_high", "codex_terra_high", "codex_sol_medium", "claude_sonnet_medium", "codex_terra_medium", "local_qwen_medium", "codex_terra_low", "codex_luna_low", "claude_haiku_low", "local_qwen_low"]
|
||||
default_target = "codex_terra_medium"
|
||||
session_affinity = false
|
||||
recent_turn_window = 8
|
||||
recent_turn_window = 4
|
||||
context_window = 272000
|
||||
tool_calling = true
|
||||
reasoning = true
|
||||
@ -261,7 +264,7 @@ data:
|
||||
targets = ["codex_sol_xhigh", "claude_opus_xhigh", "codex_sol_high", "claude_sonnet_high", "claude_opus_high", "codex_terra_high", "codex_sol_medium", "claude_sonnet_medium", "codex_terra_medium", "local_qwen_medium", "codex_terra_low", "codex_luna_low", "claude_haiku_low", "local_qwen_low"]
|
||||
default_target = "codex_terra_medium"
|
||||
session_affinity = false
|
||||
recent_turn_window = 8
|
||||
recent_turn_window = 4
|
||||
context_window = 272000
|
||||
tool_calling = true
|
||||
reasoning = true
|
||||
@ -332,7 +335,7 @@ data:
|
||||
targets = ["claude_opus_xhigh", "codex_sol_xhigh", "claude_sonnet_high", "codex_sol_high", "claude_opus_high", "codex_terra_high", "claude_sonnet_medium", "codex_sol_medium", "codex_terra_medium"]
|
||||
default_target = "claude_sonnet_high"
|
||||
session_affinity = false
|
||||
recent_turn_window = 12
|
||||
recent_turn_window = 6
|
||||
context_window = 272000
|
||||
tool_calling = true
|
||||
reasoning = true
|
||||
@ -396,7 +399,7 @@ data:
|
||||
targets = ["codex_sol_xhigh", "claude_opus_xhigh", "codex_sol_high", "claude_opus_high", "claude_sonnet_high", "codex_terra_high", "codex_sol_medium", "claude_sonnet_medium", "codex_terra_medium"]
|
||||
default_target = "codex_sol_high"
|
||||
session_affinity = false
|
||||
recent_turn_window = 16
|
||||
recent_turn_window = 6
|
||||
context_window = 272000
|
||||
tool_calling = true
|
||||
reasoning = true
|
||||
@ -458,7 +461,7 @@ data:
|
||||
targets = ["worker_codex_sol_high", "worker_claude_sonnet_high", "worker_codex_sol_xhigh", "worker_claude_opus_xhigh", "worker_codex_terra_medium", "worker_claude_sonnet_medium", "worker_codex_luna_low", "worker_claude_haiku_low"]
|
||||
default_target = "worker_codex_sol_high"
|
||||
session_affinity = false
|
||||
recent_turn_window = 16
|
||||
recent_turn_window = 6
|
||||
context_window = 272000
|
||||
tool_calling = false
|
||||
reasoning = true
|
||||
|
||||
@ -22,7 +22,7 @@ spec:
|
||||
labels:
|
||||
app: hermes-switchyard
|
||||
annotations:
|
||||
ai.bstein.dev/config-rev: "20260812-image-continuity"
|
||||
ai.bstein.dev/config-rev: "20260812-classifier-context"
|
||||
prometheus.io/scrape: "true"
|
||||
prometheus.io/port: "9005"
|
||||
prometheus.io/path: /metrics
|
||||
@ -234,6 +234,56 @@ spec:
|
||||
- name: routing-catalog
|
||||
mountPath: /routing-catalog
|
||||
readOnly: true
|
||||
- name: classifier-broker
|
||||
image: registry.bstein.dev/bstein/hermes-switchyard-brokers@sha256:ee7e95e060ef8083da505162d7e9030daba15fdd828cc047bbcbe6aa409d2083
|
||||
imagePullPolicy: IfNotPresent
|
||||
command:
|
||||
- python
|
||||
- /opt/coordinator/classifier_broker.py
|
||||
ports:
|
||||
- name: classifier
|
||||
containerPort: 9008
|
||||
protocol: TCP
|
||||
env:
|
||||
- name: HERMES_CLASSIFIER_BROKER_UPSTREAM
|
||||
value: http://ollama.ai.svc.cluster.local:11434
|
||||
- name: HERMES_CLASSIFIER_BROKER_READ_TIMEOUT
|
||||
value: "60"
|
||||
readinessProbe:
|
||||
httpGet:
|
||||
path: /health
|
||||
port: classifier
|
||||
initialDelaySeconds: 2
|
||||
periodSeconds: 10
|
||||
timeoutSeconds: 3
|
||||
livenessProbe:
|
||||
httpGet:
|
||||
path: /health
|
||||
port: classifier
|
||||
initialDelaySeconds: 10
|
||||
periodSeconds: 30
|
||||
timeoutSeconds: 5
|
||||
securityContext:
|
||||
allowPrivilegeEscalation: false
|
||||
capabilities:
|
||||
drop: [ALL]
|
||||
readOnlyRootFilesystem: true
|
||||
runAsNonRoot: true
|
||||
runAsUser: 10000
|
||||
runAsGroup: 10000
|
||||
resources:
|
||||
requests:
|
||||
cpu: 25m
|
||||
memory: 48Mi
|
||||
limits:
|
||||
cpu: 250m
|
||||
memory: 192Mi
|
||||
volumeMounts:
|
||||
- name: coordinator
|
||||
mountPath: /opt/coordinator
|
||||
readOnly: true
|
||||
- name: tmp
|
||||
mountPath: /tmp
|
||||
volumes:
|
||||
- name: config
|
||||
configMap:
|
||||
|
||||
@ -869,6 +869,82 @@ def test_switchyard_brokers_use_the_small_dedicated_image():
|
||||
)
|
||||
assert containers["claude-oauth-broker"]["image"] == expected
|
||||
assert containers["worker-route-broker"]["image"] == expected
|
||||
assert containers["classifier-broker"]["image"] == expected
|
||||
|
||||
|
||||
def test_classifier_broker_bounds_history_without_losing_routing_intent(monkeypatch):
|
||||
"""AUTO classification must fit the Jetson context without losing intent."""
|
||||
broker_path = HERMES / "scripts" / "classifier_broker.py"
|
||||
monkeypatch.setitem(sys.modules, "httpx", SimpleNamespace())
|
||||
spec = importlib.util.spec_from_file_location("hermes_classifier_broker", broker_path)
|
||||
assert spec and spec.loader
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
|
||||
payload = {
|
||||
"model": "qwen2.5:14b-instruct-q4_0",
|
||||
"messages": [
|
||||
{
|
||||
"role": "system",
|
||||
"content": "routing contract\n" + ("candidate policy " * 1000),
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"content": "Build and verify the Cassandra release safely.",
|
||||
},
|
||||
{"role": "assistant", "content": "working"},
|
||||
{
|
||||
"role": "tool",
|
||||
"content": "unbounded test output " * 10000,
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "image_url", "image_url": {"url": "data:image/png;base64,private"}},
|
||||
{"type": "text", "text": "Turn this cat into a cute clown."},
|
||||
],
|
||||
},
|
||||
],
|
||||
"tools": [{"type": "function", "function": {"name": "large_tool"}}],
|
||||
"response_format": {"type": "json_object"},
|
||||
}
|
||||
|
||||
compacted = module.compact_payload(payload)
|
||||
encoded = json.dumps(compacted)
|
||||
|
||||
assert compacted["model"] == payload["model"]
|
||||
assert compacted["response_format"] == payload["response_format"]
|
||||
assert "tools" not in compacted
|
||||
assert "Build and verify the Cassandra release safely." in encoded
|
||||
assert "Turn this cat into a cute clown." in encoded
|
||||
assert "image attachment available to the selected worker" in encoded
|
||||
assert "data:image/png;base64" not in encoded
|
||||
assert "unbounded test output " * 100 not in encoded
|
||||
assert len(encoded) < 18_000
|
||||
|
||||
|
||||
def test_switchyard_classifier_is_bounded_and_fails_open_once():
|
||||
"""A sick local judge must not hold chat through repeated long retries."""
|
||||
config = tomllib.loads(
|
||||
_documents(HERMES / "switchyard-configmap.yaml")[0]["data"]["routes.toml"]
|
||||
)
|
||||
classifier = config["llm_clients"]["classifier"]
|
||||
assert classifier["base_url"] == "http://127.0.0.1:9008/v1"
|
||||
assert classifier["max_retries"] == 0
|
||||
for route in ("auto_fast", "auto_balanced"):
|
||||
assert config["routes"][route]["recent_turn_window"] == 4
|
||||
for route in ("auto_deep", "auto_maximum", "worker_auto_maximum"):
|
||||
assert config["routes"][route]["recent_turn_window"] == 6
|
||||
|
||||
deployment = _documents(HERMES / "switchyard-deployment.yaml")[0]
|
||||
containers = {
|
||||
item["name"]: item
|
||||
for item in deployment["spec"]["template"]["spec"]["containers"]
|
||||
}
|
||||
classifier_container = containers["classifier-broker"]
|
||||
env = {item["name"]: item["value"] for item in classifier_container["env"]}
|
||||
assert env["HERMES_CLASSIFIER_BROKER_READ_TIMEOUT"] == "60"
|
||||
assert classifier_container["readinessProbe"]["httpGet"]["port"] == "classifier"
|
||||
|
||||
|
||||
def test_image_broker_returns_bytes_and_removes_owner_cache(tmp_path: Path, monkeypatch):
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user