MiroFish 学习教程
MiroFish 学习教程
从零开始掌握 MiroFish 群体智能预测引擎
教程日期:2026-03-17
目录
一、环境准备
1.1 前置知识
学习 MiroFish 前建议了解:
| 知识领域 | 重要程度 | 说明 |
|---|---|---|
| Python 编程 | ⭐⭐⭐⭐⭐ | 后端核心语言 |
| JavaScript/React | ⭐⭐⭐⭐ | 前端界面开发 |
| LLM 基础 | ⭐⭐⭐⭐ | 理解 Agent 行为 |
| 图数据库 | ⭐⭐⭐ | GraphRAG 知识存储 |
| 社交网络分析 | ⭐⭐⭐ | 理解模拟场景 |
1.2 系统要求
硬件要求:
├── CPU: 8 核以上 (推荐 16 核)
├── 内存: 32GB 以上 (推荐 64GB)
├── 存储: 100GB SSD
└── GPU: 可选 (加速 LLM 推理)
软件要求:
├── 操作系统: Linux / macOS
├── Docker: 用于快速部署
├── Node.js: 18+
└── Python: 3.11-3.12
1.3 安装步骤
方式一:Docker 部署(推荐)
# 1. 克隆仓库
git clone https://github.com/666ghj/MiroFish.git
cd MiroFish
# 2. 配置环境变量
cp .env.example .env
# 编辑 .env,填入必要配置
# 3. 启动所有服务
docker-compose up -d
# 4. 查看日志
docker-compose logs -f
方式二:手动安装
# 1. 安装 Python 包管理器 uv
curl -LsSf https://astral.sh/uv/install.sh | sh
# 2. 克隆仓库
git clone https://github.com/666ghj/MiroFish.git
cd MiroFish
# 3. 安装后端依赖
cd backend
uv venv
source .venv/bin/activate # Linux/macOS
uv pip install -r requirements.txt
# 4. 安装前端依赖
cd ../frontend
npm install
# 5. 配置环境变量
cp ../.env.example ../.env
# 编辑 .env 文件
1.4 环境变量配置
# .env 文件示例
# ===== LLM 配置 =====
OPENAI_API_KEY=sk-xxx
OPENAI_BASE_URL=https://api.openai.com/v1 # 可选,用于代理
# ===== 数据库配置 =====
REDIS_URL=redis://localhost:6379/0
MONGODB_URL=mongodb://localhost:27017/mirofish
NEO4J_URL=bolt://localhost:7687
NEO4J_USER=neo4j
NEO4J_PASSWORD=password
# ===== 服务配置 =====
API_HOST=0.0.0.0
API_PORT=8000
FRONTEND_PORT=3000
# ===== 模拟配置 =====
MAX_AGENTS=10000
DEFAULT_TIME_STEP=3600
1.5 验证安装
# 检查后端服务
curl http://localhost:8000/health
# 预期输出
{"status": "healthy", "version": "1.0.0"}
# 检查前端服务
curl http://localhost:3000
# 预期输出
<!DOCTYPE html>...
二、快速开始
2.1 Hello World:创建第一个模拟
# examples/01_hello_world.py
from mirofish import Simulation, Agent, Environment
# 1. 创建环境
env = Environment(
name="hello_world",
platform="weibo" # 模拟微博平台
)
# 2. 创建 Agent
agents = []
for i in range(10):
agent = Agent(
id=f"agent_{i}",
profile={
"name": f"用户{i}",
"age": 20 + i * 3,
"interests": ["科技", "游戏"]
}
)
agents.append(agent)
# 3. 添加 Agent 到环境
env.add_agents(agents)
# 4. 创建模拟
sim = Simulation(
environment=env,
duration=7, # 模拟 7 天
time_step=3600 # 每小时一步
)
# 5. 注入初始事件
sim.inject_event(
content="今天发布了一款革命性的 AI 产品!",
source="agent_0"
)
# 6. 运行模拟
results = sim.run()
# 7. 查看结果
print(f"总互动数: {results.total_interactions}")
print(f"传播范围: {results.reach}")
print(f"情感分布: {results.sentiment}")
2.2 通过 Web 界面使用
- 访问
http://localhost:3000 - 点击「新建模拟」
- 配置模拟参数
- 点击「开始模拟」
- 查看实时可视化结果
三、核心概念
3.1 概念图谱
┌─────────────────┐
│ Simulation │
│ (模拟) │
└────────┬────────┘
│
┌────────────────────┼────────────────────┐
│ │ │
▼ ▼ ▼
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ Environment │ │ Agent │ │ Event │
│ (环境) │ │ (智能体) │ │ (事件) │
└───────────────┘ └───────────────┘ └───────────────┘
│ │ │
│ ┌───────┴───────┐ │
│ │ │ │
▼ ▼ ▼ ▼
┌───────────────┐ ┌───────┐ ┌───────────┐ ┌───────────┐
│ Platform │ │Profile│ │Personality│ │ Content │
│ (平台) │ │ (画像) │ │ (人格) │ │ (内容) │
└───────────────┘ └───────┘ └───────────┘ └───────────┘
3.2 核心概念详解
Simulation (模拟)
模拟是 MiroFish 的顶层概念,代表一次完整的模拟实验。
class Simulation:
"""
模拟配置:
- environment: 模拟环境
- duration: 模拟时长(虚拟天数)
- time_step: 时间步长(秒)
- seed: 随机种子(可复现)
"""
def __init__(
self,
environment: Environment,
duration: int = 30,
time_step: int = 3600,
seed: int = None
):
pass
def run(self) -> SimulationResult:
"""运行模拟,返回结果"""
pass
def pause(self):
"""暂停模拟"""
pass
def resume(self):
"""恢复模拟"""
pass
Environment (环境)
环境定义了 Agent 生活的数字世界。
class Environment:
"""
环境属性:
- name: 环境名称
- platform: 平台类型 (weibo/twitter/reddit)
- agents: Agent 列表
- social_graph: 社交图谱
"""
def __init__(
self,
name: str,
platform: str = "weibo"
):
pass
def add_agents(self, agents: List[Agent]):
"""添加 Agent 到环境"""
pass
def get_feed(self, agent_id: str) -> List[Content]:
"""获取 Agent 的信息流"""
pass
Agent (智能体)
Agent 是模拟中的个体,具有独立的人格和行为。
class Agent:
"""
Agent 组成:
- id: 唯一标识
- profile: 人口统计画像
- personality: 人格特征 (大五人格)
- memory: 记忆系统 (GraphRAG)
- state: 当前状态
"""
def __init__(
self,
id: str,
profile: dict,
personality: dict = None,
memory: Memory = None
):
pass
def observe(self, environment) -> Observation:
"""观察环境,获取信息"""
pass
def decide(self, observation) -> Decision:
"""基于观察做出决策"""
pass
def act(self, decision) -> Action:
"""执行决策,生成行为"""
pass
Event (事件)
事件是注入到模拟中的外部刺激。
class Event:
"""
事件类型:
- POST: 发布内容
- TREND: 热门话题
- NEWS: 新闻事件
- POLICY: 政策变化
"""
def __init__(
self,
type: str,
content: str,
source: str,
timestamp: datetime = None
):
pass
3.3 术语表
| 术语 | 英文 | 解释 |
|---|---|---|
| 智能体 | Agent | 模拟中的个体,具有独立人格和行为 |
| 环境 | Environment | Agent 生活的数字世界 |
| 模拟 | Simulation | 一次完整的模拟实验 |
| 人设 | Profile | Agent 的人口统计信息 |
| 人格 | Personality | Agent 的心理特征(大五人格) |
| 记忆 | Memory | Agent 的知识库和经验 |
| 信息流 | Feed | Agent 看到的内容列表 |
| 事件 | Event | 注入模拟的外部刺激 |
| 社交图谱 | Social Graph | Agent 之间的关系网络 |
| 时间步 | Time Step | 模拟的最小时间单位 |
四、功能详解
4.1 Agent 人设生成
功能概述
MiroFish 支持基于真实数据生成 Agent 人设,确保模拟的真实性。
原理说明
人设生成流程:
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 人口数据 │ --> │ 采样策略 │ --> │ 人设模板 │
│ (统计) │ │ (分层) │ │ (LLM生成) │
└─────────────┘ └─────────────┘ └─────────────┘
基础用法
from mirofish import AgentGenerator
# 创建人设生成器
generator = AgentGenerator(
population_data="data/china_population.json",
llm_model="gpt-4-turbo"
)
# 生成单个 Agent
agent = generator.generate(
demographics={
"age_range": (25, 35),
"gender": "female",
"city": "北京"
}
)
print(agent.profile)
# {
# "name": "李明",
# "age": 28,
# "gender": "female",
# "occupation": "产品经理",
# "education": "硕士",
# "interests": ["科技", "阅读", "健身"]
# }
# 批量生成 Agent
agents = generator.batch_generate(
count=1000,
distribution={
"age": "realistic", # 按真实人口分布
"gender": "balanced",
"region": "weighted"
}
)
进阶用法
# 自定义人格模板
personality_template = """
你是一个{age}岁的{occupation},居住在{city}。
你的兴趣是{interests}。
你的性格特点:{traits}
请生成一个符合以上特征的社交媒体用户画像。
"""
# 使用自定义模板
agent = generator.generate(
template=personality_template,
variables={
"age": 30,
"occupation": "软件工程师",
"city": "深圳",
"interests": "编程、游戏、咖啡",
"traits": "理性、好奇、幽默"
}
)
# 注入领域知识
agent = generator.generate_with_knowledge(
demographics={"age": 25, "city": "上海"},
knowledge_base="data/tech_industry.json"
)
4.2 GraphRAG 知识注入
功能概述
将领域知识注入到 Agent 的记忆系统中,使其具备专业知识背景。
原理说明
GraphRAG 注入流程:
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 知识文档 │ --> │ 实体抽取 │ --> │ 关系抽取 │
└─────────────┘ └─────────────┘ └─────────────┘
│
▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Agent 记忆 │ <-- │ 向量嵌入 │ <-- │ 知识图谱 │
└─────────────┘ └─────────────┘ └─────────────┘
基础用法
from mirofish import KnowledgeInjector
# 创建知识注入器
injector = KnowledgeInjector(
graph_db="neo4j://localhost:7687",
embedding_model="text-embedding-3-small"
)
# 从文档注入知识
injector.inject_from_documents(
documents=["data/news_1.txt", "data/news_2.txt"],
agent_ids=["agent_0", "agent_1"]
)
# 从 URL 注入知识
injector.inject_from_url(
url="https://example.com/article",
agent_ids=["agent_0"]
)
进阶用法
# 批量知识注入
injector.batch_inject(
knowledge_base={
"tech_news": ["news1.txt", "news2.txt"],
"product_info": ["product.md"],
"user_reviews": ["reviews.json"]
},
distribution={
"tech_news": 0.8, # 80% 的 Agent 看到科技新闻
"product_info": 1.0, # 所有 Agent 了解产品信息
"user_reviews": 0.3 # 30% 的 Agent 看到用户评价
}
)
# 动态知识更新
@sim.on_time_step
def update_knowledge(sim, step):
"""每个时间步更新知识"""
new_events = fetch_latest_news()
injector.inject_from_events(new_events)
4.3 社交网络构建
功能概述
构建 Agent 之间的社交关系网络,模拟真实的社交结构。
原理说明
社交网络模型:
├── 随机网络 (Erdos-Renyi)
├── 小世界网络 (Watts-Strogatz)
├── 无标度网络 (Barabasi-Albert) ← 默认
└── 真实网络导入
基础用法
from mirofish import SocialGraphBuilder
# 创建社交图谱构建器
builder = SocialGraphBuilder()
# 构建无标度网络
graph = builder.build(
num_agents=1000,
model="barabasi_albert",
avg_connections=50 # 平均每个 Agent 50 个关注
)
# 查看网络统计
print(f"平均路径长度: {graph.avg_path_length}")
print(f"聚类系数: {graph.clustering_coefficient}")
print(f"度分布幂律指数: {graph.power_law_exponent}")
进阶用法
# 自定义网络结构
graph = builder.build_custom(
num_agents=1000,
influencer_ratio=0.01, # 1% 的意见领袖
avg_connections=50,
community_structure=True # 包含社区结构
)
# 导入真实网络
graph = builder.import_from_file(
"data/real_social_network.json",
format="edge_list"
)
# 动态网络演化
@sim.on_time_step
def evolve_network(sim, step):
"""网络随时间演化"""
graph.evolve(
new_connection_prob=0.01,
break_connection_prob=0.005
)
4.4 模拟运行与控制
功能概述
控制模拟的运行、暂停、恢复和监控。
基础用法
from mirofish import Simulation
# 创建模拟
sim = Simulation(
environment=env,
duration=30, # 30 天虚拟时间
time_step=3600, # 每小时一步
seed=42 # 可复现
)
# 运行模拟
results = sim.run()
# 带回调运行
@sim.on_progress
def on_progress(sim, progress):
print(f"进度: {progress*100:.1f}%")
@sim.on_event
def on_event(sim, event):
print(f"事件: {event.type} - {event.content}")
results = sim.run()
进阶用法
# 分步执行(用于调试)
sim.start()
for _ in range(100):
sim.step()
print(f"当前时间: {sim.current_time}")
# 暂停和恢复
sim.start()
time.sleep(10)
sim.pause()
# ... 做一些分析 ...
sim.resume()
# 检查点保存
sim.save_checkpoint("checkpoints/day_7.pkl")
# 从检查点恢复
sim = Simulation.load_checkpoint("checkpoints/day_7.pkl")
sim.run()
4.5 结果分析与报告
功能概述
分析模拟结果,生成可视化报告。
基础用法
from mirofish import ResultAnalyzer
# 创建分析器
analyzer = ResultAnalyzer(results)
# 基本统计
print(f"总互动数: {analyzer.total_interactions()}")
print(f"传播范围: {analyzer.reach()}")
print(f"平均情感: {analyzer.avg_sentiment()}")
# 情感分布
sentiment_dist = analyzer.sentiment_distribution()
# {"positive": 0.6, "neutral": 0.3, "negative": 0.1}
# 关键意见领袖
influencers = analyzer.top_influencers(top_n=10)
# 传播路径
paths = analyzer.propagation_paths(event_id="event_0")
进阶用法
# 生成可视化报告
report = analyzer.generate_report(
output="reports/simulation_report.html",
include_charts=True,
include_network=True
)
# 导出数据
analyzer.export_data(
format="csv",
output_dir="data/results/"
)
# 与 ReportAgent 对话
from mirofish import ReportAgent
agent = ReportAgent(results)
response = agent.chat("为什么负面情绪在第 3 天突然上升?")
print(response)
五、实战案例
5.1 案例:新品发布舆情预测
完整示例:预测一款新手机发布的舆情走向。
# examples/new_product_launch.py
from mirofish import (
Simulation, Environment, AgentGenerator,
KnowledgeInjector, Event, ResultAnalyzer
)
# ========== 1. 环境准备 ==========
# 创建环境
env = Environment(
name="phone_launch",
platform="weibo"
)
# ========== 2. 生成 Agent ==========
# 目标用户群体
target_demographics = {
"age_range": (18, 45),
"interests": ["科技", "手机", "数码"],
"cities": ["北京", "上海", "广州", "深圳", "成都"]
}
# 生成 10000 个 Agent
generator = AgentGenerator()
agents = generator.batch_generate(
count=10000,
demographics=target_demographics,
distribution="realistic"
)
env.add_agents(agents)
# ========== 3. 构建社交网络 ==========
from mirofish import SocialGraphBuilder
builder = SocialGraphBuilder()
graph = builder.build(
num_agents=10000,
model="barabasi_albert",
avg_connections=100,
influencer_ratio=0.02 # 2% 意见领袖
)
env.set_social_graph(graph)
# ========== 4. 注入产品知识 ==========
injector = KnowledgeInjector()
injector.inject_from_documents(
documents=[
"knowledge/product_specs.md",
"knowledge/brand_history.md",
"knowledge/competitor_comparison.md"
],
distribution=0.8 # 80% 的 Agent 了解产品
)
# ========== 5. 配置模拟 ==========
sim = Simulation(
environment=env,
duration=14, # 模拟 14 天
time_step=1800, # 每 30 分钟一步
seed=42
)
# ========== 6. 注入初始事件 ==========
# 发布会事件
sim.inject_event(Event(
type="POST",
content="""
【重磅发布】全新旗舰手机 X1 正式发布!
- 革命性的 AI 芯片
- 超长续航 72 小时
- 价格:3999 元起
#X1发布会# #科技新品#
""",
source="official_account",
timestamp=sim.start_time
))
# ========== 7. 运行模拟 ==========
results = sim.run()
# ========== 8. 分析结果 ==========
analyzer = ResultAnalyzer(results)
print("=== 舆情分析报告 ===")
print(f"传播范围: {analyzer.reach():,} 人")
print(f"总互动数: {analyzer.total_interactions():,}")
print(f"情感分布: {analyzer.sentiment_distribution()}")
print(f"关键意见领袖: {analyzer.top_influencers(5)}")
# 生成详细报告
analyzer.generate_report(
output="reports/phone_launch_report.html"
)
# ========== 9. 与 ReportAgent 对话 ==========
from mirofish import ReportAgent
report_agent = ReportAgent(results)
# 提问分析
questions = [
"发布会后舆情如何变化?",
"哪些话题引发了最多的讨论?",
"负面评价主要集中在哪些方面?",
"意见领袖的态度分布如何?"
]
for q in questions:
answer = report_agent.chat(q)
print(f"Q: {q}")
print(f"A: {answer}\n")
5.2 案例:危机公关模拟
模拟产品危机事件的传播和应对。
# examples/crisis_simulation.py
from mirofish import Simulation, Environment, Event
env = Environment(name="crisis_sim", platform="weibo")
# ... 创建 Agent 和网络 ...
sim = Simulation(env, duration=7, time_step=1800)
# ========== 场景:电池爆炸事件 ==========
# Day 1: 危机爆发
sim.schedule_event(
time="day_1_hour_8",
event=Event(
type="NEWS",
content="【突发】某用户反馈 X1 手机充电时发生冒烟",
source="news_media"
)
)
# Day 2: 官方回应(方案A:快速回应)
sim.schedule_event(
time="day_2_hour_10",
event=Event(
type="OFFICIAL",
content="""
【官方声明】我们关注到相关反馈,已第一时间联系用户,
正在全力调查。请放心,我们将对消费者负责到底。
""",
source="official_account"
),
scenario="response_a" # 标记为方案A
)
# 运行方案A
results_a = sim.run(scenario="response_a")
# ========== 对比方案B:延迟回应 ==========
sim.reset()
# Day 4: 官方回应(方案B:延迟回应)
sim.schedule_event(
time="day_4_hour_10",
event=Event(
type="OFFICIAL",
content="【官方声明】针对近期事件,我们已完成初步调查...",
source="official_account"
),
scenario="response_b"
)
results_b = sim.run(scenario="response_b")
# ========== 对比分析 ==========
from mirofish import ResultAnalyzer
analyzer_a = ResultAnalyzer(results_a)
analyzer_b = ResultAnalyzer(results_b)
print("=== 方案对比 ===")
print(f"方案A (快速回应):")
print(f" 最大负面占比: {analyzer_a.max_negative_ratio():.1%}")
print(f" 恢复时间: {analyzer_a.recovery_time()} 天")
print(f"方案B (延迟回应):")
print(f" 最大负面占比: {analyzer_b.max_negative_ratio():.1%}")
print(f" 恢复时间: {analyzer_b.recovery_time()} 天")
六、进阶主题
6.1 性能优化
批量推理
from mirofish import ScalableInferencer
# 配置批量推理
inferencer = ScalableInferencer(
batch_size=100,
max_concurrent=20,
cache_enabled=True
)
# 在模拟中使用
sim = Simulation(
environment=env,
inferencer=inferencer
)
分布式部署
# docker-compose-scale.yml
version: '3.8'
services:
inferencer:
image: mirofish/inferencer
deploy:
replicas: 4 # 4 个推理节点
environment:
- REDIS_URL=redis://redis:6379
redis:
image: redis:7
ports:
- "6379:6379"
6.2 自定义 Agent 行为
from mirofish import Agent, AgentBehavior
class CustomAgent(Agent):
"""自定义 Agent 行为"""
def decide(self, observation):
"""自定义决策逻辑"""
# 获取相关信息
feed = observation.feed
personality = self.personality
# 自定义决策规则
if self._is_interested(feed, personality):
return self._create_interaction_decision(feed)
else:
return self._create_skip_decision()
def _is_interested(self, feed, personality):
"""判断是否感兴趣"""
# 实现自定义逻辑
pass
# 注册自定义 Agent
from mirofish import register_agent_type
register_agent_type("custom", CustomAgent)
# 使用自定义 Agent
agent = CustomAgent(id="custom_1", profile={...})
6.3 与外部系统集成
from mirofish import WebhookNotifier
# 配置 Webhook 通知
notifier = WebhookNotifier(
url="https://your-server.com/webhook",
events=["simulation.start", "simulation.complete", "event.injected"]
)
sim = Simulation(
environment=env,
notifier=notifier
)
6.4 监控和日志
from mirofish import MetricsCollector
# 配置指标收集
metrics = MetricsCollector(
backend="prometheus",
endpoint="http://localhost:9090"
)
sim = Simulation(
environment=env,
metrics=metrics
)
# 自定义指标
@metrics.gauge("agent_active_count")
def count_active_agents(sim):
return len([a for a in sim.agents if a.is_active])
七、常见问题
Q1: 模拟速度太慢怎么办?
# 优化策略:
# 1. 减少时间步数
sim = Simulation(env, time_step=7200) # 2小时一步
# 2. 启用批量推理
inferencer = ScalableInferencer(batch_size=200)
# 3. 使用更快的 LLM
sim = Simulation(env, llm_model="gpt-3.5-turbo")
# 4. 减少 Agent 数量(测试阶段)
agents = generator.batch_generate(count=1000)
Q2: 如何确保模拟的可复现性?
# 设置随机种子
sim = Simulation(env, seed=42)
# 保存检查点
sim.save_checkpoint("checkpoint.pkl")
# 记录所有随机决策
sim = Simulation(env, log_decisions=True)
Q3: GraphRAG 注入失败怎么办?
# 检查 Neo4j 连接
curl http://localhost:7474
# 检查嵌入模型
python -c "from openai import OpenAI; print(OpenAI().embeddings.create)"
# 查看日志
tail -f logs/mirofish.log
Q4: 如何处理中文语料?
# 使用中文优化的嵌入模型
injector = KnowledgeInjector(
embedding_model="text-embedding-3-large",
language="zh"
)
# 配置中文分词
from mirofish import ChineseTokenizer
injector.set_tokenizer(ChineseTokenizer())
Q5: 内存不足怎么办?
# 启用流式处理
sim = Simulation(env, streaming=True)
# 分批加载 Agent
for batch in agent_batches:
env.add_agents(batch)
sim.run_steps(100)
env.clear_inactive_agents()
# 使用数据库存储
sim = Simulation(env, storage_backend="mongodb")
八、参考资料
官方资源
学术论文
- OASIS: Open Agent Social Interaction Simulations
- GraphRAG: Unlocking LLM Discovery on Narrative Private Data
- Generative Agents: Interactive Simulacra of Human Behavior
相关技术
学习路线图
Week 1: 基础入门
├── Day 1-2: 环境搭建,运行 Hello World
├── Day 3-4: 理解核心概念
└── Day 5-7: 完成第一个完整模拟
Week 2: 功能深入
├── Day 1-2: Agent 人设生成
├── Day 3-4: GraphRAG 知识注入
└── Day 5-7: 社交网络构建
Week 3: 实战应用
├── Day 1-3: 新品发布舆情预测案例
├── Day 4-5: 危机公关模拟案例
└── Day 6-7: 自定义场景开发
Week 4: 进阶优化
├── Day 1-2: 性能优化
├── Day 3-4: 分布式部署
└── Day 5-7: 生产环境最佳实践
教程持续更新中,如有问题请在 GitHub 提 Issue