Set variable values in an agent and create a new version. Args: agent_name: Name of the agent to update variable_updates: Dictionary mapping variable names to new values. For agents, this is typically {"code": new_code_string} new_
(self, agent_name: str, variable_updates: Dict[str, Any], new_version: Optional[str] = None, description: Optional[str] = None)
| 1183 | return trainable_variables |
| 1184 | |
| 1185 | async def set_variables(self, agent_name: str, variable_updates: Dict[str, Any], new_version: Optional[str] = None, description: Optional[str] = None) -> AgentConfig: |
| 1186 | """Set variable values in an agent and create a new version. |
| 1187 | |
| 1188 | Args: |
| 1189 | agent_name: Name of the agent to update |
| 1190 | variable_updates: Dictionary mapping variable names to new values. |
| 1191 | For agents, this is typically {"code": new_code_string} |
| 1192 | new_version: New version string. If None, auto-increments from current version. |
| 1193 | description: Description for this version update |
| 1194 | |
| 1195 | Returns: |
| 1196 | AgentConfig: Updated agent configuration |
| 1197 | """ |
| 1198 | async with self._variables_lock: |
| 1199 | original_config = self._agent_configs.get(agent_name) |
| 1200 | if original_config is None: |
| 1201 | raise ValueError(f"Agent {agent_name} not found. Use register() to register a new agent.") |
| 1202 | |
| 1203 | # For agents, variable_updates format is {"name": "agent_name", "variables": "agent code"} |
| 1204 | # Extract the new code from "variables" field |
| 1205 | if "variables" not in variable_updates: |
| 1206 | raise ValueError(f"variable_updates must contain 'variables' field with agent code, got: {list(variable_updates.keys())}") |
| 1207 | |
| 1208 | new_code = variable_updates["variables"] |
| 1209 | if not isinstance(new_code, str): |
| 1210 | raise ValueError(f"Agent code must be a string, got {type(new_code)}") |
| 1211 | |
| 1212 | # Load agent class from code |
| 1213 | class_name = dynamic_manager.extract_class_name_from_code(new_code) |
| 1214 | if not class_name: |
| 1215 | raise ValueError(f"Cannot extract class name from code") |
| 1216 | |
| 1217 | try: |
| 1218 | agent_cls = dynamic_manager.load_class( |
| 1219 | new_code, |
| 1220 | class_name=class_name, |
| 1221 | base_class=Agent, |
| 1222 | context="agent" |
| 1223 | ) |
| 1224 | except Exception as e: |
| 1225 | logger.error(f"| ❌ Failed to load agent class from code: {e}") |
| 1226 | raise ValueError(f"Failed to load agent class from code: {e}") |
| 1227 | |
| 1228 | # Use update() function to handle version management and persistence |
| 1229 | # Pass the code directly to avoid re-extracting from dynamically created class |
| 1230 | update_description = description or f"Updated code for {agent_name}" |
| 1231 | return await self.update( |
| 1232 | agent_cls=agent_cls, |
| 1233 | agent_config_dict=original_config.config, |
| 1234 | new_version=new_version, |
| 1235 | description=update_description, |
| 1236 | code=new_code # Pass code directly since agent_cls is dynamically created |
| 1237 | ) |
| 1238 | |
| 1239 | async def cleanup(self): |
| 1240 | """Cleanup all active agents.""" |
nothing calls this directly
no test coverage detected