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.
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.
