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