Introduction
OVADARE’s Resolution Engine provides flexible mechanisms for resolving conflicts in multi-agent systems. Developers can customize resolution strategies to suit specific workflows and align with organizational policies.
Customizing Strategies
1. Define Custom Resolution Logic
Extend the default behavior of the ResolutionEngine
to include tailored resolution actions.
from ovadare.conflicts.resolution_engine import ResolutionEngine
from ovadare.conflicts.resolution import Resolution
from ovadare.conflicts.conflict import Conflict
class CustomResolutionEngine(ResolutionEngine):
def _create_resolution(self, conflict: Conflict) -> Resolution:
# Custom resolution logic
corrective_action = {
"action": "limit_access",
"details": {
"agent_id": conflict.related_agent_id,
"message": f"Access restricted due to policy: {conflict.policy_id}"
}
}
explanation = f"Custom resolution applied for conflict {conflict.conflict_id}"
return Resolution(
conflict_id=conflict.conflict_id,
corrective_action=corrective_action,
explanation=explanation
)
# Example usage
custom_engine = CustomResolutionEngine()
conflicts = [conflict_detector.get_conflict_by_id("Conflict001")]
resolutions = custom_engine.generate_resolutions(conflicts)
for resolution in resolutions:
print(f"Resolution: {resolution.to_dict()}")
2. Add Resolution Strategies Based on Policies
Define specific rules within the PolicyManager
to guide conflict resolution.
from ovadare.policies.policy_manager import PolicyManager, Policy
policy_manager = PolicyManager()
# Example: Define a custom policy
custom_policy = Policy(
name="DataAccessPolicy",
rules={
"agent_role": "DataAnalyst",
"resource": "ConfidentialReport",
"action": "read_only"
}
)
policy_manager.add_policy(custom_policy)
print("Custom policy added successfully.")
3. Handle Multi-Agent Scenarios
Use OVADARE to resolve conflicts involving multiple agents simultaneously.
from ovadare.agents.agent_registry import AgentRegistry
# Example: Notify multiple agents
for conflict in conflicts:
agents = AgentRegistry.get_agents_by_resource(conflict.action["resource"])
for agent in agents:
agent.notify_resolution(resolution)
4. Test Resolution Scenarios
Simulate and test custom resolution strategies to ensure effectiveness:
from ovadare.utils.testing import simulate_conflict
conflicts = simulate_conflict(agent_id="Agent001", action={"action": "modify_data", "resource": "RestrictedFile"})
resolutions = custom_engine.generate_resolutions(conflicts)
print("Test Resolutions:", [res.to_dict() for res in resolutions])
Example Use Case
Customize a resolution strategy to prioritize actions:
- Scenario: Two agents attempt to modify the same file.
- Resolution: The system resolves the conflict by granting priority to the agent with higher permissions.
class PriorityResolutionEngine(ResolutionEngine):
def _create_resolution(self, conflict: Conflict) -> Resolution:
priority_agent = "AgentWithHighPriority"
corrective_action = {
"action": "grant_priority",
"details": {
"agent_id": priority_agent,
"message": f"Priority granted to {priority_agent} for action on resource {conflict.action['resource']}."
}
}
return Resolution(
conflict_id=conflict.conflict_id,
corrective_action=corrective_action,
explanation="Resolved based on agent priority."
)
Best Practices
- Leverage Policy Rules: Define clear rules to guide resolution strategies.
- Iterate on Feedback: Use conflict logs to refine and improve strategies over time.
- Test Thoroughly: Simulate diverse scenarios to validate resolution effectiveness.
Learn More
Explore related topics: