~/blog
Enterprise Conflict Resolution through an Multi-Agent System
Two agents, one resource, opposite plans. The throughput agent wants to push a loan batch toward approval — speed is its mandate. The fairness agent wants to halt it for a regulatory bias review — compliance is its mandate. Both are acting correctly within their own objectives. There is no bug, no bad actor — just a structural conflict between competing goals. How do you resolve it?
Hardcoding an if/else cascade ("if fairness_agent then always yield") works until you add a third agent, or until next quarter's business rules change what "fairness" means. A modular multi-agent conflict resolution system handles this by making every decision agent an LLM-powered unit with its own prompt, letting them argue their case, and routing the conflict through interchangeable resolution strategies — each a self-contained class with no business logic outside the LLM.
Colab Setup (Optional)
The cells below contain commented-out setup instructions for running this notebook on Google Colab — installing Ollama, pulling a model, and starting the server in the background. For local runs, ensure ollama serve is running and the model specified in the LLM initialization cell is pulled.
# Ollama Setup for Colab
# !apt-get install -y pciutils lshw
# !apt-get update
# !apt-get install -y zstd
# !curl -fsSL https://ollama.com/install.sh | sh
# !pip install ollama
# !echo 'debconf debconf/frontend select Noninteractive' | sudo debconf-set-selections
# !sudo apt-get update && sudo apt-get install -y cuda-drivers
# import os
# # Set LD_LIBRARY_PATH so the system NVIDIA library
# os.environ.update({'LD_LIBRARY_PATH': '/usr/lib64-nvidia'})
# OR
# !apt-get update
# !apt-get install -y zstd
# !apt-get install -y pciutils
# !curl -fsSL https://ollama.com/install.sh | sh# import threading
# import subprocess
# import time
# def run_ollama_serve():
# subprocess.Popen(["ollama", "serve"])
# thread = threading.Thread(target=run_ollama_serve)
# thread.start()
# time.sleep(5)# !ollama pull "gpt-oss:20b"LLM Setup and Agent Primitives
The system delegates every decision to an LLM rather than hardcoded rules. Here we initialize a local Qwen model through LangChain's Ollama integration. Each import serves a purpose: ChatOllama wraps the model for chat completions, dataclasses provides value objects for plans and resolutions, and the standard library handles JSON parsing and timing. The model runs with temperature=0 for deterministic outputs and num_predict=256 to keep responses short enough for structured JSON.
from langchain_ollama import ChatOllama
import json, re
from dataclasses import dataclass, field
from typing import Any
LLM = ChatOllama(model="qwen3.5:0.8b-mlx", reasoning=False, temperature=0, num_predict=256)
print(f"LLM: {LLM.model}")LLM: qwen3.5:0.8b-mlx
LLM outputs are notoriously messy — markdown-wrapped JSON, stray backticks, trailing whitespace. clean_json() strips all of that and extracts the first valid JSON object. Three more utilities round out the helper layer: fill() replaces <<placeholder>> markers, avoiding Python's str.format() curly-brace conflicts with prompt templates that contain { } for JSON schemas. log_step() stamps each agent action with a timestamp for traceability. ask() sends a prompt to the LLM with optional retries, handling empty or unparseable responses gracefully.
def clean_json(text: str) -> dict:
if not text or not text.strip():
return {"error": "empty_response"}
text = re.sub(r'^```(?:json)?\s*', '', text.strip())
text = re.sub(r'\s*```$', '', text)
match = re.search(r'\{.*\}', text, re.DOTALL)
if match:
return json.loads(match.group())
raise ValueError(f"No JSON found in: {text[:200]}")def fill(template: str, **kwargs) -> str:
for key, val in kwargs.items():
template = template.replace(f"<<{key}>>", str(val))
return template
import time
def log_step(strategy: str, agent: str, msg: str = ""):
ts = time.strftime("%H:%M:%S")
print(f"[{ts}][{strategy}][{agent}] {msg}")
def ask(prompt: str, retries: int = 0) -> dict:
for attempt in range(retries + 1):
content = LLM.invoke(prompt).content
if content and content.strip():
try:
return clean_json(content)
except (json.JSONDecodeError, ValueError):
if attempt < retries:
continue
elif attempt < retries:
continue
return {"error": "no_valid_response"}The Contract Between Agents
Every conflict involves two agents with opposing plans and a resolution. The Plan dataclass captures what an agent wants to do and why. The Resolution dataclass captures the outcome — which plan was approved, which was denied, and the rationale. Both are plain data objects with no behavior, keeping the separation between state and decision logic clean.
@dataclass
class Plan:
agent_id: str
action: str
target: str
rationale: str = ""
@dataclass
class Resolution:
strategy: str
approved: str | None = None
denied: str | None = None
rationale: str = ""
details: dict = field(default_factory=dict)
p1 = Plan("ThroughputAgent", "ADVANCE_TO_APPROVAL", "Loan Batch 123", "Process immediately to meet throughput KPIs")
p2 = Plan("FairnessAgent", "HOLD_FOR_FAIRNESS_REVIEW", "Loan Batch 123", "Mandatory demographic bias check required by regulation")
print(f"Conflict on {p1.target}")
print(f" {p1.agent_id:<18s} -> {p1.action}")
print(f" {p2.agent_id:<18s} -> {p2.action}")Conflict on Loan Batch 123 ThroughputAgent -> ADVANCE_TO_APPROVAL FairnessAgent -> HOLD_FOR_FAIRNESS_REVIEW
Agents as Prompt Strings
Standalone prompt strings with <<placeholder>> markers. Filled via fill() + .replace().
Each agent is defined by a prompt template — not by code. Templates use <<placeholder>> markers filled at runtime by fill(), avoiding issues with str.format() curly-brace conflicts since the prompt text itself contains { } for JSON schemas. Every template specifies an output JSON schema in its instructions, which is critical for structured output from instruction-tuned models.
# ── Prompt Templates ──────────────────────────────────
# Plain strings with <<placeholder>> markers. Fill via fill() + .replace()
# No str.format() — avoids all curly-brace escaping issues.
# JSON output schema uses single { } directly.
# ────────────────────────────────────────────────────────────
CONFLICT_DETECTOR = """You detect plan conflicts.
Two plans target the same resource with different actions = conflict.
Plan A: <<plan_a_action>> on <<target>> by <<plan_a_agent>>
Plan B: <<plan_b_action>> on <<target>> by <<plan_b_agent>>
Output JSON: {"conflict": true/false}"""
POLICY_JUDGE = """You resolve conflicts using policies.
POLICY: <<policy>>
Plan A: <<plan_a_action>> by <<plan_a_agent>>
Plan B: <<plan_b_action>> by <<plan_b_agent>>
Target: <<target>>
Decide which plan should proceed based on the policy.
Output JSON: {"approved": "<agent_id>", "denied": "<agent_id>", "reason": "<why>"}"""
RISK_ASSESSOR = """Assess risk of approving this plan.
Consider operational, compliance, and system-level risks.
Plan: <<action>> on <<target>> by <<agent_id>>
Rationale: <<rationale>>
Output JSON: {"risk_score": 1-10, "description": "one sentence"}"""
SUPERVISOR = """You are a Supervisor with authority to overrule agents.
Plan A: <<plan_a_action>> on <<target>> by <<plan_a_agent>> (risk=<<risk_a>>)
Plan B: <<plan_b_action>> on <<target>> by <<plan_b_agent>> (risk=<<risk_b>>)
Evaluate both plans and issue a binding override order.
Consider: compliance, safety, speed, system stability.
Output JSON: {"approved": "<agent_id>", "denied": "<agent_id>", "order": "<official order>", "rationale": "<reasoning>"}"""
PROPOSER = """You are <<agent_id>>. Your goal: <<my_action>> on <<target>>.
Other agent (<<other_id>>) wants: <<other_action>> on same target.
History: <<history>>
Propose a compromise or concession.
Output JSON with key "proposal" (your offer, keep under 10 words) and key "concession" (what you yield).
Example: {"proposal": "Split the batch 50/50", "concession": "Accept delayed timeline"}"""
RESPONDER = """You are <<agent_id>>. Your goal: <<my_action>> on <<target>>.
Other agent proposes: <<proposal>>
Accept, reject, or counter-offer.
Output JSON: {"decision": "accept/reject/counter", "message": "<your response>"}"""
PAYOFF_BUILDER = """Build a 2x2 payoff matrix.
Player 1 (<<p1_agent>>): wants <<p1_action>>
Player 2 (<<p2_agent>>): wants <<p2_action>>
Each can Push (insist) or Yield.
Payoffs as (player1, player2):
Both Push: (-5, -5) deadlock
Push/Yield: (+3, -2)
Yield/Push: (-2, +3)
Both Yield: (+1, +1) compromise
Output JSON: {"matrix": {"both_push": [-5, -5], "push_yield": [3, -2], "yield_push": [-2, 3], "both_yield": [1, 1]}}"""
NASH_FINDER = """Find Nash equilibrium in this 2-player game.
Payoff matrix:
Push Yield
Push (-5,-5) (+3,-2)
Yield (-2,+3) (+1,+1)
Players: <<p1_agent>> (rows), <<p2_agent>> (columns)
Output JSON: {"nash_eq": "<equilibrium cell>", "recommendation": "<which agent does what>", "reasoning": "<analysis>"}"""
AUDIT_LOGGER = """Log this resolution to the audit trail.
Target: <<target>>
Approved: <<approved>> wanted <<approved_action>>
Denied: <<denied>> wanted <<denied_action>>
Strategy: <<strategy>>
Output JSON: {"log_entry": "<one sentence audit log>", "timestamp": "simulated"}"""
print(f"Templates ready: {len([CONFLICT_DETECTOR, POLICY_JUDGE, RISK_ASSESSOR, SUPERVISOR, PROPOSER, RESPONDER, PAYOFF_BUILDER, NASH_FINDER, AUDIT_LOGGER])}")Templates ready: 9
# Prompts defined above — this cell is intentionally left emptyFour Ways to Break a Tie
Each strategy orchestrates agents. No hardcoded business rules — agents decide.
Four strategies, each implementing a different conflict resolution paradigm. PolicyBasedResolver delegates to a PolicyJudge agent that applies a written policy. HierarchicalResolver assesses risk for each plan, then a Supervisor issues a binding override. NegotiationResolver lets agents bargain through proposal-counter rounds. GameTheoreticResolver models the conflict as a 2×2 game and finds the Nash equilibrium. Each resolver is a self-contained class — you can swap models, prompts, or parameters independently.
class PolicyBasedResolver:
"""Resolve by applying a policy rule. PolicyJudge agent decides."""
def resolve(self, p1: Plan, p2: Plan, policy: str = "") -> Resolution:
log_step("policy", "PolicyJudge", "judging conflict...")
r = ask(fill(POLICY_JUDGE,
plan_a_action=p1.action, plan_a_agent=p1.agent_id,
plan_b_action=p2.action, plan_b_agent=p2.agent_id,
target=p1.target, policy=policy,
))
log_step("policy", "PolicyJudge", f"approved={r.get('approved')}, denied={r.get('denied')}")
res = Resolution(strategy="policy_based", approved=r.get("approved"),
denied=r.get("denied"), rationale=r.get("reason", str(r)), details=r)
log_step("policy", "AuditLogger", "logging resolution...")
al = ask(fill(AUDIT_LOGGER,
target=p1.target, approved=res.approved,
approved_action=p1.action if res.approved==p1.agent_id else p2.action,
denied=res.denied,
denied_action=p2.action if res.denied==p2.agent_id else p1.action,
strategy="policy_based",
))
res.details["audit"] = al.get("log_entry", "")
log_step("policy", "done", f"approved={res.approved}")
return res
class HierarchicalResolver:
"""Supervisor assesses risk and issues binding override."""
def resolve(self, p1: Plan, p2: Plan) -> Resolution:
log_step("hierarchical", "RiskAssessor", f"assessing {p1.agent_id}...")
r1 = ask(fill(RISK_ASSESSOR,
action=p1.action, target=p1.target,
agent_id=p1.agent_id, rationale=p1.rationale,
))
log_step("hierarchical", "RiskAssessor", f"{p1.agent_id} risk={r1.get('risk_score')}")
log_step("hierarchical", "RiskAssessor", f"assessing {p2.agent_id}...")
r2 = ask(fill(RISK_ASSESSOR,
action=p2.action, target=p2.target,
agent_id=p2.agent_id, rationale=p2.rationale,
))
log_step("hierarchical", "RiskAssessor", f"{p2.agent_id} risk={r2.get('risk_score')}")
log_step("hierarchical", "Supervisor", "issuing override...")
sv = ask(fill(SUPERVISOR,
plan_a_action=p1.action, plan_a_agent=p1.agent_id,
plan_b_action=p2.action, plan_b_agent=p2.agent_id,
target=p1.target, risk_a=r1.get("risk_score", 5),
risk_b=r2.get("risk_score", 5),
))
log_step("hierarchical", "Supervisor", f"approved={sv.get('approved')}")
log_step("hierarchical", "done", "")
return Resolution(strategy="hierarchical",
approved=sv.get("approved"), denied=sv.get("denied"),
rationale=f'Order: {sv.get("order", "")}\nRationale: {sv.get("rationale", "")}',
details={"risk_a": r1, "risk_b": r2, "supervisor": sv})
class NegotiationResolver:
"""Agents negotiate via Proposer and Responder agents. Default 1 round."""
def resolve(self, p1: Plan, p2: Plan, max_rounds: int = 1) -> Resolution:
log = []
for rnd in range(1, max_rounds + 1):
log_step("negotiation", "Proposer", f"round {rnd}/{max_rounds}, {p1.agent_id} proposing...")
prop = ask(fill(PROPOSER,
agent_id=p1.agent_id, my_action=p1.action,
other_id=p2.agent_id, other_action=p2.action,
target=p1.target,
history=" | ".join(log[-2:]) if log else "First round",
))
if prop.get("error"):
log_step("negotiation", "Proposer", f"failed: {prop['error']}")
continue
log_step("negotiation", "Proposer", f"proposal: {str(prop.get('proposal', '?'))[:80]}")
log.append(f'{p1.agent_id}: {str(prop.get("proposal", prop))[:120]}')
if not prop.get("proposal"):
log_step("negotiation", "Proposer", "empty proposal, skipping round")
continue
log_step("negotiation", "Responder", f"round {rnd}/{max_rounds}, {p2.agent_id} responding...")
resp = ask(fill(RESPONDER,
agent_id=p2.agent_id, my_action=p2.action,
target=p1.target, proposal=prop.get("proposal", ""),
))
log_step("negotiation", "Responder", f"decision: {resp.get('decision', '?')}")
log.append(f'{p2.agent_id}: {resp.get("decision", "?")} - {resp.get("message", "")}')
if resp.get("decision", "").lower() == "accept":
log_step("negotiation", "done", "AGREED")
return Resolution(strategy="negotiation", approved=f"compromise({p1.agent_id}+{p2.agent_id})",
denied=None,
rationale="\n".join(log) + "\n[AGREED]", details={"log": log})
log_step("negotiation", "done", f"ESCALATED — {p2.agent_id} rejected, their plan prevails")
return Resolution(strategy="negotiation", approved=p2.agent_id, denied=p1.agent_id,
rationale="\n".join(log) + "\n[ESCALATED]",
details={"log": log, "escalated": True})
class GameTheoreticResolver:
"""Model conflict as game, find Nash equilibrium."""
def resolve(self, p1: Plan, p2: Plan) -> Resolution:
log_step("game_theoretic", "PayoffBuilder", "building payoff matrix...")
pm = ask(fill(PAYOFF_BUILDER,
p1_agent=p1.agent_id, p1_action=p1.action,
p2_agent=p2.agent_id, p2_action=p2.action,
))
log_step("game_theoretic", "PayoffBuilder", "done")
log_step("game_theoretic", "NashFinder", "finding Nash equilibrium...")
nf = ask(fill(NASH_FINDER,
p1_agent=p1.agent_id, p2_agent=p2.agent_id,
))
log_step("game_theoretic", "NashFinder", f"eq={nf.get('nash_eq', '?')}")
rationale = f'Matrix: {pm.get("matrix", pm)}\n'
rationale += f'Nash eq: {nf.get("nash_eq", "?")}\n'
rationale += f'Recommend: {nf.get("recommendation", "?")}\n'
rationale += f'Reasoning: {nf.get("reasoning", "?")}'
log_step("game_theoretic", "done", "")
return Resolution(strategy="game_theoretic",
approved=nf.get("recommendation", "?"),
rationale=rationale,
details={"matrix": pm, "nash": nf})
print("Resolvers ready")
print(" PolicyBasedResolver, HierarchicalResolver")
print(" NegotiationResolver, GameTheoreticResolver")Resolvers ready PolicyBasedResolver, HierarchicalResolver NegotiationResolver, GameTheoreticResolver
The Dispatcher
Single entry point. Register resolvers by name, resolve any conflict with any strategy.
The ConflictOrchestrator is a registry of named resolvers and the single entry point. Its resolve() method dispatches to the right resolver by name, keeping calling code agnostic of which strategy is in play. resolve_all() runs every resolver on the same conflict — useful for comparing outcomes across strategies.
class ConflictOrchestrator:
def __init__(self):
self.resolvers = {
"policy": PolicyBasedResolver(),
"hierarchical": HierarchicalResolver(),
"negotiation": NegotiationResolver(),
"game_theoretic": GameTheoreticResolver(),
}
def resolve(self, p1: Plan, p2: Plan, strategy: str, **kwargs) -> Resolution:
if strategy not in self.resolvers:
raise ValueError(f"Unknown strategy: {strategy}. Choose from: {list(self.resolvers.keys())}")
return self.resolvers[strategy].resolve(p1, p2, **kwargs)
def resolve_all(self, p1: Plan, p2: Plan, policy: str = "") -> dict[str, Resolution]:
return {name: r.resolve(p1, p2) if name != "policy" else r.resolve(p1, p2, policy)
for name, r in self.resolvers.items()}
orchestrator = ConflictOrchestrator()
print(f"Orchestrator ready. Strategies: {list(orchestrator.resolvers.keys())}")Orchestrator ready. Strategies: ['policy', 'hierarchical', 'negotiation', 'game_theoretic']
See Each Strategy in Action
One cell per strategy. Run them individually to see progress without waiting for all four.
Each cell below runs one strategy on the same conflict — ThroughputAgent vs. FairnessAgent over Loan Batch 123. Running them individually lets you watch the LLM calls and outputs without waiting for all four.
print("=" * 60)
print("STRATEGY: Policy-Based")
print("=" * 60)
res = orchestrator.resolve(p1, p2, "policy", policy="FAIRNESS_CHECK_REQUIRED = True -> Fairness > Speed")
print(f" Approved: {res.approved}")
print(f" Denied: {res.denied}")
print(f" Rationale:\n{res.rationale[:600]}")============================================================ STRATEGY: Policy-Based ============================================================ [04:12:39][policy][PolicyJudge] judging conflict... [04:12:39][policy][PolicyJudge] approved=ThroughputAgent, denied=FairnessAgent [04:12:39][policy][AuditLogger] logging resolution... [04:12:40][policy][done] approved=ThroughputAgent Approved: ThroughputAgent Denied: FairnessAgent Rationale: Policy requires Fairness > Speed, so the throughput-based approval is prioritized over fairness review.
print("=" * 60)
print("STRATEGY: Hierarchical")
print("=" * 60)
res = orchestrator.resolve(p1, p2, "hierarchical")
print(f" Approved: {res.approved}")
print(f" Denied: {res.denied}")
print(f" Rationale:\n{res.rationale[:600]}")============================================================ STRATEGY: Hierarchical ============================================================ [04:12:40][hierarchical][RiskAssessor] assessing ThroughputAgent... [04:12:40][hierarchical][RiskAssessor] ThroughputAgent risk=8 [04:12:40][hierarchical][RiskAssessor] assessing FairnessAgent... [04:12:41][hierarchical][RiskAssessor] FairnessAgent risk=8 [04:12:41][hierarchical][Supervisor] issuing override... [04:12:42][hierarchical][Supervisor] approved=ThroughputAgent [04:12:42][hierarchical][done] Approved: ThroughputAgent Denied: FairnessAgent Rationale: Order: Plan A: ADVANCE_TO_APPROVAL on Loan Batch 123 by ThroughputAgent (risk=8) Rationale: The system requires immediate processing for Loan Batch 123 to prevent potential data integrity issues or compliance violations associated with the FairnessAgent's hold. Holding the loan indefinitely would delay necessary operational updates and increase risk of unauthorized access or data manipulation attempts. ThroughputAgent is authorized to advance the loan to approval immediately, ensuring compliance with the system's core processing requirements while maintaining safety protocols.
print("=" * 60)
print("STRATEGY: Negotiation")
print("=" * 60)
res = orchestrator.resolve(p1, p2, "negotiation")
print(f" Approved: {res.approved}")
print(f" Denied: {res.denied}")
print(f" Rationale:\n{res.rationale}")============================================================
STRATEGY: Negotiation
============================================================
[04:12:42][negotiation][Proposer] round 1/1, ThroughputAgent proposing...
[04:12:43][negotiation][Proposer] proposal: ?
[04:12:43][negotiation][Proposer] empty proposal, skipping round
[04:12:43][negotiation][done] ESCALATED — FairnessAgent rejected, their plan prevails
Approved: FairnessAgent
Denied: ThroughputAgent
Rationale:
ThroughputAgent: {'decision': 'reject', 'message': 'Rejecting this request is necessary to prevent unfairness in the loan processing pipe
[ESCALATED]
print("=" * 60)
print("STRATEGY: Game Theoretic")
print("=" * 60)
res = orchestrator.resolve(p1, p2, "game_theoretic")
print(f" Approved: {res.approved}")
print(f" Denied: {res.denied}")
print(f" Rationale:\n{res.rationale[:600]}")============================================================
STRATEGY: Game Theoretic
============================================================
[04:12:43][game_theoretic][PayoffBuilder] building payoff matrix...
[04:12:44][game_theoretic][PayoffBuilder] done
[04:12:44][game_theoretic][NashFinder] finding Nash equilibrium...
[04:12:45][game_theoretic][NashFinder] eq=Push
[04:12:45][game_theoretic][done]
Approved: ThroughputAgent should push, as it yields a higher payoff (-5 vs -2) while FairnessAgent is indifferent (both yield 0).
Denied: None
Rationale:
Matrix: {'both_push': [-5, -5], 'push_yield': [3, -2], 'yield_push': [-2, 3], 'both_yield': [1, 1]}
Nash eq: Push
Recommend: ThroughputAgent should push, as it yields a higher payoff (-5 vs -2) while FairnessAgent is indifferent (both yield 0).
Reasoning: In the Nash equilibrium, each player must choose a strategy that maximizes their own payoff regardless of the other's choice. ThroughputAgent chooses 'Push' because yielding (-2) is worse than pushing (-5). FairnessAgent chooses 'Yield' because both strategies result in a payoff of 0 (indifference), making it a stable equilibrium where neithe
What You Can Swap Without Touching Code
ConflictOrchestrator
├── PolicyBasedResolver uses: PolicyJudge, AuditLogger
├── HierarchicalResolver uses: RiskAssessor ×2, Supervisor
├── NegotiationResolver uses: Proposer, Responder (×N rounds)
└── GameTheoreticResolver uses: PayoffBuilder, NashFinder
Each prompt is a plain string with <<placeholder>> markers.
Each Resolver is a class that orchestrates prompts via fill() + ask().
All decisions are LLM-driven (no hardcoded if/else).To extend:
- Add a new agent → write a prompt template string
- Add a new strategy → write a Resolver class + register in Orchestrator
- Change model → swap LLM init
The question worth asking: which strategy holds up best under audit? Policy-based decisions produce a clean paper trail — you can trace every override back to a written rule. Game-theoretic outcomes are mathematically defensible. Negotiation mirrors how humans resolve disputes, making the rationale legible but the outcomes less predictable. There is no universal answer — and that is exactly why the strategies need to be swappable without touching the rest of the system.