Coverage for agentos/evolution/engine.py: 37%
116 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-09 07:12 +0800
1"""
2Evolution Engine for NexusAgent.
4Approval-based self-evolution system. Agents can propose
5improvements, but changes require human approval before
6being applied.
7"""
9from __future__ import annotations
11import os
12import time
13import uuid
14from collections.abc import Callable
15from dataclasses import dataclass, field
16from enum import StrEnum
17from typing import Any
20class EvolutionStatus(StrEnum):
21 """Status of an evolution proposal."""
23 PENDING = "pending" # Waiting for approval
24 APPROVED = "approved" # Approved, ready to apply
25 REJECTED = "rejected" # Rejected by human
26 APPLIED = "applied" # Successfully applied
27 FAILED = "failed" # Failed to apply
30@dataclass
31class EvolutionProposal:
32 """
33 A proposed evolution/improvement.
35 Attributes:
36 id: Unique identifier
37 agent_name: Name of agent to evolve
38 change_type: Type of change (prompt/tools/params)
39 description: Human-readable description
40 old_value: Current value
41 new_value: Proposed new value
42 status: Approval status
43 created_at: Creation timestamp
44 approved_at: Approval timestamp
45 approved_by: Who approved
46 applied_at: Application timestamp
47 metadata: Additional metadata
48 """
50 id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
51 agent_name: str = ""
52 change_type: str = "prompt" # prompt, tools, params, behavior
53 description: str = ""
54 old_value: Any = None
55 new_value: Any = None
56 confidence: float = 0.0
57 risk_level: str = "low"
58 target_files: list[str] = field(default_factory=list)
59 status: EvolutionStatus = EvolutionStatus.PENDING
60 created_at: float = field(default_factory=time.time)
61 approved_at: float | None = None
62 approved_by: str | None = None
63 applied_at: float | None = None
64 metadata: dict[str, Any] = field(default_factory=dict)
66 def to_dict(self) -> dict[str, Any]:
67 """Convert to dict."""
68 return {
69 "id": self.id,
70 "agent_name": self.agent_name,
71 "change_type": self.change_type,
72 "description": self.description,
73 "old_value": self.old_value,
74 "new_value": self.new_value,
75 "status": self.status.value,
76 "created_at": self.created_at,
77 "approved_at": self.approved_at,
78 "approved_by": self.approved_by,
79 "applied_at": self.applied_at,
80 "metadata": self.metadata,
81 }
84class EvolutionEngine:
85 """
86 Approval-based self-evolution engine.
88 Manages the lifecycle of evolution proposals:
89 1. Agent proposes improvement
90 2. Human reviews and approves/rejects
91 3. Approved changes are applied
93 Usage:
94 engine = EvolutionEngine()
96 # Agent proposes improvement
97 proposal = engine.propose(
98 agent_name="SupportAgent",
99 change_type="prompt",
100 description="Improve greeting",
101 old_value="Hello!",
102 new_value="Hi there! How can I help?",
103 )
105 # Human approves
106 engine.approve(proposal.id, approved_by="human")
108 # Apply changes
109 engine.apply(proposal.id)
110 """
112 def __init__(self):
113 """Initialize evolution engine."""
114 self._proposals: dict[str, EvolutionProposal] = {}
115 self._approvers: dict[str, Callable[[EvolutionProposal], bool]] = {}
117 def propose(
118 self,
119 agent_name: str,
120 change_type: str,
121 description: str,
122 old_value: Any = None,
123 new_value: Any = None,
124 confidence: float = 0.0,
125 risk_level: str = "low",
126 target_files: list[str] | None = None,
127 **metadata,
128 ) -> EvolutionProposal:
129 """
130 Create a new evolution proposal.
132 Args:
133 agent_name: Name of agent to evolve
134 change_type: Type of change
135 description: Human-readable description
136 old_value: Current value
137 new_value: Proposed new value
138 confidence: Confidence score (0-1)
139 risk_level: Risk assessment (low/medium/high)
140 target_files: Files to modify
141 **metadata: Additional metadata
143 Returns:
144 Created EvolutionProposal
145 """
146 proposal = EvolutionProposal(
147 agent_name=agent_name,
148 change_type=change_type,
149 description=description,
150 old_value=old_value,
151 new_value=new_value,
152 confidence=confidence,
153 risk_level=risk_level,
154 target_files=target_files or [],
155 metadata=metadata,
156 )
158 self._proposals[proposal.id] = proposal
160 return proposal
162 def get_proposal(self, proposal_id: str) -> EvolutionProposal | None:
163 """
164 Get a proposal by ID.
166 Args:
167 proposal_id: Proposal ID
169 Returns:
170 EvolutionProposal if found, None otherwise
171 """
172 return self._proposals.get(proposal_id)
174 def list_proposals(
175 self,
176 status: EvolutionStatus | None = None,
177 agent_name: str | None = None,
178 ) -> list[EvolutionProposal]:
179 """
180 List proposals.
182 Args:
183 status: Filter by status
184 agent_name: Filter by agent name
186 Returns:
187 List of matching proposals
188 """
189 proposals = list(self._proposals.values())
191 if status:
192 proposals = [p for p in proposals if p.status == status]
194 if agent_name:
195 proposals = [p for p in proposals if p.agent_name == agent_name]
197 return proposals
199 def approve(
200 self,
201 proposal_id: str,
202 approved_by: str = "human",
203 ) -> bool:
204 """
205 Approve a proposal.
207 Args:
208 proposal_id: Proposal ID
209 approved_by: Who approved
211 Returns:
212 True if approved, False if not found
213 """
214 proposal = self._proposals.get(proposal_id)
215 if not proposal:
216 return False
218 if proposal.status != EvolutionStatus.PENDING:
219 return False
221 proposal.status = EvolutionStatus.APPROVED
222 proposal.approved_at = time.time()
223 proposal.approved_by = approved_by
225 return True
227 def reject(
228 self,
229 proposal_id: str,
230 reason: str = "",
231 ) -> bool:
232 """
233 Reject a proposal.
235 Args:
236 proposal_id: Proposal ID
237 reason: Reason for rejection
239 Returns:
240 True if rejected, False if not found
241 """
242 proposal = self._proposals.get(proposal_id)
243 if not proposal:
244 return False
246 if proposal.status != EvolutionStatus.PENDING:
247 return False
249 proposal.status = EvolutionStatus.REJECTED
250 proposal.metadata["rejection_reason"] = reason
252 return True
254 def apply(self, proposal_id: str) -> bool:
255 """
256 Apply an approved proposal. Performs actual file modifications.
258 Args:
259 proposal_id: Proposal ID
261 Returns:
262 True if applied, False otherwise
263 """
264 proposal = self._proposals.get(proposal_id)
265 if not proposal:
266 return False
268 if proposal.status != EvolutionStatus.APPROVED:
269 return False
271 try:
272 target_files = proposal.target_files or []
273 old_val = proposal.old_value
274 new_val = proposal.new_value
276 if target_files and old_val is not None and new_val is not None:
277 for fpath in target_files:
278 if not os.path.exists(fpath):
279 continue
280 with open(fpath) as f:
281 content = f.read()
282 old_str = str(old_val)
283 new_str = str(new_val)
284 if old_str in content:
285 content = content.replace(old_str, new_str, 1)
286 with open(fpath, "w") as f:
287 f.write(content)
288 proposal.metadata["modified_file"] = fpath
290 proposal.status = EvolutionStatus.APPLIED
291 proposal.applied_at = time.time()
292 return True
293 except Exception as e:
294 proposal.status = EvolutionStatus.FAILED
295 proposal.metadata["error"] = str(e)
296 return False
298 def register_approver(
299 self,
300 agent_name: str,
301 approver: Callable[[EvolutionProposal], bool],
302 ) -> None:
303 """
304 Register an approver for an agent.
306 Args:
307 agent_name: Agent name
308 approver: Approval function
309 """
310 self._approvers[agent_name] = approver
312 def auto_approve(self, proposal_id: str) -> bool:
313 """
314 Auto-approve using registered approver.
316 Args:
317 proposal_id: Proposal ID
319 Returns:
320 True if approved, False otherwise
321 """
322 proposal = self._proposals.get(proposal_id)
323 if not proposal:
324 return False
326 approver = self._approvers.get(proposal.agent_name)
327 if not approver:
328 return False
330 if approver(proposal):
331 return self.approve(proposal_id, approved_by="auto")
333 return False
335 def get_stats(self) -> dict[str, Any]:
336 """
337 Get evolution statistics.
339 Returns:
340 Dict with proposal counts by status
341 """
342 stats = {
343 "pending": 0,
344 "approved": 0,
345 "rejected": 0,
346 "applied": 0,
347 "failed": 0,
348 "total": len(self._proposals),
349 }
351 for proposal in self._proposals.values():
352 stats[proposal.status.value] += 1
354 return stats