Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| from urllib.parse import urlparse | |
| from typing import Any | |
| import requests | |
| from .config import resolve_agent_config | |
| DEFAULT_SERVICE_URL = "https://helmfridsson-qa-agent-service.hf.space/api/v1" | |
| OUTPUT_CONTRACTS = { | |
| "requirements_analyst": "requirements_analyst.v1", | |
| "test_designer": "test_designer.v1", | |
| "reviewer": "reviewer.v1", | |
| } | |
| class AgentServiceError(RuntimeError): | |
| pass | |
| def normalize_service_url(raw_url: Any) -> str: | |
| value = str(raw_url or DEFAULT_SERVICE_URL).strip() | |
| if not value: | |
| value = DEFAULT_SERVICE_URL | |
| if value.startswith("https://huggingface.co/spaces/"): | |
| parsed = urlparse(value) | |
| parts = [part for part in parsed.path.split("/") if part] | |
| if len(parts) >= 3: | |
| owner, space = parts[1], parts[2] | |
| return f"https://{owner.lower()}-{space.lower()}.hf.space/api/v1" | |
| parsed = urlparse(value) | |
| base = f"{parsed.scheme}://{parsed.netloc}" if parsed.scheme and parsed.netloc else "" | |
| path = parsed.path.rstrip("/") | |
| normalized_path = path.lower() | |
| if normalized_path.endswith("/docs"): | |
| path = path[: -len("/docs")] | |
| normalized_path = path.lower() | |
| elif normalized_path.endswith("/openapi.json"): | |
| path = path[: -len("/openapi.json")] | |
| normalized_path = path.lower() | |
| elif normalized_path.endswith("/health"): | |
| path = path[: -len("/health")] | |
| normalized_path = path.lower() | |
| elif normalized_path.endswith("/tools"): | |
| path = path[: -len("/tools")] | |
| normalized_path = path.lower() | |
| elif "/logs/recent" in normalized_path: | |
| path = path[: normalized_path.index("/logs/recent")] | |
| normalized_path = path.lower() | |
| elif "/agents/" in normalized_path: | |
| path = path[: normalized_path.index("/agents/")] | |
| normalized_path = path.lower() | |
| if normalized_path.endswith("/api/v1"): | |
| normalized_api_path = path | |
| elif path: | |
| normalized_api_path = f"{path}/api/v1" | |
| else: | |
| normalized_api_path = "/api/v1" | |
| if base: | |
| return f"{base}{normalized_api_path}".rstrip("/") | |
| return value.rstrip("/") | |
| def service_base_url(global_config: dict[str, Any]) -> str: | |
| return normalize_service_url(global_config.get("qa_agent_service_url") or DEFAULT_SERVICE_URL) | |
| def validate_agent_service(global_config: dict[str, Any]) -> None: | |
| base_url = service_base_url(global_config) | |
| try: | |
| response = requests.get(f"{base_url}/health", timeout=5) | |
| response.raise_for_status() | |
| except requests.RequestException as exc: | |
| status_code = getattr(getattr(exc, "response", None), "status_code", None) | |
| if status_code == 404: | |
| raise ConnectionError( | |
| f"{base_url} is reachable, but it does not expose the REST /health endpoint. " | |
| "If this is the MCP server, use the FastAPI REST server instead." | |
| ) from exc | |
| raise ConnectionError(f"Could not connect to QA Agent Service at {base_url}.") from exc | |
| try: | |
| data = response.json() | |
| except ValueError as exc: | |
| content_type = response.headers.get("content-type", "") | |
| raise ConnectionError( | |
| f"{base_url}/health responded, but not with REST JSON. " | |
| f"content-type={content_type!r}. The configured endpoint may be serving the Streamlit monitor instead of FastAPI." | |
| ) from exc | |
| if data.get("status") != "ok": | |
| raise ConnectionError(f"QA Agent Service at {base_url} returned unexpected health response: {data}") | |
| def _shared_memory_payload(state: dict[str, Any]) -> dict[str, Any]: | |
| memory = state.get("working_memory") or {} | |
| shared = memory.get("shared") or {} | |
| return { | |
| "facts": shared.get("facts", shared if isinstance(shared, dict) else {}), | |
| "assumptions": shared.get("assumptions", []), | |
| "decisions": shared.get("decisions", []), | |
| "timeline": memory.get("timeline", []), | |
| } | |
| def _agent_private_memory_payload(state: dict[str, Any], agent_key: str) -> dict[str, Any]: | |
| memory = state.get("working_memory") or {} | |
| private = memory.get("agent_private") or {} | |
| return {"data": private.get(agent_key, {})} | |
| def merge_memory_updates(state: dict[str, Any], agent_key: str, updates: dict[str, Any]) -> dict[str, Any]: | |
| memory = dict(state.get("working_memory") or {"shared": {}, "agent_private": {}, "timeline": []}) | |
| shared = dict(memory.get("shared") or {}) | |
| facts = dict(shared.get("facts", {})) | |
| facts.update(updates.get("facts") or {}) | |
| shared["facts"] = facts | |
| shared["assumptions"] = list(shared.get("assumptions", [])) + list(updates.get("assumptions") or []) | |
| shared["decisions"] = list(shared.get("decisions", [])) + list(updates.get("decisions") or []) | |
| private = dict(memory.get("agent_private") or {}) | |
| private[agent_key] = {**dict(private.get(agent_key, {})), **dict(updates.get("agent_private") or {})} | |
| memory["shared"] = shared | |
| memory["agent_private"] = private | |
| memory["timeline"] = list(memory.get("timeline", [])) + list(updates.get("timeline") or []) | |
| return memory | |
| def invoke_service_agent( | |
| state: dict[str, Any], | |
| agent_key: str, | |
| agent_input: dict[str, Any], | |
| ) -> dict[str, Any]: | |
| global_config = state.get("global_llm_config", {}) | |
| agent_config = resolve_agent_config(global_config, state.get("agent_configs", []), agent_key) | |
| timeout = int(agent_config.get("timeout_seconds") or global_config.get("service_timeout_seconds") or 150) | |
| payload = { | |
| "run_id": state.get("run_metadata", {}).get("run_id", "run_unknown"), | |
| "iteration": int(state.get("iterations", 1) or 1), | |
| "stage_index": len(state.get("stage_traces", [])) + 1, | |
| "scenario": { | |
| "title": state.get("title", ""), | |
| "source_requirements": state.get("source_requirements", ""), | |
| }, | |
| "agent_input": agent_input, | |
| "shared_memory": _shared_memory_payload(state), | |
| "agent_private_memory": _agent_private_memory_payload(state, agent_key), | |
| "instructions": {"additional_directives": agent_config.get("directives", "")}, | |
| "output_contract": OUTPUT_CONTRACTS[agent_key], | |
| } | |
| model_id = agent_config.get("model_id") | |
| provider = agent_config.get("provider") | |
| if model_id and provider: | |
| payload["model_config"] = { | |
| "provider": provider, | |
| "model_id": model_id, | |
| "temperature": float(agent_config.get("temperature", 0.2)), | |
| "timeout_seconds": timeout, | |
| } | |
| base_url = service_base_url(global_config) | |
| try: | |
| response = requests.post(f"{base_url}/agents/{agent_key}/invoke", json=payload, timeout=timeout) | |
| response.raise_for_status() | |
| except requests.RequestException as exc: | |
| raise AgentServiceError(f"QA Agent Service call failed for {agent_key}: {exc}") from exc | |
| data = response.json() | |
| if data.get("status") != "completed": | |
| messages = [error.get("message", "") for error in data.get("errors", []) if isinstance(error, dict)] | |
| detail = "; ".join(message for message in messages if message) or f"status={data.get('status')}" | |
| raise AgentServiceError(f"QA Agent Service returned failed response for {agent_key}: {detail}") | |
| return data | |