114 lines
3.7 KiB
Python
114 lines
3.7 KiB
Python
"""Weekly external research loop for Architect Governance Agent.
|
|
|
|
The agent plans and normalizes weekly research tasks. Actual internet access is
|
|
provided by the runtime/search worker; this module defines the audited workflow,
|
|
findings, impact analysis, and V/R/C recommendations.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ResearchTarget:
|
|
id: str
|
|
topic: str
|
|
source_type: str
|
|
query: str
|
|
cadence: str = "weekly"
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ResearchFinding:
|
|
id: str
|
|
target_id: str
|
|
title: str
|
|
summary: str
|
|
source: str
|
|
relevance: str
|
|
impact: str
|
|
recommendation: str
|
|
created_at: str = field(default_factory=utc_now)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ResearchDecision:
|
|
status: str
|
|
version_bump: str
|
|
reason: str
|
|
findings: list[ResearchFinding]
|
|
created_at: str = field(default_factory=utc_now)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"status": self.status,
|
|
"version_bump": self.version_bump,
|
|
"reason": self.reason,
|
|
"findings": [f.to_dict() for f in self.findings],
|
|
"created_at": self.created_at,
|
|
}
|
|
|
|
|
|
class ArchitectWeeklyResearchAgent:
|
|
DEFAULT_TARGETS = (
|
|
ResearchTarget("agent-frameworks", "Agent framework versions", "web", "latest multi-agent orchestration framework releases"),
|
|
ResearchTarget("model-ecosystem", "Model ecosystem changes", "web", "latest LLM model releases tool calling agent features"),
|
|
ResearchTarget("github-top-agentops", "GitHub top AgentOps implementations", "github", "top repositories agent observability multi agent orchestration"),
|
|
ResearchTarget("security-agent-risks", "Agent security changes", "web", "latest AI agent security sandbox prompt injection tool use risks"),
|
|
)
|
|
|
|
def plan(self) -> dict[str, Any]:
|
|
return {
|
|
"status": "planned",
|
|
"cadence": "weekly",
|
|
"targets": [target.to_dict() for target in self.DEFAULT_TARGETS],
|
|
"created_at": utc_now(),
|
|
}
|
|
|
|
def normalize_finding(
|
|
self,
|
|
target_id: str,
|
|
title: str,
|
|
summary: str,
|
|
source: str,
|
|
relevance: str = "medium",
|
|
impact: str = "none",
|
|
recommendation: str = "monitor",
|
|
) -> ResearchFinding:
|
|
return ResearchFinding(
|
|
id=f"finding_{uuid4().hex[:12]}",
|
|
target_id=target_id,
|
|
title=title,
|
|
summary=summary,
|
|
source=source,
|
|
relevance=relevance,
|
|
impact=impact,
|
|
recommendation=recommendation,
|
|
)
|
|
|
|
def decide(self, findings: list[ResearchFinding]) -> ResearchDecision:
|
|
if any(f.impact == "breaking" for f in findings):
|
|
return ResearchDecision("action_required", "V", "breaking external ecosystem change", findings)
|
|
if any(f.impact in {"architecture", "new_capability"} for f in findings):
|
|
return ResearchDecision("optimization_recommended", "R", "external capability may improve architecture", findings)
|
|
if any(f.impact in {"bugfix", "minor"} for f in findings):
|
|
return ResearchDecision("minor_change_recommended", "C", "minor compatible improvement", findings)
|
|
return ResearchDecision("no_action", "none", "no relevant external change", findings)
|
|
|
|
|
|
__all__ = ["ResearchTarget", "ResearchFinding", "ResearchDecision", "ArchitectWeeklyResearchAgent"]
|