MiroFish 深度实战：群体智能如何重构预测范式——从数字沙盘到"上帝视角"万物推演
前言：当预测不再是"猜数字"
"不预测个体，而是仿真群体。"
这是 MiroFish 项目的核心理念，也是我在深入研究这个项目后最深刻的体会。
传统预测工具的逻辑是什么？收集历史数据 → 训练统计模型 → 输出一个数字。这种方法在稳定环境下效果不错，但现实世界从来不是稳定的——舆论风向会突变、市场情绪会反转、政策影响会滞后涌现。线性回归能拟合一条曲线，但它永远无法解释"为什么这条曲线在这里突然拐弯了"。
MiroFish 换了一种完全不同的思路：不拟合，而是重演。
通过构建一个包含数千个 AI 智能体的虚拟社会，让这些"数字人"在一个高保真的平行世界中自由交互、演化、博弈。最终输出的不是"销量会是 100 万"这样的数字，而是一份完整的"群体行为演化报告"——告诉你在这个事件发生后，群体会如何反应、舆论会如何扩散、最终会走向什么结局。
这听起来像是科幻小说，但它已经是一个 GitHub 56k+ Stars 的开源项目了。
一、为什么需要群体智能预测？
1.1 传统预测的困境
让我们先来理解一下为什么传统预测方法越来越不够用了。
假设你是一家消费品公司的市场总监，公司正准备推出一款新产品，你需要预测：
- 新品上市后第一周的销量会是多少？
- 用户对产品的评价会呈现什么样的趋势？
- 竞争对手会如何应对？
用传统方法，你能做的大概是：
- 分析历史同类产品的销售曲线
- 参考市场调研机构的报告
- 做一个基于经验的线性外推
但问题是：你的新产品有一个竞争对手没有的重大技术创新。历史数据没有这种先例，调研报告也只反映当下认知。等你上市后用户实际体验了，一切才会真实展开——但那时候已经来不及调整了。
这就是"黑天鹅"困境的本质：我们习惯用过去预测未来，但真正改变格局的恰恰是那些"过去没发生过"的事。
1.2 群体行为的经济学真相
经济学中有一个概念叫"涌现"（Emergence）：当大量个体相互作用时，会产生远超个体行为简单加总的整体效应。
- 单个网民的情绪 ≠ 网络舆论的风向
- 单个投资者的决策 ≠ 市场的走向
- 单个用户的选择 ≠ 产品的口碑
舆论学家 Everett Rogers 在《创新的扩散》中指出，一个创新能否成功扩散，取决于采纳者的类型分布：创新者(2.5%) → 早期采纳者(13.5%) → 早期多数(34%) → 晚期多数(34%) → 落后者(16%)。这个规律的背后是复杂的社会网络效应。
所以，要预测群体行为，你不能只研究"平均人"，你必须研究"群体结构"本身。
1.3 MiroFish 的破局思路
MiroFish 的核心洞察是：与其预测数字，不如仿真过程。
它不试图告诉你"销量会是多少"，而是告诉你"在这个市场环境下，消费者会如何互动、口碑会如何传播、最终市场会形成什么样的均衡状态"。
这个思路的优势在于：
- 可解释性：你可以看到每一步推理过程，理解"为什么"
- 可干预性：你可以注入变量，做 what-if 分析
- 可探索性：你可以与模拟世界中的人物对话，深入挖掘
二、技术架构：五层核心组件深度剖析
2.1 整体架构概览
MiroFish 的技术架构可以分为五个核心层次：
┌─────────────────────────────────────────────────────────────┐
│ 用户交互层 (Web UI) │
├─────────────────────────────────────────────────────────────┤
│ API 网关层 (FastAPI) │
├─────────────────────────────────────────────────────────────┤
│ 仿真编排层 (Orchestrator) │
├──────────────┬──────────────┬──────────────┬─────────────────┤
│ 知识图谱层 │ 智能体层 │ 仿真引擎层 │ 报告生成层 │
│ (GraphRAG) │ (Agents) │ (OASIS) │ (ReportAgent) │
├──────────────┴──────────────┴──────────────┴─────────────────┤
│ 记忆存储层 (Zep Cloud) │
├─────────────────────────────────────────────────────────────┤
│ LLM 层 (OpenAI SDK Compatible) │
└─────────────────────────────────────────────────────────────┘
2.2 仿真引擎层：OASIS
MiroFish 的仿真引擎基于 CAMEL-AI 团队开源的 OASIS 框架。这是一个专为大规模多智能体仿真设计的引擎，核心能力包括：
百万级并发支持
OASIS 采用了高效的消息队列和异步调度机制，能够同时管理百万级别的智能体实例。这不是简单的"多线程"，而是一种特殊的"虚拟时间调度"——每个智能体有自己的"主观时间"，系统通过事件驱动来协调它们的交互。
通信协议设计
# OASIS 的核心通信协议简化版
class OASISMessage:
def __init__(
self,
sender_id: str, # 发送者 ID
receiver_id: str | None, # 接收者 ID (None = 广播)
content: dict, # 消息内容
timestamp: float, # 逻辑时间戳
message_type: str # 消息类型: SOCIAL/ACTION/EVENT
):
pass
class Agent:
def __init__(self, agent_id: str, personality: dict):
self.id = agent_id
self.personality = personality
self.beliefs = [] # 信念状态
self.memories = [] # 记忆序列
self.social_graph = {} # 社交关系图谱
def process_message(self, msg: OASISMessage) -> List[OASISMessage]:
"""处理接收到的消息，返回响应消息列表"""
pass
def step(self, sim_time: float) -> List[OASISMessage]:
"""单步执行：更新状态、产生行为、生成消息"""
pass
涌现行为机制
OASIS 的精妙之处在于，它不预设"宏观行为"，而是让宏观行为从微观交互中自然涌现。这依赖于三个关键设计：
- 有限理性：每个智能体的决策只基于局部信息，而非全局视图
- 社会影响：智能体会被其社交网络中的其他智能体影响
- 时序依赖：记忆会影响后续决策，形成路径依赖
2.3 知识图谱层：增强版 GraphRAG
GraphRAG（Graph Retrieval-Augmented Generation）是微软研究院 2024 年提出的技术，它结合了知识图谱和 RAG（检索增强生成），特别适合处理复杂的关系推理任务。
MiroFish 在此基础上做了两个关键增强：
时序图谱
标准 GraphRAG 关注的是"是什么"（实体和关系），但社会行为更重要的是"什么时候"和"先后顺序"。MiroFish 的时序图谱为每个节点和边增加了时间维度：
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime
@dataclass
class TemporalNode:
entity_id: str
entity_type: str # PERSON/ORG/EVENT/TOPIC
content: str
created_at: datetime
last_modified: datetime
# 时序增强字段
temporal_pattern: str # PERSISTENT/RECURRING/ONE-TIME
relevance_decay: float # 信息衰减系数
@dataclass
class TemporalEdge:
source_id: str
target_id: str
relation_type: str # KNOWS/SUPPORTS/OPPOSES/INFLUENCES
weight: float # 关系强度 [0, 1]
valid_from: datetime
valid_until: Optional[datetime]
causal_direction: bool # True = A导致B, False = 相关性
群体一致性约束
社会心理学研究表明，个体行为受"从众心理"强烈影响。MiroFish 通过图谱中的"一致性约束"来模拟这种效应：
class ConsistencyConstraint:
"""确保智能体的行为与其社交网络保持合理一致"""
def __init__(self, social_graph: nx.Graph, constraint_strength: float = 0.7):
self.graph = social_graph
self.strength = constraint_strength
def compute_opinion_shift(
self,
agent_id: str,
new_information: str
) -> float:
"""
计算在社交压力下，智能体观点的偏移量
"""
agent_neighbors = list(self.graph.neighbors(agent_id))
# 获取邻居的平均观点倾向
neighbor_opinions = [
self.graph.nodes[n].get('opinion_score', 0.5)
for n in agent_neighbors
]
avg_neighbor_opinion = sum(neighbor_opinions) / len(neighbor_opinions)
# 获取新信息暗示的观点
implied_opinion = self._extract_opinion_from(new_information)
# 在"坚持己见"和"从众"之间加权平均
original_opinion = self.graph.nodes[agent_id].get('opinion_score', 0.5)
shifted_opinion = (
original_opinion * (1 - self.strength) +
(0.3 * implied_opinion + 0.7 * avg_neighbor_opinion) * self.strength
)
return shifted_opinion
2.4 记忆存储层：Zep Cloud
Zep 是一个专为 AI 应用设计的记忆存储服务，在 MiroFish 中承担了两个关键角色：
智能体的"大脑"
每个智能体在 Zep 中有自己的记忆空间，包含：
- 事实记忆：直接观察到的事件（如"看到某条新闻"）
- 社交记忆：与其他智能体的交互历史
- 元认知记忆：对自身行为的反思（如"我觉得我之前判断错了"）
跨会话持久化
传统 AI 应用每次对话都是"失忆"的。Zep 通过向量数据库和图数据库的混合存储，让智能体的记忆能够跨会话持久化：
# Zep Cloud 的记忆写入示例
from zep_cloud import ZepClient
from zep_cloud.message import Message
client = ZepClient(api_key=ZEP_API_KEY)
# 为智能体创建一个新的记忆片段
memory_data = {
"event_type": "opinion_exposure",
"content": "在社交媒体上看到关于#新能源车#的讨论，正面评价居多",
"source_agent": "agent_0042",
"emotional_tone": "positive",
"influence_weight": 0.8
}
# Zep 自动处理向量化和图谱更新
await client.memory.add(
user_id="sim_world_user",
session_id="episode_2024_01",
messages=[
Message(
role="system",
content=json.dumps(memory_data),
metadata={"type": "agent_memory"}
)
]
)
2.5 智能体层：人格生成与角色分配
MiroFish 的每个智能体都不是"平均人"，而是拥有独特人格的个体。人格生成是通过 LLM 自动完成的：
from pydantic import BaseModel
from typing import List, Optional
class AgentPersonality(BaseModel):
"""智能体人格配置"""
# 基础人口属性
age_range: str # "25-34"
occupation: str # "软件工程师"
income_level: str # "中高收入"
location: str # "一线城市"
# 心理特征 (Big Five 模型简化)
openness: float # 开放性 [0, 1]
conscientiousness: float # 尽责性
extraversion: float # 外向性
agreeableness: float # 宜人性
neuroticism: float # 神经质
# 领域知识
domain_knowledge: List[str] # ["新能源汽车", "环保议题"]
# 行为倾向
risk_preference: float # 风险偏好 [0, 1]
social_influence: float # 社会影响力 [0, 1]
opinion_learning_rate: float # 观点学习速度
def to_llm_prompt(self) -> str:
"""转换为 LLM 可理解的角色描述"""
return f"""
你是一个{self.age_range}岁的{self.occupation}，住在{self.location}。
性格特征：
- 开放性（对新事物的接受度）：{"高" if self.openness > 0.6 else "中" if self.openness > 0.3 else "低"}
- 社交活跃度：{"活跃" if self.extraversion > 0.6 else "一般" if self.extraversion > 0.3 else "内敛"}
- 风险态度：{"冒险型" if self.risk_preference > 0.6 else "稳健型"}
你在{self.domain_knowledge}领域有一定的了解和关注。
"""
class ProfileGenerator:
"""使用 LLM 生成智能体人设"""
def __init__(self, llm_client):
self.llm = llm_client
async def generate_population(
self,
population_size: int,
demographic_distribution: dict
) -> List[AgentPersonality]:
"""
生成一个具有真实人口分布的智能体群体
demographic_distribution: {
"age_ranges": {"18-24": 0.2, "25-34": 0.4, ...},
"income_levels": {...},
...
}
"""
# 使用 LLM 批量生成人设
prompt = self._build_generation_prompt(
population_size, demographic_distribution
)
response = await self.llm.acomplete(prompt)
personalities = self._parse_response(response)
return personalities
三、核心流程：五阶段仿真流水线深度解析
MiroFish 的仿真流程分为五个阶段，每个阶段都有独特的技术挑战和解决方案。
3.1 第一阶段：种子提取 (Seed Extraction)
输入：原始材料（新闻报道、政策草案、产品说明书、小说章节等）
输出：结构化的知识图谱，包含实体、关系、事件
技术实现：
from typing import List, Dict, Any
import json
class SeedExtractor:
"""将原始材料转化为结构化知识"""
def __init__(self, llm_client, graph_builder):
self.llm = llm_client
self.graph = graph_builder
async def extract(self, seed_content: str) -> Dict[str, Any]:
"""
从种子内容中提取结构化信息
Returns:
{
"entities": [...],
"relations": [...],
"events": [...],
"key_topics": [...],
"stakeholders": [...]
}
"""
# 使用 LLM 进行信息抽取
extraction_prompt = f"""
你是一个专业的信息分析师。请从以下内容中提取结构化信息。
【输入内容】
{seed_content}
【提取要求】
1. 识别所有关键实体（人物、组织、事件、产品等）
2. 识别实体之间的关系（支持、对抗、影响等）
3. 识别关键事件及其时间线
4. 识别主要利益相关方及其立场
请以 JSON 格式输出：
{{
"entities": [
{{"name": "实体名", "type": "PERSON|ORG|EVENT|PRODUCT|TOPIC", "description": "描述", "importance": 0-1}}
],
"relations": [
{{"source": "实体A", "target": "实体B", "type": "关系类型", "strength": 0-1}}
],
"events": [
{{"name": "事件名", "description": "描述", "temporal_order": 1}}
],
"key_topics": ["话题1", "话题2"],
"stakeholders": [
{{"name": "利益方", "position": "立场描述", "influence_level": 0-1}}
]
}}
"""
response = await self.llm.acomplete(extraction_prompt)
result = json.loads(response)
# 写入图谱
await self._build_graph(result)
return result
async def _build_graph(self, extracted: Dict[str, Any]):
"""将提取结果写入图数据库"""
for entity in extracted.get("entities", []):
await self.graph.add_node(
entity["name"],
entity_type=entity["type"],
description=entity["description"],
importance=entity.get("importance", 0.5)
)
for relation in extracted.get("relations", []):
await self.graph.add_edge(
relation["source"],
relation["target"],
relation_type=relation["type"],
weight=relation.get("strength", 0.5)
)
关键设计考量：
- 多模态支持：不仅支持文本，还支持从图片、PDF 中提取关键信息
- 增量更新：支持多次输入的增量合并，而非每次重新生成
- 置信度评分：LLM 的抽取结果带有置信度，用于后续处理
3.2 第二阶段：世界构建 (World Building)
输入：结构化知识图谱
输出：完整的模拟世界配置，包括人口统计、初始状态、行为规则
技术实现：
class WorldBuilder:
"""构建高保真模拟世界"""
def __init__(
self,
population_generator: ProfileGenerator,
graph: ZepClient,
config: dict
):
self.population_gen = population_generator
self.graph = graph
self.config = config
async def build_world(
self,
extracted_knowledge: dict,
population_size: int = 1000
) -> SimulationWorld:
"""
构建完整的模拟世界
population_size: 智能体数量，默认 1000
"""
# Step 1: 确定人口分布
demographic_config = self._derive_demographics(extracted_knowledge)
# Step 2: 生成智能体群体
print(f"正在生成 {population_size} 个智能体人设...")
personalities = await self.population_gen.generate_population(
population_size=population_size,
demographic_distribution=demographic_config
)
# Step 3: 构建社交网络
print("正在构建社交网络...")
social_graph = await self._build_social_network(personalities)
# Step 4: 注入初始知识
print("正在注入初始知识...")
await self._inject_initial_knowledge(personalities, extracted_knowledge)
# Step 5: 设置环境参数
environment = self._configure_environment(extracted_knowledge)
return SimulationWorld(
agents=personalities,
social_graph=social_graph,
knowledge_graph=extracted_knowledge,
environment=environment
)
async def _build_social_network(
self,
agents: List[AgentPersonality]
) -> nx.scale_free_graph:
"""
构建无标度社交网络
真实社交网络往往呈现"富者愈富"的特点：
- 高影响力的人更容易被连接
- 社交圈层化明显
"""
import networkx as nx
import numpy as np
G = nx.Graph()
# 添加所有节点
for i, agent in enumerate(agents):
G.add_node(i, agent=agent)
# 基于影响力构建边（优先连接高影响力节点）
influence_scores = [a.social_influence for a in agents]
# 幂律分布：少数高影响力节点连接大量其他节点
for i, agent in enumerate(agents):
# 计算连接数量（影响力越高，连接越多）
base_connections = int(3 + agent.social_influence * 20)
target_connections = min(base_connections, len(agents) - 1)
# 优先选择高影响力的节点作为连接目标
for _ in range(target_connections):
# 按影响力加权概率选择连接目标
weights = np.array(influence_scores)
weights[i] = 0 # 不连接自己
weights = weights / weights.sum()
target = np.random.choice(len(agents), p=weights)
if not G.has_edge(i, target):
G.add_edge(i, target, weight=0.5)
return G
关键设计考量：
- 人口真实性：人口统计特征需要符合真实分布（如收入金字塔、年龄金字塔）
- 社交网络结构：采用无标度网络（Scale-free Network）模拟真实社交的"富者愈富"现象
- 领域知识注入：确保智能体对模拟领域有合理的基础认知
3.3 第三阶段：涌现演化 (Emergence Simulation)
输入：完整模拟世界
输出：动态演化过程数据（舆论传播、行为轨迹、关键事件）
OASIS 仿真核心：
class OASESimulator:
"""OASIS 仿真引擎封装"""
def __init__(
self,
world: SimulationWorld,
event_queue: asyncio.PriorityQueue,
config: dict
):
self.world = world
self.event_queue = event_queue
self.config = config
self.current_time = 0
self.event_log = []
async def run(self, simulation_steps: int = 100):
"""
运行仿真
simulation_steps: 仿真步数，每步代表一个逻辑时间单位
"""
print(f"开始仿真，共 {simulation_steps} 步...")
for step in range(simulation_steps):
self.current_time = step
# 每个时间步执行所有活跃智能体
messages = []
for agent_id, agent in enumerate(self.world.agents):
# 获取该智能体收到的消息
inbox = self._get_inbox(agent_id)
# 处理消息，更新状态
agent_messages = await self._process_agent_step(
agent_id, agent, inbox
)
messages.extend(agent_messages)
# 广播/路由消息
await self._dispatch_messages(messages)
# 检查并触发事件
await self._check_event_triggers()
# 记录日志（每隔 10 步打印进度）
if step % 10 == 0:
print(f" 进度: {step}/{simulation_steps}")
print("仿真完成！")
return self._compile_results()
async def _process_agent_step(
self,
agent_id: int,
agent: AgentPersonality,
inbox: List[dict]
) -> List[OASISMessage]:
"""单步处理一个智能体"""
# 1. 感知：收集环境信息
perception = self._gather_perception(agent_id)
# 2. 决策：基于人格和信息做决策
decision = await self._make_decision(agent, perception, inbox)
# 3. 行动：产生行为
action = self._take_action(agent_id, decision)
# 4. 表达：生成社交消息
messages = self._generate_messages(agent_id, action)
# 5. 记忆：更新记忆
await self._update_memory(agent_id, perception, action)
return messages
async def _make_decision(
self,
agent: AgentPersonality,
perception: dict,
inbox: List[dict]
) -> dict:
"""基于人格和信息做决策"""
# 构建决策上下文
context = self._build_decision_context(agent, perception, inbox)
# 使用 LLM 生成决策
decision_prompt = f"""
你是模拟世界中的一个虚拟角色。
【角色信息】
{agent.to_llm_prompt()}
【当前时间】
第 {self.current_time} 步
【你观察到的情况】
{json.dumps(perception, ensure_ascii=False, indent=2)}
【你收到的消息】
{json.dumps(inbox[:5], ensure_ascii=False, indent=2)}
【决策任务】
基于你的角色性格，决定你会采取什么行动。
请输出你的决策：
{{
"action_type": "POST_SOCIAL|MODIFY_OPINION|INTERACT|DO_NOTHING",
"action_details": {{...}},
"opinion_shift": -0.2 到 0.2 之间的值，表示观点变化
}}
行动要符合你的性格特点！
"""
response = await self.llm.acomplete(decision_prompt)
return json.loads(response)
涌现机制详解：
class EmergenceAnalyzer:
"""分析仿真过程中的涌现现象"""
def __init__(self, world: SimulationWorld):
self.world = world
def detect_opinion_clusters(self) -> List[OpinionCluster]:
"""
检测观点簇
在社交网络中，观点相近的个体会形成"回音室"
"""
import networkx as nx
from sklearn.cluster import SpectralClustering
# 构建观点相似度矩阵
similarity_matrix = self._build_similarity_matrix()
# 谱聚类识别观点簇
clustering = SpectralClustering(
n_clusters=5, # 假设有 5 个主要观点阵营
affinity='precomputed'
).fit_predict(similarity_matrix)
clusters = []
for i in range(5):
members = [j for j, c in enumerate(clustering) if c == i]
avg_opinion = np.mean([
self.world.agents[j].current_opinion
for j in members
])
clusters.append(OpinionCluster(
cluster_id=i,
member_count=len(members),
dominant_opinion=avg_opinion
))
return clusters
def detect_cascade_events(self) -> List[CascadeEvent]:
"""
检测级联传播事件
当某个观点或行为在短时间内大量传播时触发
"""
events = []
for topic in self.world.topics:
timeline = self._get_topic_timeline(topic)
# 检测"爆发"模式
spread_rate = np.diff(timeline) # 变化率
# 局部最大值 = 爆发点
peaks = np.where(
(spread_rate[1:-1] > spread_rate[:-2]) &
(spread_rate[1:-1] > spread_rate[2:])
)[0] + 1
for peak in peaks:
if spread_rate[peak] > self.world.cascade_threshold:
events.append(CascadeEvent(
topic=topic,
time=peak,
intensity=spread_rate[peak],
affected_agents=self._count_affected(peak)
))
return events
3.4 第四阶段：深度报告 (Deep Report Generation)
输入：仿真过程数据
输出：结构化预测报告 + 可交互的模拟世界
ReportAgent 实现：
class ReportAgent:
"""生成预测报告的专用智能体"""
def __init__(self, llm_client, simulation_data: dict):
self.llm = llm_client
self.data = simulation_data
async def generate_report(self) -> PredictionReport:
"""生成完整的预测报告"""
# 并行执行多个分析维度
analyses = await asyncio.gather(
self._analyze_opinion_evolution(),
self._analyze_social_dynamics(),
self._analyze_key_events(),
self._analyze_potential_outcomes()
)
opinion_analysis, social_analysis, event_analysis, outcome_analysis = analyses
# 综合生成报告
report_prompt = f"""
你是 MiroFish 的预测报告生成专家。
【仿真数据摘要】
- 模拟时长: {self.data['simulation_steps']} 步
- 智能体数量: {self.data['population_size']}
- 初始关注度: {self.data['initial_engagement']:.2%}
- 最终关注度: {self.data['final_engagement']:.2%}
【观点演化分析】
{opinion_analysis}
【社交动态分析】
{social_analysis}
【关键事件时间线】
{event_analysis}
【可能结果预测】
{outcome_analysis}
【报告要求】
请生成一份结构化的预测报告，包括：
1. 执行摘要（100字以内）
2. 核心发现（3-5个关键洞察）
3. 观点演化路径分析
4. 潜在风险与机会
5. What-If 情景建议
6. 结论与建议
报告要：
- 基于数据，有理有据
- 突出最重要的发现
- 提供可操作的建议
- 解释"为什么"而非仅仅"是什么"
"""
report_content = await self.llm.acomplete(report_prompt)
return PredictionReport(
content=report_content,
metrics=self._compute_summary_metrics(),
visualizations=self._generate_visualizations()
)
async def _analyze_opinion_evolution(self) -> str:
"""分析观点演化"""
prompt = f"""
分析以下观点演化数据，找出关键规律：
【观点分布变化】
初始: {self.data['initial_opinion_distribution']}
中期: {self.data['mid_opinion_distribution']}
最终: {self.data['final_opinion_distribution']}
【观点极化程度变化】
初始极化指数: {self.data['initial_polarization']:.3f}
最终极化指数: {self.data['final_polarization']:.3f}
请生成一段分析文字。
"""
return await self.llm.acomplete(prompt)
3.5 第五阶段：深度交互 (Deep Interaction)
输入：报告 + 模拟世界
输出：与模拟世界中任意角色的对话能力
这是 MiroFish 最"科幻"的功能：你可以直接与模拟世界中的任意智能体对话，就像在平行宇宙中与他们交流。
class InteractiveSimulation:
"""提供与模拟世界深度交互的能力"""
def __init__(self, world: SimulationWorld, llm_client):
self.world = world
self.llm = llm_client
async def chat_with_agent(
self,
agent_id: int,
user_message: str
) -> str:
"""
与模拟世界中的特定智能体对话
这是"上帝视角"与"当事人视角"的切换：
- 报告给你的是宏观视角
- 对话给你的是微观细节
"""
agent = self.world.agents[agent_id]
prompt = f"""
你是模拟世界中的一个角色。请回答下面的问题。
【你的角色】
{agent.to_llm_prompt()}
【你的记忆】
{self._get_agent_memory(agent_id)}
【你的当前观点】
{self._get_agent_current_opinion(agent_id)}
【最近经历】
{self._get_recent_experiences(agent_id)}
【用户问题】
{user_message}
请用第一人称回答，展现角色的真实感受和思考过程。
"""
return await self.llm.acomplete(prompt)
async def run_survey(
self,
agent_ids: List[int],
question: str
) -> SurveyResult:
"""
对一批智能体进行问卷调查
用于获取某个观点在特定群体中的分布
"""
responses = await asyncio.gather(
*[self._ask_single_agent(aid, question) for aid in agent_ids]
)
return SurveyResult(
question=question,
responses=responses,
distribution=self._compute_distribution(responses),
sentiment=self._compute_sentiment(responses)
)
四、生产级实战：从零构建一个舆情预测系统
4.1 环境准备
# 系统要求
# - Node.js 18+
# - Python 3.11+
# - Docker & Docker Compose
# - LLM API (OpenAI / 阿里百炼 / 等)
# 克隆项目
git clone https://github.com/666ghj/MiroFish
cd MiroFish
# 使用 Docker 快速启动
docker-compose up -d
# 或者本地开发模式
# 安装后端依赖
cd backend
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
pip install -r requirements.txt
# 安装前端依赖
cd ../frontend
npm install
4.2 配置文件
# 复制环境变量模板
cp backend/.env.example backend/.env
# 编辑 .env 文件
cat > backend/.env << 'EOF'
# LLM 配置 (支持 OpenAI SDK 兼容的任何 API)
LLM_PROVIDER=openai
OPENAI_API_KEY=sk-your-api-key
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_MODEL=gpt-4o
# 或者使用阿里百炼
# LLM_PROVIDER=openai
# OPENAI_API_KEY=your-api-key
# OPENAI_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
# OPENAI_MODEL=qwen-plus
# Zep Cloud 记忆服务 (可选但推荐)
ZEP_API_KEY=your-zep-api-key
# 服务器配置
HOST=0.0.0.0
PORT=8000
EOF
4.3 发起预测任务
import requests
import json
import time
API_BASE = "http://localhost:8000/api"
def create_prediction_task(seed_content: str, config: dict):
"""创建预测任务"""
response = requests.post(
f"{API_BASE}/tasks",
json={
"seed_content": seed_content,
"population_size": config.get("population_size", 1000),
"simulation_steps": config.get("simulation_steps", 100),
"model": config.get("model", "qwen-plus")
}
)
task_id = response.json()["task_id"]
print(f"任务已创建: {task_id}")
return task_id
def wait_for_completion(task_id: str, poll_interval: int = 10):
"""等待任务完成"""
while True:
response = requests.get(f"{API_BASE}/tasks/{task_id}")
status = response.json()
print(f"状态: {status['status']}")
if status["status"] == "completed":
return status["result"]
elif status["status"] == "failed":
raise Exception(f"任务失败: {status['error']}")
time.sleep(poll_interval)
def main():
# 种子内容：一篇关于某地出台新房产政策的新闻
seed_content = """
【突发】某一线城市今日发布房产新政：
1. 首付比例从 30% 降至 20%
2. 取消限购令，外地户籍可直接购房
3. 贷款利率下调 15 个基点
4. 推出"以旧换新"补贴政策，最高补贴 10 万元
业内人士分析，此举旨在刺激刚需购房需求，
但也有人担心会引发新一轮投机潮。
"""
# 创建任务
task_id = create_prediction_task(seed_content, {
"population_size": 2000, # 2000 个智能体
"simulation_steps": 200, # 200 步仿真
})
# 等待完成
print("正在仿真，请耐心等待...")
result = wait_for_completion(task_id)
# 输出报告
print("\n" + "=" * 60)
print("预测报告")
print("=" * 60)
print(result["report"])
# 保存结果
with open("prediction_result.json", "w", encoding="utf-8") as f:
json.dump(result, f, ensure_ascii=False, indent=2)
if __name__ == "__main__":
main()
4.4 案例：预测某科技产品发布会的市场反应
让我们用一个实际案例来演示完整流程：
# 种子内容：某科技公司即将发布革命性 AI 产品的预告
seed_content = """
【独家】据内部消息，某科技巨头将于下月发布一款"改变游戏规则"的 AI 产品。
知情人士透露，这款产品被内部称为"Project Phoenix"，
主打"全模态理解"和"实时个性化"两大特性。
产品亮点：
- 支持文本、图像、视频、音频的实时融合理解
- 能够根据用户习惯实时调整交互策略
- 内置"记忆胶囊"功能，实现跨会话的个性化体验
市场分析师对此反应不一：
- 乐观派：这是继 ChatGPT 之后最大的技术突破
- 保守派：技术成熟度存疑，落地能力待验证
- 竞争派：其他巨头可能很快会推出类似产品
该公司 CEO 在社交媒体上发文称："准备好迎接惊喜吧。"
"""
task_id = create_prediction_task(seed_content, {
"population_size": 3000,
"simulation_steps": 300,
})
运行后，系统会输出类似这样的预测报告：
# 市场反应预测报告
## 执行摘要
Project Phoenix 的发布预计将引发市场强烈关注，但最终影响取决于产品实际表现和竞争对手的应对策略。
## 核心发现
### 1. 初期关注度将爆发
- 预测发布会后 48 小时内，社交媒体讨论量将增长 300-500%
- "全模态"概念将获得最多关注（预估 45% 的讨论聚焦于此）
### 2. 两周内将出现明显的阵营分化
- **支持派**（约 35%）：技术乐观主义者，倾向于早期尝试
- **观望派**（约 40%）：等待实际评测后再做判断
- **质疑派**（约 25%）：对数据隐私和AI伦理表示担忧
### 3. 竞争对手将在 3-6 个月内做出反应
- 预测主要竞品会加速类似产品的研发
- 可能出现"贴脸竞争"式的营销策略
### 4. 长期影响取决于生态系统建设
- 如果能快速建立开发者生态，产品将持续领跑
- 如果生态建设滞后，可能被后来者超越
## 观点演化路径
[图表：舆论热度变化曲线]
## What-If 分析
| 情景 | 概率 | 可能走向 |
|------|------|----------|
| 产品超出预期 | 30% | 股价大涨，市场份额快速提升 |
| 产品符合预期 | 45% | 稳步增长，观望者逐步转化 |
| 产品低于预期 | 25% | 舆论反转，股价承压 |
五、性能优化与最佳实践
5.1 仿真规模与资源消耗
MiroFish 的资源消耗与仿真规模呈线性关系：
| 规模 | 智能体数 | 典型内存 | 典型耗时 | 适用场景 |
|---|---|---|---|---|
| 小规模 | 100-500 | 2-4 GB | 10-30 分钟 | 快速验证 |
| 中规模 | 1000-2000 | 8-16 GB | 1-3 小时 | 正式分析 |
| 大规模 | 5000+ | 32+ GB | 半天+ | 学术研究 |
5.2 LLM 调用优化
LLM 是最大的成本和时间瓶颈。以下是优化策略：
class OptimizedLLMCaller:
"""优化 LLM 调用的策略"""
def __init__(self, base_client):
self.client = base_client
async def batch_complete(
self,
prompts: List[str],
batch_size: int = 10,
max_retries: int = 3
):
"""
批量完成，利用并发减少总耗时
注意：不要一次性提交太多请求，可能触发 API 限流
"""
results = []
for i in range(0, len(prompts), batch_size):
batch = prompts[i:i + batch_size]
# 并发执行这一批
batch_results = await asyncio.gather(
*[self._call_with_retry(p, max_retries) for p in batch],
return_exceptions=True
)
results.extend(batch_results)
# 避免触发限流
if i + batch_size < len(prompts):
await asyncio.sleep(1)
return results
async def _call_with_retry(
self,
prompt: str,
max_retries: int
) -> str:
"""带重试的调用"""
for attempt in range(max_retries):
try:
return await self.client.acomplete(prompt)
except RateLimitError:
# 指数退避
wait_time = 2 ** attempt
await asyncio.sleep(wait_time)
except Exception as e:
if attempt == max_retries - 1:
raise
await asyncio.sleep(1)
return "" # 不太可能到这里
5.3 缓存策略
from functools import lru_cache
import hashlib
class CachedProfileGenerator:
"""带缓存的人设生成器"""
def __init__(self, base_generator):
self.base = base_generator
self.cache = {}
def _cache_key(self, seed: str, index: int) -> str:
"""生成缓存键"""
content = f"{seed}:{index}"
return hashlib.md5(content.encode()).hexdigest()
async def generate_personality(
self,
seed: str,
index: int,
constraints: dict
) -> AgentPersonality:
"""带缓存的生成"""
key = self._cache_key(seed, index)
if key in self.cache:
return self.cache[key]
result = await self.base.generate(seed, index, constraints)
self.cache[key] = result
return result
5.4 生产部署建议
# docker-compose.yml 优化版本
version: '3.8'
services:
backend:
build: ./backend
environment:
- LLM_BATCH_SIZE=5 # 减小并发量
- ENABLE_CACHING=true
- MAX_WORKERS=4
deploy:
resources:
limits:
memory: 16G
reservations:
devices:
- driver: nvidia
count: 0 # 如有 GPU 可设为 1
redis:
image: redis:7-alpine
# 用于缓存 LLM 响应
command: redis-server --maxmemory 1gb --maxmemory-policy allkeys-lru
postgres:
image: postgres:15
volumes:
- postgres_data:/var/lib/postgresql/data
六、局限性与未来展望
6.1 当前局限性
MiroFish 不是一个完美的预测工具，以下局限性需要清醒认识：
1. 现实扭曲假设
仿真假设"人会理性地根据接收到的信息更新观点"，但现实中存在大量非理性因素：
- 确认偏误：人们倾向于接受支持已有观点的信息
- 情绪传染：恐惧、愤怒等情绪往往比理性更能驱动行为
- 信息茧房：算法推荐会强化极化效应
2. 涌现的不确定性
复杂系统的涌现行为本质上不可预测。仿真只能告诉你"可能出现什么"，不能保证"一定会出现什么"。
3. 初始条件的敏感性
"蝴蝶效应"意味着初始条件的微小变化可能导致完全不同的结果。MiroFish 多次运行同一任务可能得到不同结果。
6.2 未来发展方向
多模态仿真
目前的仿真主要基于文本，未来可能扩展到图像、视频等模态，更真实地模拟信息在社交媒体上的传播。
实时数据接入
与 Twitter/X、微博等平台的实时数据对接，让仿真能够基于最新动态实时调整预测。
跨领域扩展
从社会舆情扩展到金融市场、生物进化、生态系统等领域。
结语：预测的本质是理解
MiroFish 给我最大的启发，不是它能做出多准确的预测，而是它改变了一种思维方式：
从"预测数字"到"理解过程"。
当我们试图预测"销量会是多少"时，我们是在猜测一个结果。但如果我们能理解"消费者会如何互动、舆论会如何演化、最终会形成什么样的均衡"，我们得到的不仅是预测本身，而是对整个系统的深层理解。
这种理解的价值远超预测本身——它让你能够干预、引导、优化，而不仅仅是"预知"。
就像 MiroFish 的项目名一样：Miro（镜子）+ Fish（鱼群）= 用数字沙盘照见现实世界的运行规律。
这不是预测未来，而是创造理解未来的能力。
附录：关键资源
- GitHub: https://github.com/666ghj/MiroFish
- 相关项目: BettaFish (情感数据采集) - https://github.com/666ghj/BettaFish
- OASIS 框架: https://github.com/camel-ai/oasis
- CAMEL-AI: https://github.com/camel-ai/camel
- Zep Cloud: https://www.getzep.com/
本文首发于程序员茄子 (chenxutan.com)，如需转载，请注明出处。