团队决策总达不成一致？用 AI 协商系统实现自动化公平调解
在分布式团队和多利益相关方的项目中，达成共识往往是最耗时的环节。代码审查意见分歧、产品优先级争论、资源分配冲突——这些场景每天都在发生。传统的解决方案要么依赖冗长的会议，要么由管理者强行拍板，但两种方式都难以保证公平性和透明度。
最近，Mediator.ai 这个项目引起了我的注意。它使用纳什博弈理论和 LLM 结合，系统化地实现公平协商。这篇文章将深入解析这类 AI 协商系统的工作原理，并带你从零开始构建一个实用的团队决策调解工具。
为什么需要 AI 协商系统？
在开始编码之前，先理解问题的本质。传统决策流程的痛点：
- 信息不对称：各方掌握的信息不同，难以站在同一视角讨论
- 情绪干扰：人际冲突往往掩盖了实际问题
- 权力失衡：资深成员或管理者的意见权重过大
- 缺乏追溯：决策过程没有记录，事后难以复盘
AI 协商系统的核心价值在于：将主观的”谁说了算”转化为客观的”什么方案最优”。
核心架构设计
一个完整的 AI 协商系统包含四个关键组件：
┌─────────────────────────────────────────────────────────┐ │ AI 协商系统架构 │ ├─────────────────────────────────────────────────────────┤ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ 提案收集器 │ │ 利益分析器 │ │ 公平性评估 │ │ │ │ Proposal │ │ Interest │ │ Fairness │ │ │ │ Collector │ │ Analyzer │ │ Evaluator │ │ │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ │ │ │ │ │ │ └─────────────────┼─────────────────┘ │ │ ▼ │ │ ┌─────────────────┐ │ │ │ LLM 调解引擎 │ │ │ │ Mediation LLM │ │ │ │ + Nash Solver │ │ │ └────────┬────────┘ │ │ │ │ │ ▼ │ │ ┌─────────────────┐ │ │ │ 共识方案输出 │ │ │ │ Consensus Plan │ │ │ └─────────────────┘ │ └─────────────────────────────────────────────────────────┘
1. 提案收集器
负责结构化地收集各方的立场和论据：
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime
@dataclass
class Stakeholder:
"""利益相关方"""
id: str
name: str
role: str
weight: float = 1.0 # 决策权重，默认平等
@dataclass
class Proposal:
"""提案"""
id: str
stakeholder_id: str
title: str
description: str
pros: List[str] # 优点
cons: List[str] # 缺点
priority: int # 1-5 优先级
created_at: datetime = None
def __post_init__(self):
if self.created_at is None:
self.created_at = datetime.now()
2. 利益分析器（LLM 驱动）
使用 LLM 分析各方提案背后的真实利益诉求，而非表面立场：
import json
from openai import OpenAI
class InterestAnalyzer:
"""利益分析器 - 识别提案背后的真实诉求"""
def __init__(self, api_key: str, model: str = "gpt-4o"):
self.client = OpenAI(api_key=api_key)
self.model = model
def analyze_interests(self, proposals: List[Proposal]) -> dict:
"""分析所有提案，识别共同利益和冲突点"""
prompt = self._build_analysis_prompt(proposals)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": self._get_system_prompt()},
{"role": "user", "content": prompt}
],
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
def _get_system_prompt(self) -> str:
return """你是一个专业的谈判调解专家，擅长使用哈佛谈判原则识别各方的真实利益诉求。
请分析以下提案，输出：
1. common_interests: 各方共同的利益点（至少 3 个）
2. conflicts: 核心冲突点及原因
3. hidden_concerns: 各方未明说的潜在担忧
4. trade_off_opportunities: 可能的交换/妥协空间
使用中文输出，保持客观中立。"""
def _build_analysis_prompt(self, proposals: List[Proposal]) -> str:
lines = ["以下是各方提交的提案：\n"]
for i, p in enumerate(proposals, 1):
lines.append(f"""
提案{i}: {p.title}
提交者：{p.stakeholder_id}
描述：{p.description}
优点：{', '.join(p.pros)}
缺点：{', '.join(p.cons)}
优先级：{p.priority}/5
---
""")
return '\n'.join(lines)
3. 纳什博弈求解器
将协商问题转化为纳什议价模型，计算最优公平解：
import numpy as np
from typing import Tuple
class NashBargainingSolver:
"""纳什议价求解器 - 计算公平的最优解"""
def __init__(self, disagreement_point: Tuple[float, float] = (0, 0)):
"""
Args:
disagreement_point: 谈判破裂时的收益点 (d1, d2)
"""
self.d = np.array(disagreement_point)
def solve(self, utility_possibility_set: List[Tuple[float, float]]) -> Tuple[float, float]:
"""
求解纳什议价解
纳什议价解最大化：(u1 - d1) * (u2 - d2)
其中 u 是效用，d 是破裂点
Args:
utility_possibility_set: 所有可能的效用组合 [(u1, u2), ...]
Returns:
最优效用组合 (u1*, u2*)
"""
best_solution = None
best_nash_product = float('-inf')
for u in utility_possibility_set:
u_array = np.array(u)
# 只有当双方收益都大于破裂点时才可行
if np.all(u_array > self.d):
nash_product = np.prod(u_array - self.d)
if nash_product > best_nash_product:
best_nash_product = nash_product
best_solution = u
return best_solution if best_solution else self.d
def calculate_utility(self, proposal: Proposal, stakeholder: Stakeholder) -> float:
"""
计算某提案对特定利益方的效用值
效用 = 优先级权重 × 提案质量分 × 利益相关方权重
"""
priority_weight = proposal.priority / 5.0
# 简化：用 pros/cons 数量比作为质量分
quality_score = len(proposal.pros) / (len(proposal.pros) + len(proposal.cons) + 1)
return priority_weight * quality_score * stakeholder.weight
4. LLM 调解引擎
整合分析结果，生成可执行的共识方案：
class MediationEngine:
"""AI 调解引擎 - 生成最终共识方案"""
def __init__(self, api_key: str, model: str = "claude-sonnet-4-5-20260514"):
from anthropic import Anthropic
self.client = Anthropic(api_key=api_key)
self.model = model
def generate_consensus(
self,
proposals: List[Proposal],
interest_analysis: dict,
nash_solution: Tuple[float, float]
) -> str:
"""生成最终调解方案"""
prompt = self._build_mediation_prompt(proposals, interest_analysis, nash_solution)
message = self.client.messages.create(
model=self.model,
max_tokens=2000,
system=self._get_system_prompt(),
messages=[{"role": "user", "content": prompt}]
)
return message.content[0].text
def _get_system_prompt(self) -> str:
return """你是一个专业的 AI 调解员，你的任务是：
1. 基于各方提案和利益分析，生成一个公平的共识方案
2. 方案必须具体可执行，包含明确的行动项和责任人
3. 解释为什么这个方案是公平的（引用纳什议价结果）
4. 语气中立、专业，避免偏袒任何一方
输出格式：
## 共识方案
### 核心决策
### 执行计划
### 公平性说明
### 后续跟进"""
def _build_mediation_prompt(
self,
proposals: List[Proposal],
analysis: dict,
nash_solution: Tuple[float, float]
) -> str:
return f"""
【利益分析结果】
共同利益：{', '.join(analysis.get('common_interests', []))}
核心冲突：{analysis.get('conflicts', '无')}
潜在担忧：{analysis.get('hidden_concerns', '无')}
妥协空间：{analysis.get('trade_off_opportunities', '无')}
【纳什议价结果】
最优效用组合：方 1={nash_solution[0]:.3f}, 方 2={nash_solution[1]:.3f}
【原始提案】
{chr(10).join([f'- {p.title} (提交者：{p.stakeholder_id}, 优先级：{p.priority}/5)' for p in proposals])}
请基于以上信息生成调解方案。"""
完整工作流示例
将以上组件整合为一个完整的协商流程：
async def run_mediation_session(
topic: str,
stakeholders: List[Stakeholder],
proposals: List[Proposal],
openai_key: str,
anthropic_key: str
) -> str:
"""运行一次完整的 AI 调解会话"""
# 1. 利益分析
analyzer = InterestAnalyzer(api_key=openai_key)
interest_analysis = analyzer.analyze_interests(proposals)
print("📊 利益分析完成")
print(f" 共同利益：{len(interest_analysis.get('common_interests', []))} 个")
print(f" 核心冲突：{len(interest_analysis.get('conflicts', []))} 个")
# 2. 计算效用矩阵
solver = NashBargainingSolver()
utility_set = []
for p in proposals:
utilities = [solver.calculate_utility(p, s) for s in stakeholders]
# 简化为两方场景演示
if len(utilities) >= 2:
utility_set.append((utilities[0], utilities[1]))
nash_solution = solver.solve(utility_set)
print(f"🎯 纳什议价解：({nash_solution[0]:.3f}, {nash_solution[1]:.3f})")
# 3. 生成调解方案
mediator = MediationEngine(api_key=anthropic_key)
consensus = mediator.generate_consensus(proposals, interest_analysis, nash_solution)
return consensus
# 使用示例
if __name__ == "__main__":
stakeholders = [
Stakeholder(id="dev_lead", name="开发负责人", role="technical"),
Stakeholder(id="pm", name="产品经理", role="business"),
]
proposals = [
Proposal(
id="p1",
stakeholder_id="dev_lead",
title="先重构再开发新功能",
description="当前技术债务严重，建议花 2 周重构核心模块",
pros=["降低维护成本", "提高开发效率", "减少 Bug"],
cons=["新功能延迟", "短期无用户可见价值"],
priority=4
),
Proposal(
id="p2",
stakeholder_id="pm",
title="优先交付客户承诺的功能",
description="Q2 有重要客户交付，建议先完成功能开发",
pros=["满足客户期望", "保证收入", "维护信誉"],
cons=["技术债务累积", "长期维护成本高"],
priority=5
),
]
result = asyncio.run(run_mediation_session(
topic="技术重构 vs 功能交付优先级",
stakeholders=stakeholders,
proposals=proposals,
openai_key="your-openai-key",
anthropic_key="your-anthropic-key"
))
print("\n" + "="*60)
print("📋 AI 调解方案")
print("="*60)
print(result)
实际输出示例
运行上述代码后，你可能得到类似这样的调解方案：
## 共识方案 ### 核心决策 采用**并行双轨制**：分配 60% 资源交付客户功能，40% 资源进行关键模块重构。 ### 执行计划 1. **第 1-2 周**： - 开发团队：完成客户功能的核心框架（60% 人力） - 架构小组：重构认证和日志模块（40% 人力） 2. **第 3-4 周**： - 开发团队：完成客户功能的全部开发 - 架构小组：重构数据库访问层 3. **第 5 周**： - 统一测试和集成 - 客户功能交付 + 技术债务清理报告 ### 公平性说明 根据纳什议价模型，此方案使双方效用最大化： - 技术方：关键模块得到重构，效用值 0.72 - 业务方：客户功能按时交付，效用值 0.81 - 纳什积：0.58（高于单一方案的 0.45） ### 后续跟进 - 每周五下午进行进度同步 - 使用共享看板追踪双轨进度 - 如有偏差，触发重新调解流程
进阶优化方向
1. 多轮协商支持
真实的协商往往需要多轮迭代。可以添加反馈循环：
class IterativeMediator:
"""多轮迭代调解器"""
def __init__(self, max_rounds: int = 3):
self.max_rounds = max_rounds
def mediate(self, initial_proposals: List[Proposal]) -> str:
current_proposals = initial_proposals
for round_num in range(1, self.max_rounds + 1):
consensus = run_mediation_session(..., current_proposals)
# 收集各方对共识方案的反馈
feedback = self.collect_feedback(consensus)
if self.is_consensus_reached(feedback):
return consensus
# 根据反馈调整提案
current_proposals = self.revise_proposals(current_proposals, feedback)
return self.force_decision(current_proposals)
2. 历史决策学习
记录历史调解案例，让系统越来越懂你的团队：
class DecisionHistory:
"""决策历史学习"""
def __init__(self, db_path: str = "decisions.db"):
self.db_path = db_path
def record_decision(self, topic: str, consensus: str, outcome: str):
"""记录决策及后续结果"""
# 存储到数据库，用于后续分析
def find_similar_cases(self, current_topic: str) -> List[dict]:
"""查找相似历史案例"""
# 使用向量相似度搜索
def get_team_bias_profile(self) -> dict:
"""分析团队决策偏好"""
# 例如：技术优先 vs 业务优先的倾向
3. 集成到现有工具链
将调解系统集成到 Slack、飞书、Jira 等工具：
# Slack Bot 集成示例
from slack_bolt import App
app = App(token="xoxb-your-token")
@app.command("/mediate")
def handle_mediate_command(ack, command, say):
ack()
topic = command["text"]
# 创建调解会话
session_id = create_mediation_session(topic)
say(f"🤖 已创建调解会话：{topic}")
say(f"请使用 /submit-proposal 提交你的方案")
say(f"会话 ID: `{session_id}`")
注意事项与局限性
适用场景
- ✅ 技术路线选择争议
- ✅ 资源分配冲突
- ✅ 产品优先级讨论
- ✅ 跨团队协调整合
不适用场景
- ❌ 涉及法律合规的决策（需要人工审核）
- ❌ 高度情绪化的人际冲突（需要 HR 介入）
- ❌ 紧急危机处理（需要快速决策）
- ❌ 涉及机密信息的讨论（注意数据隐私）
伦理考量
- 透明度：向所有参与者说明 AI 如何做出建议
- 可解释性：确保调解方案的推理过程可追溯
- 人工否决权：保留人类最终决策权
- 偏见检测：定期审计算法是否存在系统性偏见
总结
AI 协商系统不是要取代人类决策，而是提供一个结构化、透明、可追溯的决策框架。通过结合 LLM 的理解能力和博弈论的数学严谨性，我们可以在保持人性化的同时，大幅提升团队决策的效率和质量。
Mediator.ai 这类工具的出现，标志着 AI 正在从”执行者”向”协调者”演进。对于开发者而言，理解并掌握这类系统的设计原理，将是在 AI 时代构建高效协作团队的关键技能。
参考资料：