Load agents from code files. JSON file content example: { "metadata": { "saved_at": str, # "YYYY-MM-DD HH:MM:SS" "num_agents": int, # total agent count "num_versions": int # total version count },
(self)
| 258 | return agent_configs |
| 259 | |
| 260 | async def _load_from_code(self): |
| 261 | """Load agents from code files. |
| 262 | |
| 263 | JSON file content example: |
| 264 | { |
| 265 | "metadata": { |
| 266 | "saved_at": str, # "YYYY-MM-DD HH:MM:SS" |
| 267 | "num_agents": int, # total agent count |
| 268 | "num_versions": int # total version count |
| 269 | }, |
| 270 | "agents": { |
| 271 | "agent_name": { |
| 272 | "current_version": "1.0.0", |
| 273 | "versions": { |
| 274 | "1.0.0": { |
| 275 | "name": str, |
| 276 | "description": str, |
| 277 | "metadata": dict, |
| 278 | "version": str, |
| 279 | "cls": Type[Agent], |
| 280 | "config": dict, |
| 281 | "instance": Agent, # will be built when needed |
| 282 | "function_calling": dict, |
| 283 | "text": str, |
| 284 | "args_schema": BaseModel, |
| 285 | "code": str |
| 286 | }, |
| 287 | ... |
| 288 | } |
| 289 | } |
| 290 | } |
| 291 | } |
| 292 | """ |
| 293 | |
| 294 | agent_configs: Dict[str, AgentConfig] = {} |
| 295 | |
| 296 | # If save file does not exist yet, nothing to load |
| 297 | if not os.path.exists(self.save_path): |
| 298 | logger.info(f"| 📂 Agent config file not found at {self.save_path}, skipping code-based loading") |
| 299 | return agent_configs |
| 300 | |
| 301 | # Load all agent configs from json file |
| 302 | try: |
| 303 | with open(self.save_path, "r", encoding="utf-8") as f: |
| 304 | load_data = json.load(f) |
| 305 | except json.JSONDecodeError as e: |
| 306 | logger.warning(f"| ⚠️ Failed to parse agent config JSON from {self.save_path}: {e}") |
| 307 | return agent_configs |
| 308 | |
| 309 | metadata = load_data.get("metadata", {}) |
| 310 | agents_data = load_data.get("agents", {}) |
| 311 | |
| 312 | async def register_agent_class(agent_name: str, agent_data: Dict[str, Any]) -> Optional[Tuple[str, Dict[str, AgentConfig], Optional[AgentConfig]]]: |
| 313 | """Load all versions for a single agent from JSON.""" |
| 314 | try: |
| 315 | current_version = agent_data.get("current_version", "1.0.0") |
| 316 | versions = agent_data.get("versions", {}) |
| 317 |
no test coverage detected