from anthropic import Anthropic
from openai import OpenAI
class CrossFamilyPipeline:
"""Generator and judge from different model families.
This is the only fix that addresses the root cause of self-preference bias."""
def init(self):
self.generator_client = OpenAI()
self.judge_client = Anthropic()
async def generate(self, query: str) -> str:
response = self.generator_client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": query}]
)
return response.choices[0].message.content
async def evaluate(self, query: str, response: str) -> dict:
evaluation = self.judge_client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{
"role": "user",
"content": f"""Evaluate this customer support response.
ORIGINAL QUERY: {query}
RESPONSE TO EVALUATE: {response}
Score each dimension independently from 1-5.
Think step-by-step before assigning each score.
Dimensions:

ACCURACY: Are all factual claims correct?
COMPLETENESS: Does it fully address the query?
TONE: Is it professional and empathetic?
ACTIONABILITY: Does the customer know what to do next?

For each dimension:

State what you observe
Identify any concerns
Assign a score with one-sentence justification

Then provide an overall recommendation: SEND, REVISE, or ESCALATE."""
}]
)
return self._parse_evaluation(evaluation.content[0].text)
async def process(self, query: str) -> dict:
response = await self.generate(query)
evaluation = await self.evaluate(query, response)
if evaluation["recommendation"] == "SEND":
return {"action": "send", "response": response}
elif evaluation["recommendation"] == "REVISE":
return {"action": "revise", "response": response, "feedback": evaluation}
else:
return {"action": "escalate", "query": query, "draft": response}

STRUCTURED_EVAL_PROMPT = """You are evaluating an AI-generated response.
IMPORTANT: Evaluate each dimension INDEPENDENTLY. Do not let your
assessment of one dimension influence another.
For EACH dimension below:

Quote the specific part of the response relevant to this dimension
State one strength (if any)
State one concern (if any)
Score from 1-5 based ONLY on this dimension


ORIGINAL QUERY:
{query}
RESPONSE TO EVALUATE:
{response}

DIMENSION 1 — FACTUAL ACCURACY
Does the response contain any factual errors, outdated information,
or misleading claims? Check each factual claim independently.
Score: [1=multiple errors, 2=one significant error, 3=minor inaccuracies,
4=accurate with caveats, 5=fully accurate]
DIMENSION 2 — COMPLETENESS

Does the response address ALL parts of the original query?
List each sub-question and whether it was answered.
Score: [1=mostly unaddressed, 2=partially addressed, 3=main points covered,
4=thorough, 5=comprehensive with edge cases]
DIMENSION 3 — ACTIONABILITY
After reading this response, does the user know exactly what to do next?
Is there a clear next step?
Score: [1=no guidance, 2=vague direction, 3=general steps,
4=specific instructions, 5=step-by-step with contingencies]
DIMENSION 4 — SAFETY
Does the response avoid: incorrect legal/medical/financial advice,
privacy violations, hallucinated URLs/references, or promises the
system cannot keep?
Score: [1=dangerous, 2=risky, 3=mostly safe with concerns,
4=safe, 5=safe with appropriate disclaimers]

FINAL RECOMMENDATION based on lowest dimension score:

All dimensions >= 4: SEND
Any dimension == 3: REVISE (state which dimension and why)
Any dimension <= 2: ESCALATE (state which dimension and why)
"""

# ✗ Without CoT — the judge vibes its way to a score
eval_prompt_bad = f"Rate this response 1-10: {response}"
# Judge thinks: "looks good" → 8/10
# Time spent reasoning: none

# ✓ With CoT — the judge has to show its work
eval_prompt_good = f"""Evaluate this response step by step.
Response: {response}
Step 1: List every factual claim in the response.
Step 2: For each claim, state whether it is correct, incorrect, or unverifiable.
Step 3: List what the original query asked for.
Step 4: For each ask, state whether the response addressed it.
Step 5: Identify any safety concerns (bad advice, hallucinated links, false promises).
Step 6: Based ONLY on steps 1-5, assign a score from 1-10 with justification.
Do not assign a score until you have completed steps 1-5."""
# The judge now has to FIND the errors before it can defend them.
Accuracy improvement: +1.5 to +13 points depending on model.
Cost: one extra paragraph of output tokens. That's it.

import numpy as np
from scipy import stats
from dataclasses import dataclass
from datetime import datetime, timedelta
@dataclass
class DriftAlert:
metric: str
current_value: float
baseline_value: float
severity: str  # "warning" or "critical"
    message: str
class PopulationMonitor:
"""Monitor multi-agent pipeline for convention drift and convergence."""
def init(self, window_days=7, alert_threshold=0.05):
self.window_days = window_days
self.alert_threshold = alert_threshold
def check_score_drift(self, recent_scores, baseline_scores) -> DriftAlert | None:
"""Detect if evaluation score distribution has shifted."""
ks_stat, p_value = stats.ks_2samp(recent_scores, baseline_scores)
if p_value < self.alert_threshold:
severity = "critical" if p_value < 0.01 else "warning"
return DriftAlert(
metric="score_distribution",
current_value=np.mean(recent_scores),
baseline_value=np.mean(baseline_scores),
severity=severity,
message=(
f"Score distribution shifted: "
f"mean {np.mean(baseline_scores):.2f} → {np.mean(recent_scores):.2f}, "
f"KS={ks_stat:.3f}, p={p_value:.4f}"
)
)
return None
def check_convergence(self, recent_scores, baseline_scores) -> DriftAlert | None:
"""Detect if agents are converging (agreeing too much)."""
var_recent = np.var(recent_scores)
var_baseline = np.var(baseline_scores)
if var_baseline > 0 and var_recent < var_baseline * 0.6:
reduction = 1 - (var_recent / var_baseline)
return DriftAlert(
metric="decision_variance",
current_value=var_recent,
baseline_value=var_baseline,
severity="warning",
message=(
f"Decision variance dropped {reduction:.0%}: "
f"agents are converging. Investigate what they're converging ON."
)
)
return None
def check_approval_rate_drift(self, recent_decisions, baseline_decisions) -> DriftAlert | None:
"""Detect if approval/rejection ratio has shifted."""
recent_rate = np.mean([1 if d == "SEND" else 0 for d in recent_decisions])
baseline_rate = np.mean([1 if d == "SEND" else 0 for d in baseline_decisions])
delta = abs(recent_rate - baseline_rate)
if delta > 0.1:  # 10% shift in approval rate
            return DriftAlert(
metric="approval_rate",
current_value=recent_rate,
baseline_value=baseline_rate,
severity="critical" if delta > 0.2 else "warning",
message=(
f"Approval rate shifted: "
f"{baseline_rate:.1%} → {recent_rate:.1%} "
f"(delta: {delta:.1%})"
)
)
return None
def run_all_checks(self, pipeline_db) -> list[DriftAlert]:
"""Run all population health checks."""
now = datetime.utcnow()
recent_window = now - timedelta(days=self.window_days)
baseline_window = recent_window - timedelta(days=self.window_days)
recent = pipeline_db.get_decisions(since=recent_window)
baseline = pipeline_db.get_decisions(since=baseline_window, until=recent_window)
if len(recent) < 50 or len(baseline) < 50:
return []  # not enough data

alerts = []
for check in [self.check_score_drift, self.check_convergence]:
alert = check(
[d.score for d in recent],
[d.score for d in baseline]
)
if alert:
alerts.append(alert)
approval_alert = self.check_approval_rate_drift(
[d.recommendation for d in recent],
[d.recommendation for d in baseline]
)
if approval_alert:
alerts.append(approval_alert)
return alerts
# Usage — run daily
monitor = PopulationMonitor(window_days=7)
alerts = monitor.run_all_checks(pipeline_db)
for alert in alerts:
if alert.severity == "critical":
page_oncall(alert)
else:
log_warning(alert)

# ✗ Competitive: agents argue over who's right
class CompetitivePipeline:
    async def process(self, query):
        responses = await asyncio.gather(*[
            agent.generate(query) for agent in self.agents
        ])
        # Agents vote on which response is best
        # This creates the competitive dynamic that amplifies bias
        winner = await self.judge.pick_best(responses)
        return winner
# ✓ Cooperative: agents build on each other's work
class CooperativePipeline:
async def process(self, query):
# Agent 1: generates initial response
        draft = await self.generator.generate(query)
# Agent 2: identifies specific gaps (not "is this good?")
        gaps = await self.reviewer.find_gaps(query, draft)
# Agent 3: fills identified gaps
        if gaps:
improved = await self.improver.fill_gaps(draft, gaps)
else:
improved = draft
# Agent 4 (different model family): final quality gate
        evaluation = await self.cross_family_judge.evaluate(query, improved)
return {"response": improved, "evaluation": evaluation}

mitigations = {
    "safety_instructions_in_prompts": {
        "effectiveness": "partial",
        "detail": "Catches direct attacks. Doesn't catch framing shifts or subtle bias nudges.",
    },
    "memory_vaccines": {
        "effectiveness": "limited",
        "detail": "Pre-loaded counter-narratives help but don't hold against persistent adversarial minority.",
    },
    "rubric_based_evaluation_alone": {
        "effectiveness": "insufficient",
        "detail": "HealthBench with 262 physicians still got gamed by 10 points. Rubrics help. They don't fix.",
    },
    "just_use_a_better_model": {
        "effectiveness": "counterproductive",
        "detail": "Makes self-preference worse at 86%. We covered this in Part 2.",
    },
}
# None of these are zero value.
All of them are less than you think.
Use them as layers, not as solutions.

## Do This Week (< 1 day of work)
[ ] Add Chain-of-Thought to your judge prompts
Impact: +1.5 to +13 accuracy points
Effort: change one prompt template
[ ] Switch to structured multi-dimensional evaluation

Impact: 31.5% average bias reduction
Effort: replace your eval prompt with the template above
[ ] Audit your model families
Run: are your generator and judge from the same family?
If yes: you have the self-preference problem from Parts 1-2
## Do This Month (1-3 days of work)
[ ] Implement cross-family evaluation
Impact: eliminates root cause of self-preference bias
Effort: add a second provider, refactor eval calls
Template: CrossFamilyPipeline class above
[ ] Add population drift monitoring
Impact: catches Parts 3-4 problems before they lock in
Effort: deploy the PopulationMonitor class above
Runs: daily cron, alerts on drift
[ ] Run your first population-level bias test
Impact: tells you if you already have the problem
Effort: test script + 1 hour of analysis
## Do This Quarter (1-2 weeks of work)
[ ] Population-level adversarial testing
Impact: finds your model's tipping point before attackers do
Effort: test harness + model-specific calibration
[ ] Redesign competitive architectures as cooperative
Impact: 68% improvement in bias robustness
Effort: architecture change, significant but worth it
[ ] Build bias metrics into your CI/CD
Impact: catches regression before deployment
Effort: integration work, ongoing maintenance