|
| 1 | +"""Static agent backend for deterministic testing without LLM calls. |
| 2 | +
|
| 3 | +Provides StaticAgent and StaticAgentRegistry that return canned responses |
| 4 | +from YAML rules. Swap at the DI container level — no mocks, no pydantic-ai. |
| 5 | +""" |
| 6 | + |
| 7 | +import re |
| 8 | +from typing import ( |
| 9 | + Any, |
| 10 | + Optional, |
| 11 | +) |
| 12 | + |
| 13 | +import yaml |
| 14 | + |
| 15 | +from .base import ( |
| 16 | + ActionSuggestion, |
| 17 | + AgentResponse, |
| 18 | + BaseGalaxyAgent, |
| 19 | + GalaxyAgentDependencies, |
| 20 | +) |
| 21 | +from .registry import AgentRegistry |
| 22 | + |
| 23 | + |
| 24 | +class StaticAgent(BaseGalaxyAgent): |
| 25 | + """Agent that returns canned responses from YAML rules. |
| 26 | +
|
| 27 | + Subclasses BaseGalaxyAgent but skips pydantic-ai Agent creation entirely. |
| 28 | + Only process() is meaningful — all other BaseGalaxyAgent methods are stubs. |
| 29 | + """ |
| 30 | + |
| 31 | + agent_type = "static" # overridden per-instance |
| 32 | + |
| 33 | + def __init__( |
| 34 | + self, |
| 35 | + agent_type_str: str, |
| 36 | + rules: list[dict[str, Any]], |
| 37 | + fallback: dict[str, Any], |
| 38 | + defaults: dict[str, Any], |
| 39 | + ): |
| 40 | + # Intentionally skip super().__init__() — no pydantic-ai Agent needed. |
| 41 | + self.agent_type = agent_type_str |
| 42 | + self._rules = rules |
| 43 | + self._fallback = fallback |
| 44 | + self._defaults = defaults |
| 45 | + |
| 46 | + def _create_agent(self): |
| 47 | + raise NotImplementedError("StaticAgent does not use pydantic-ai") |
| 48 | + |
| 49 | + def get_system_prompt(self) -> str: |
| 50 | + return "" |
| 51 | + |
| 52 | + async def process(self, query: str, context: Optional[dict[str, Any]] = None) -> AgentResponse: |
| 53 | + for rule in self._rules: |
| 54 | + if self._rule_matches(rule.get("match", {}), query, context): |
| 55 | + return self._make_response(rule["response"]) |
| 56 | + return self._make_response(self._fallback) |
| 57 | + |
| 58 | + def _rule_matches(self, match: dict[str, Any], query: str, context: Optional[dict[str, Any]]) -> bool: |
| 59 | + if "agent_type" in match and match["agent_type"] != self.agent_type: |
| 60 | + return False |
| 61 | + if "query" in match and not re.search(match["query"], query): |
| 62 | + return False |
| 63 | + if "context" in match: |
| 64 | + if not context: |
| 65 | + return False |
| 66 | + for field, pattern in match["context"].items(): |
| 67 | + if field not in context or not re.search(pattern, str(context[field])): |
| 68 | + return False |
| 69 | + return True |
| 70 | + |
| 71 | + def _make_response(self, resp: dict[str, Any]) -> AgentResponse: |
| 72 | + raw_suggestions = resp.get("suggestions", []) |
| 73 | + suggestions = [ActionSuggestion(**s) for s in raw_suggestions] |
| 74 | + return AgentResponse( |
| 75 | + content=resp.get("content", self._fallback.get("content", "")), |
| 76 | + confidence=resp.get("confidence", self._defaults.get("confidence", "medium")), |
| 77 | + agent_type=resp.get("agent_type", self.agent_type), |
| 78 | + suggestions=suggestions, |
| 79 | + metadata={**resp.get("metadata", {}), "static_backend": True}, |
| 80 | + reasoning=resp.get("reasoning"), |
| 81 | + ) |
| 82 | + |
| 83 | + |
| 84 | +class StaticAgentRegistry(AgentRegistry): |
| 85 | + """Registry that returns StaticAgent instances from YAML config. |
| 86 | +
|
| 87 | + Subclasses AgentRegistry so it's type-compatible with the DI container. |
| 88 | + """ |
| 89 | + |
| 90 | + def __init__(self, config_path: str): |
| 91 | + super().__init__() |
| 92 | + with open(config_path) as f: |
| 93 | + self._config: dict[str, Any] = yaml.safe_load(f) or {} |
| 94 | + self._rules: list[dict[str, Any]] = self._config.get("rules", []) |
| 95 | + self._fallback: dict[str, Any] = self._config.get("fallback", {}) |
| 96 | + self._defaults: dict[str, Any] = self._config.get("defaults", {}) |
| 97 | + |
| 98 | + # Collect known agent_types from rules |
| 99 | + self._known_types: set[str] = set() |
| 100 | + for rule in self._rules: |
| 101 | + match = rule.get("match", {}) |
| 102 | + if "agent_type" in match: |
| 103 | + self._known_types.add(match["agent_type"]) |
| 104 | + |
| 105 | + def get_agent(self, agent_type: str, deps: GalaxyAgentDependencies) -> StaticAgent: |
| 106 | + """Return a StaticAgent that matches rules for this agent_type.""" |
| 107 | + applicable = [r for r in self._rules if r.get("match", {}).get("agent_type", agent_type) == agent_type] |
| 108 | + return StaticAgent(agent_type, applicable, self._fallback, self._defaults) |
| 109 | + |
| 110 | + def is_registered(self, agent_type: str) -> bool: |
| 111 | + return agent_type in self._known_types or bool(self._fallback) |
| 112 | + |
| 113 | + def list_agents(self) -> list[str]: |
| 114 | + return sorted(self._known_types) |
| 115 | + |
| 116 | + def get_agent_info(self, agent_type: str) -> dict[str, Any]: |
| 117 | + return { |
| 118 | + "agent_type": agent_type, |
| 119 | + "class_name": "StaticAgent", |
| 120 | + "module": "galaxy.agents.static_backend", |
| 121 | + "metadata": {"static_backend": True}, |
| 122 | + "description": "Static test agent", |
| 123 | + } |
| 124 | + |
| 125 | + def list_agent_info(self) -> list[dict[str, Any]]: |
| 126 | + return [self.get_agent_info(t) for t in sorted(self._known_types)] |
0 commit comments