​
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:

Detect Conflicts DynamicallyMonitor Agent Conflicts