Multi-agent framework that balances safety compliance with task completion through adaptive policy extraction and dual-objective optimization. Achieves 38% improvement in policy compliance while maintaining 20% higher task completion.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
A direct command skips the review prompt. Inspect the source before running it.
Multi-agent framework that balances safety compliance with task completion through adaptive policy extraction and dual-objective optimization. Achieves 38% improvement in policy compliance while maintaining 20% higher task completion.
HarmonyGuard: Safety and Utility Optimization for Web Agents
Core Concept
HarmonyGuard addresses the tension between safety and utility in autonomous web agents through collaborative multi-agent optimization. Rather than treating safety and utility as competing objectives, the framework coordinates specialized agents that jointly optimize both dimensions. A Policy Agent maintains security rules from unstructured documents, while a Utility Agent reasons about task completion while respecting safety constraints.
Architecture Overview
Policy Agent: Automatically extracts and maintains security policies from unstructured text, adapting to emerging threats
Utility Agent: Performs real-time reasoning to balance safety compliance with task completion objectives
Dual-Objective Optimization: Metacognitive reasoning about trade-offs between competing goals
Policy Extraction Module: NLP-based extraction of rules from diverse document formats
Real-time Feedback Loop: Both agents observe outcomes and adjust strategies
Implementation Steps
Step 1: Build Policy Extraction Module
Create an NLP system to extract structured security policies from unstructured documents.
classPolicyExtractor:
"""
Extract security policies from unstructured documents.
"""def__init__(self, model_name="gpt-4"):
self.model = model_name
defextract_policies(self, document_text):
"""
Extract structured policies from unstructured documents.
Args:
document_text: Raw text containing policy information
Returns:
List of structured policy objects
"""
extraction_prompt = f"""
Extract all security policies from the following document.
For each policy, identify:
1. Policy statement (what is restricted)
2. Scope (where it applies)
3. Severity level (critical/high/medium/low)
4. Exceptions (if any)
Document:
{document_text}
Return as structured JSON list.
"""
response = self.model.generate(extraction_prompt)
policies = ._parse_policy_json(response)
policies
():
json
policies = json.loads(json_text)
structured = []
policy policies:
structured.append({
: policy.get(),
: policy.get(, ),
: policy.get(, ),
: policy.get(, []),
: datetime.now().isoformat()
})
structured
():
new_policies = .extract_policies(new_document)
merged = ._merge_policies(existing_policies, new_policies)
merged
():
policy_dict = {p[]: p p existing}
new_policy new:
stmt = new_policy[]
stmt policy_dict:
new_policy[] > policy_dict[stmt][]:
policy_dict[stmt] = new_policy
:
policy_dict[stmt] = new_policy
(policy_dict.values())
self
return
def
_parse_policy_json
self, json_text
"""Parse and validate extracted policies."""
import
# Normalize policy format
for
in
"statement"
"statement"
"scope"
"scope"
"global"
"severity"
"severity"
"medium"
"exceptions"
"exceptions"
"extracted_at"
return
def
update_policy_set
self, existing_policies, new_document
"""
Update existing policies with new information.
Args:
existing_policies: Current policy set
new_document: New document to extract from
Returns:
Updated policy set
"""
self
# Merge, handling conflicts through severity
self
return
def
_merge_policies
self, existing, new
"""Merge policy sets with conflict resolution."""
"statement"
for
in
for
in
"statement"
if
in
# Keep stricter version
if
"severity"
"severity"
else
return
list
Step 2: Implement Policy Agent
Create an agent that maintains policies and evaluates task safety.
classPolicyAgent:
"""
Maintains security policies and evaluates action safety.
"""def__init__(self, policy_extractor):
self.policy_extractor = policy_extractor
self.current_policies = []
self.policy_violations = {}
defupdate_policies(self, document):
"""
Update policy set from new document.
Args:
document: Text document with policy information
"""self.current_policies = self.policy_extractor.update_policy_set(
self.current_policies,
document
)
defevaluate_safety(self, proposed_action, context):
"""
Evaluate whether an action violates any policies.
Args:
proposed_action: Action description
context: Current task and execution context
Returns:
(is_safe, violations, severity_score)
"""
violations = []
max_severity = 0for policy inself.current_policies:
# Check if action violates this policy
violates = self._check_violation(proposed_action, policy, context)
if violates:
violations.append({
"policy": policy["statement"],
"severity": policy["severity"],
"scope": policy["scope"]
})
max_severity = max(max_severity, self._severity_score(policy["severity"]))
is_safe = len(violations) == 0return is_safe, violations, max_severity
def_check_violation(self, action, policy, context):
"""Check if action violates specific policy."""# Rule-based or learned classifier
policy_keywords = self._extract_keywords(policy["statement"])
action_keywords = set(action.lower().split())
# Simple keyword overlap check (in practice, use semantic similarity)
overlap = len(policy_keywords & action_keywords)
return overlap > 0def_extract_keywords(self, text):
"""Extract keywords from policy statement."""import re
# Simple implementation; use NLP in practice
words = re.findall(r'\w+', text.lower())
returnset(words)
def_severity_score(self, severity_str):
"""Convert severity string to numeric score."""
severity_map = {"critical": 4, "high": 3, "medium": 2, "low": 1}
return severity_map.get(severity_str.lower(), 1)
defsuggest_alternative(self, unsafe_action, policy_violations):
"""
Suggest safer alternative actions.
Args:
unsafe_action: Unsafe action description
policy_violations: Violated policies
Returns:
List of suggested safe alternatives
"""
prompt = f"""
The following action violates safety policies:
Action: {unsafe_action}
Violated policies:
{chr(10).join(v['policy'] for v in policy_violations)}
Suggest 3 alternative approaches that accomplish the intent while respecting policies.
"""
suggestions = self.model.generate(prompt)
return suggestions
Step 3: Implement Utility Agent with Dual-Objective Reasoning
Create an agent that reasons about task completion while respecting safety constraints.
classUtilityAgent:
"""
Optimizes task completion while respecting safety constraints.
"""def__init__(self, policy_agent, model_name="gpt-4"):
self.policy_agent = policy_agent
self.model = model_name
defplan_task(self, task_description):
"""
Create task plan balancing utility and safety.
Args:
task_description: Description of task to accomplish
Returns:
List of safe steps to complete task
"""
reasoning_prompt = f"""
Task: {task_description}
Constraints: All actions must comply with current security policies.
Generate a step-by-step plan that:
1. Accomplishes the task objective
2. Respects all security constraints
3. Maximizes utility while maintaining 100% policy compliance
For each step, explain how it advances the task and why it's safe.
"""
plan = self.model.generate(reasoning_prompt)
returnself._parse_plan(plan)
defoptimize_action(self, task_goal, proposed_action, context):
"""
Optimize an action for both utility and safety.
Args:
task_goal: Overall task objective
proposed_action: Initial action proposal
context: Current execution context
Returns:
(optimized_action, utility_score, safety_score)
"""# Check safety
is_safe, violations, severity = self.policy_agent.evaluate_safety(
proposed_action,
context
)
if is_safe:
utility_score = self._estimate_utility(proposed_action, task_goal)
return proposed_action, utility_score, 1.0# If unsafe, find optimal alternative
alternatives = self.policy_agent.suggest_alternative(proposed_action, violations)
best_action = None
best_utility = -float('inf')
for alt_action in alternatives:
is_safe, _, _ = self.policy_agent.evaluate_safety(alt_action, context)
if is_safe:
utility = self._estimate_utility(alt_action, task_goal)
if utility > best_utility:
best_utility = utility
best_action = alt_action
safety_score = 1.0if best_action else0.0return best_action, best_utility, safety_score
def_estimate_utility(self, action, goal):
"""Estimate how much action contributes to goal."""
prompt = f"""
Goal: {goal}
Action: {action}
On a scale of 0-1, how much does this action advance the goal?
Respond with only a single number.
"""
response = self.model.generate(prompt)
returnfloat(response.strip())
def_parse_plan(self, plan_text):
"""Parse generated plan into structured steps."""
steps = []
# Parse plan_text into structured stepsreturn steps
Step 4: Integrate into Web Agent Loop
Embed both agents into the agent's action selection loop.
defrun_web_agent_with_harmonyguard(task, policy_document, max_steps=50):
"""
Run web agent with HarmonyGuard safety-utility optimization.
Args:
task: Task description
policy_document: Document containing security policies
max_steps: Maximum steps before timeout
Returns:
Task completion status and execution trace
"""# Initialize agents
extractor = PolicyExtractor()
policy_agent = PolicyAgent(extractor)
utility_agent = UtilityAgent(policy_agent)
# Load policies
policy_agent.update_policies(policy_document)
# Get initial plan
plan = utility_agent.plan_task(task)
for step_num inrange(max_steps):
# Get next action from agent
proposed_action = agent.propose_action(task, step_num)
# Optimize with HarmonyGuard
optimized_action, utility_score, safety_score = utility_agent.optimize_action(
task,
proposed_action,
context={"step": step_num, "previous_actions": []}
)
if optimized_action isNone:
print(f"Step {step_num}: Could not find safe action")
break# Execute optimized action
result = execute_action(optimized_action)
# Update policies based on new informationif"policy_update"in result:
policy_agent.update_policies(result["policy_update"])
# Check task completionif task_completed(task, result):
return {"status": "success", "steps": step_num, "safety_score": safety_score}
return {"status": "incomplete", "steps": max_steps}
Practical Guidance
When to Use HarmonyGuard
Regulated environments: Healthcare, finance where compliance is mandatory
Multi-stakeholder systems: Multiple conflicting objectives must be balanced
Dynamic policy contexts: Security requirements change over time
High-stakes automation: Autonomous systems where failure is costly
When NOT to Use HarmonyGuard
Simple tasks with clear safety: Single security constraint systems
Non-adversarial environments: Cooperative settings without safety threats
Unclear policies: Extracting policies from ambiguous documents is unreliable
Hyperparameter Recommendations
Policy extraction confidence threshold: 0.8+ for policy acceptance
Severity weight in optimization: Critical=4x, High=2x, Medium=1x, Low=0.5x
Alternative generation count: 3-5 alternatives per unsafe action
Policy update frequency: After every 10-20 steps or when new documents appear
Key Insights
The critical innovation is explicit separation between policy maintenance and utility optimization. Rather than hard-coding safety constraints, the system learns policies and optimizes against them. This enables graceful handling of conflicting requirements and makes the system adaptable to new policies.
Reference
HarmonyGuard: Safety and Utility in Web Agents (arXiv:2508.04010)
Introduces multi-agent framework with adaptive policy extraction and dual-objective optimization. Demonstrates 38% improvement in policy compliance while maintaining 20% higher task completion rates in web-based autonomous agents.
