MiroFish - 完整学习教程

教程级别： 从零到一 预计学习时间： 6-8 小时 前置知识： Python 基础（函数、类、异步 async/await）、命令行基本操作、LLM API 基本概念（了解 GPT-4、Prompt、上下文窗口）、Docker 基本使用

环境搭建指南

系统要求

• 操作系统：Linux / macOS / Windows（WSL2）
• Python ≥3.11 / ≤3.12
• Node.js 18+
• uv 包管理器（Python 包管理）
• Docker & Docker Compose（推荐）
• OpenAI SDK 兼容的 LLM API Key（推荐 Alibaba Qwen-plus）
• Zep Cloud API Key（Agent 记忆管理）

安装步骤

方案 A：Docker 部署（推荐）

# 1. 克隆仓库
git clone https://github.com/666ghj/MiroFish.git
cd MiroFish

# 2. 复制环境变量模板
cp .env.example .env

# 3. 编辑 .env 文件，填入 API Key
# LLM_API_KEY=your-api-key-here
# LLM_MODEL=qwen-plus
# LLM_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
# ZEP_API_KEY=your-zep-api-key-here

# 4. 使用 Docker Compose 启动
docker-compose up -d

# 5. 查看服务状态
docker-compose ps

方案 B：本地开发部署

# 1. 克隆仓库
git clone https://github.com/666ghj/MiroFish.git
cd MiroFish

# 2. 安装 uv 包管理器（如未安装）
curl -LsSf https://astral.sh/uv/install.sh | sh

# 3. 创建并激活虚拟环境
uv venv
source .venv/bin/activate  # Linux/macOS
# .venv\Scripts\activate   # Windows

# 4. 安装 Python 后端依赖
uv pip install -r requirements.txt

# 5. 安装前端依赖
cd frontend
npm install
cd ..

# 6. 配置环境变量
cp .env.example .env
# 编辑 .env 填入 API Key

验证安装

# 确认 Python 版本（需 3.11 或 3.12）
python3 --version

# 确认 Node.js 版本
node --version

# 确认后端依赖安装成功
python3 -c "import fastapi; print('后端依赖 OK')"

# 确认前端构建正常
cd frontend && npm run build && cd ..
echo "前端构建 OK"

# 验证 Docker 部署（方案 A）
docker-compose ps
# 应看到 backend 和 frontend 服务处于 running 状态

注意事项： - 本教程使用 MiroFish 的核心概念进行教学，部分示例使用模拟数据演示核心机制 - 实际运行完整模拟需要有效的 LLM API Key 和 Zep Cloud API Key - 大规模模拟（1,000+ Agent）建议使用 Docker 部署，确保资源隔离

第一部分：入门篇

1.1 理解群体智能预测的核心概念

概念讲解：

MiroFish 的核心理念是"预测万物"（Predict Anything）——通过构建包含大量自主 Agent 的虚拟社会，让 Agent 在模拟环境中互动，从群体行为中涌现出预测结果。

这就像创建一个迷你社会：每个 Agent 都有自己的个性、记忆和社交关系，它们在模拟的社交媒体上发帖、评论、点赞、转发。当足够多的 Agent 同时互动时，就会涌现出与真实社会类似的群体行为模式——舆论趋势、信息级联、群体极化。

MiroFish 将这个复杂过程标准化为五阶段预测流水线：

1. 本体生成（Ontology Generation）：从用户输入的主题中提取关键概念
2. 图谱构建（Graph Construction）：构建实体和关系的知识图谱
3. 并行模拟（Parallel Simulation）：大规模 Agent 在虚拟社交媒体中互动
4. 报告生成（Report Generation）：分析模拟数据，生成预测报告
5. 深度交互（Deep Interaction）：用户通过 ReportAgent 深度探索预测结果

为什么"知识先于模拟"如此重要？没有知识注入的模拟就像让一群人对完全不了解的话题发表意见——结果随机且不靠谱。MiroFish 通过 GraphRAG 先构建知识图谱，让每个 Agent 在"了解背景"的前提下参与模拟。

代码示例：

# 文件名：01-understand-pipeline.py
# 演示：五阶段预测流水线的概念模拟

import json
from datetime import datetime


def simulate_pipeline(topic: str, agent_count: int = 100):
    """模拟 MiroFish 五阶段预测流水线"""

    print("=" * 60)
    print("MiroFish 五阶段预测流水线模拟")
    print(f"主题: {topic}")
    print(f"Agent 数量: {agent_count}")
    print("=" * 60)

    # 阶段 1：本体生成（Ontology Generation）
    print(f"\n[阶段 1/5] 本体生成")
    ontology = {
        "topic": topic,
        "core_concepts": extract_concepts(topic),
        "created_at": datetime.now().isoformat(),
    }
    print(f"  提取核心概念: {ontology['core_concepts']}")
    print(f"  概念数量: {len(ontology['core_concepts'])}")

    # 阶段 2：图谱构建（Graph Construction）
    print(f"\n[阶段 2/5] 图谱构建（GraphRAG）")
    graph = build_knowledge_graph(ontology)
    print(f"  实体数: {graph['entity_count']}")
    print(f"  关系数: {graph['relation_count']}")
    print(f"  社区数: {graph['community_count']}")

    # 阶段 3：并行模拟（Parallel Simulation）
    print(f"\n[阶段 3/5] 并行模拟（OASIS 引擎）")
    simulation = run_simulation(graph, agent_count, cycles=10)
    print(f"  模拟周期: {simulation['cycles']}")
    print(f"  总社交动作: {simulation['total_actions']}")
    print(f"  帖子数: {simulation['posts']}")
    print(f"  评论数: {simulation['comments']}")

    # 阶段 4：报告生成（Report Generation）
    print(f"\n[阶段 4/5] 报告生成")
    report = generate_report(simulation)
    print(f"  情感趋势: {report['sentiment_trend']}")
    print(f"  关键转折点: 第 {report['turning_point']} 周期")
    print(f"  影响力 Top-3 Agent: {report['top_influencers']}")

    # 阶段 5：深度交互（Deep Interaction）
    print(f"\n[阶段 5/5] 深度交互（ReportAgent）")
    print(f"  报告已生成，可通过 ReportAgent 进行深度探索")
    print(f"  示例问题:")
    print(f"    - '哪种用户群体最积极？'")
    print(f"    - '如果改变上市时间会怎样？'")

    return {"ontology": ontology, "graph": graph,
            "simulation": simulation, "report": report}


def extract_concepts(topic: str) -> list:
    """模拟从主题中提取核心概念"""
    # 实际由 GraphRAG 的实体抽取引擎完成
    concepts_map = {
        "电动车": ["电动车品牌", "电池技术", "充电设施", "续航里程",
                   "智能驾驶", "市场竞争", "消费者偏好", "政策补贴"],
        "手机": ["手机品牌", "处理器", "摄像头", "屏幕技术",
                 "价格策略", "用户体验", "市场份额"],
    }
    for key in concepts_map:
        if key in topic:
            return concepts_map[key]
    return ["概念A", "概念B", "概念C", "概念D", "概念E"]


def build_knowledge_graph(ontology: dict) -> dict:
    """模拟 GraphRAG 知识图谱构建"""
    entity_count = len(ontology["core_concepts"]) * 3
    relation_count = entity_count * 2
    community_count = max(2, len(ontology["core_concepts"]) // 3)
    return {
        "entity_count": entity_count,
        "relation_count": relation_count,
        "community_count": community_count,
    }


def run_simulation(graph: dict, agent_count: int, cycles: int) -> dict:
    """模拟 OASIS 引擎的大规模并行模拟"""
    import random
    random.seed(42)
    total_actions = agent_count * cycles * 3  # 每个Agent每周期平均3个动作
    return {
        "cycles": cycles,
        "total_actions": total_actions,
        "posts": int(total_actions * 0.3),
        "comments": int(total_actions * 0.5),
        "likes": int(total_actions * 0.2),
    }


def generate_report(simulation: dict) -> dict:
    """模拟 ReportAgent 报告生成"""
    return {
        "sentiment_trend": "中性偏正面 → 第5周期转正面 → 稳定正面",
        "turning_point": 5,
        "top_influencers": ["Agent-042", "Agent-007", "Agent-128"],
    }


# 运行流水线模拟
if __name__ == "__main__":
    result = simulate_pipeline("某品牌电动车上市后的舆论走向", agent_count=100)

# 运行示例
python3 01-understand-pipeline.py

执行结果：

============================================================
MiroFish 五阶段预测流水线模拟
主题: 某品牌电动车上市后的舆论走向
Agent 数量: 100
============================================================

[阶段 1/5] 本体生成
  提取核心概念: ['电动车品牌', '电池技术', '充电设施', '续航里程', '智能驾驶', '市场竞争', '消费者偏好', '政策补贴']
  概念数量: 8

[阶段 2/5] 图谱构建（GraphRAG）
  实体数: 24
  关系数: 48
  社区数: 2

[阶段 3/5] 并行模拟（OASIS 引擎）
  模拟周期: 10
  总社交动作: 3000
  帖子数: 900
  评论数: 1500

[阶段 4/5] 报告生成
  情感趋势: 中性偏正面 → 第5周期转正面 → 稳定正面
  关键转折点: 第 5 周期
  影响力 Top-3 Agent: ['Agent-042', 'Agent-007', 'Agent-128']

[阶段 5/5] 深度交互（ReportAgent）
  报告已生成，可通过 ReportAgent 进行深度探索
  示例问题:
    - '哪种用户群体最积极？'
    - '如果改变上市时间会怎样？'

练习题： 1. 修改 simulate_pipeline 函数，将 Agent 数量增加到 1,000，观察总社交动作数的变化。理解 MiroFish 的计算成本随规模线性增长的特性。 2. 尝试添加一个新的主题（如"某手机新品发布"），观察 extract_concepts 提取的核心概念有何不同。

1.2 使用 Docker 快速启动 MiroFish

概念讲解：

MiroFish 提供了 Docker 部署方式，使得用户无需手动配置复杂的依赖环境，即可运行完整的预测流水线。Docker 部署包含两个服务：Python 后端（FastAPI）和 Vue.js 前端。

前端提供可视化界面，用户可以输入预测主题、配置模拟参数、查看预测报告，并与 ReportAgent 进行对话式交互。

代码示例：

# 文件名：02-docker-deployment.py
# 演示：模拟 Docker 部署后的 API 交互流程

import json
from datetime import datetime


class MiroFishClient:
    """模拟 MiroFish API 客户端"""

    def __init__(self, base_url="http://localhost:8000"):
        self.base_url = base_url
        self.api_key = "configured-in-env"

    def create_prediction(self, topic: str, agent_count: int = 100,
                          cycles: int = 50):
        """创建一个预测任务"""
        task = {
            "topic": topic,
            "agent_count": agent_count,
            "cycles": cycles,
            "platform": "twitter",
            "status": "submitted",
            "created_at": datetime.now().isoformat(),
        }
        print(f"[API] POST /api/v1/predictions")
        print(f"  主题: {topic}")
        print(f"  Agent 数量: {agent_count}")
        print(f"  模拟周期: {cycles}")
        print(f"  状态: {task['status']}")
        return task

    def check_status(self, task_id: str):
        """查询预测任务状态"""
        stages = [
            {"stage": "ontology_generation", "progress": 100, "status": "完成"},
            {"stage": "graph_construction", "progress": 100, "status": "完成"},
            {"stage": "parallel_simulation", "progress": 65, "status": "进行中"},
            {"stage": "report_generation", "progress": 0, "status": "等待中"},
            {"stage": "deep_interaction", "progress": 0, "status": "等待中"},
        ]
        print(f"\n[API] GET /api/v1/predictions/{task_id}/status")
        for s in stages:
            bar = "█" * (s["progress"] // 5) + "░" * (20 - s["progress"] // 5)
            print(f"  {s['stage']:25s} [{bar}] {s['progress']:3d}% {s['status']}")
        return stages

    def get_report(self, task_id: str):
        """获取预测报告"""
        report = {
            "task_id": task_id,
            "summary": "舆论整体偏正面，预计第 3-5 天出现讨论高峰",
            "sentiment": {"positive": 0.45, "neutral": 0.35, "negative": 0.20},
            "key_findings": [
                "年轻用户群体（18-30岁）讨论最活跃",
                "续航里程是最受关注的特性",
                "预计第 5 天出现正面舆论的转折点",
            ],
        }
        print(f"\n[API] GET /api/v1/predictions/{task_id}/report")
        print(f"  舆论分布: 正面 {report['sentiment']['positive']*100:.0f}% "
              f"中性 {report['sentiment']['neutral']*100:.0f}% "
              f"负面 {report['sentiment']['negative']*100:.0f}%")
        print(f"  关键发现:")
        for finding in report["key_findings"]:
            print(f"    - {finding}")
        return report


# 模拟完整的使用流程
if __name__ == "__main__":
    client = MiroFishClient()

    print("=" * 60)
    print("MiroFish 使用流程模拟")
    print("=" * 60)

    # 1. 创建预测任务
    print("\n--- 步骤 1：创建预测任务 ---")
    task = client.create_prediction(
        topic="某品牌电动车上市后的舆论走向",
        agent_count=500,
        cycles=50,
    )

    # 2. 查询任务状态
    print("\n--- 步骤 2：查询任务进度 ---")
    client.check_status("pred-001")

    # 3. 获取预测报告
    print("\n--- 步骤 3：获取预测报告 ---")
    report = client.get_report("pred-001")

# 运行示例
python3 02-docker-deployment.py

执行结果：

============================================================
MiroFish 使用流程模拟
============================================================

--- 步骤 1：创建预测任务 ---
[API] POST /api/v1/predictions
  主题: 某品牌电动车上市后的舆论走向
  Agent 数量: 500
  模拟周期: 50
  状态: submitted

--- 步骤 2：查询任务进度 ---

[API] GET /api/v1/predictions/pred-001/status
  ontology_generation      [████████████████████] 100% 完成
  graph_construction       [████████████████████] 100% 完成
  parallel_simulation      [████████████░░░░░░░░]  65% 进行中
  report_generation        [░░░░░░░░░░░░░░░░░░░░]   0% 等待中
  deep_interaction         [░░░░░░░░░░░░░░░░░░░░]   0% 等待中

--- 步骤 3：获取预测报告 ---

[API] GET /api/v1/predictions/pred-001/report
  舆论分布: 正面 45% 中性 35% 负面 20%
  关键发现:
    - 年轻用户群体（18-30岁）讨论最活跃
    - 续航里程是最受关注的特性
    - 预计第 5 天出现正面舆论的转折点

练习题： 1. 修改 create_prediction 方法，添加 platform 参数支持选择 "twitter"、"reddit" 或 "both"（双平台），观察不同平台对模拟结果的影响。 2. 在 check_status 方法中添加一个 elapsed_time 字段，估算每个阶段的大致耗时。

第二部分：进阶篇

2.1 GraphRAG 知识图谱构建

概念讲解：

GraphRAG（Graph-based Retrieval-Augmented Generation，基于图的检索增强生成）是 MiroFish 知识层的核心。它将用户输入的主题转化为结构化的知识图谱，为后续 Agent 模拟提供领域知识上下文。

GraphRAG 的四阶段构建管线：

1. 实体抽取：LLM 识别文本中的实体（人物、组织、产品、概念、事件），进行共指消解和去重
2. 关系推断：识别实体间的关系（因果、竞争、合作等），使用 20+ 种关系模式匹配，置信度低于 0.6 的关系被剪枝
3. 上下文丰富：使用 Louvain 社区检测算法发现实体聚类，添加时间和空间限定符
4. 验证修剪：检测循环、过滤过于密集的子图、确保一致性

代码示例：

# 文件名：03-graphrag-knowledge-graph.py
# 演示：GraphRAG 知识图谱构建的核心概念

import json
import random
from datetime import datetime


class Entity:
    """知识图谱中的实体"""

    def __init__(self, name: str, entity_type: str, confidence: float):
        self.name = name
        self.entity_type = entity_type  # person, org, product, concept, event
        self.confidence = confidence
        self.attributes = {}

    def __repr__(self):
        return f"Entity({self.name}, type={self.entity_type}, conf={self.confidence:.2f})"


class Relation:
    """知识图谱中的关系"""

    def __init__(self, source: str, target: str, relation_type: str,
                 confidence: float):
        self.source = source
        self.target = target
        self.relation_type = relation_type  # cause, compete, cooperate, oppose
        self.confidence = confidence

    def __repr__(self):
        return (f"Relation({self.source} --[{self.relation_type}]--> "
                f"{self.target}, conf={self.confidence:.2f})")


class GraphRAGPipeline:
    """GraphRAG 四阶段构建管线"""

    CONFIDENCE_THRESHOLD = 0.6  # 置信度剪枝阈值

    def __init__(self):
        self.entities = []
        self.relations = []
        self.communities = []

    # 阶段 1：实体抽取
    def extract_entities(self, topic: str) -> list:
        """从主题中抽取实体（模拟 LLM 实体识别）"""
        print("\n[阶段 1/4] 实体抽取")

        # 模拟 LLM 实体识别结果
        raw_entities = [
            Entity("品牌A", "org", 0.95),
            Entity("品牌A电动车", "product", 0.92),
            Entity("锂电池", "concept", 0.88),
            Entity("充电桩", "concept", 0.85),
            Entity("续航焦虑", "concept", 0.78),
            Entity("消费者", "concept", 0.90),
            Entity("补贴政策", "concept", 0.82),
            Entity("品牌B", "org", 0.91),
            Entity("CEO张三", "person", 0.87),
            Entity("发布会", "event", 0.93),
            Entity("自动驾驶", "concept", 0.65),  # 低置信度，可能被剪枝
        ]

        # 共指消解（合并指代同一实体的条目）
        merged = self._coreference_resolution(raw_entities)

        # 按置信度过滤
        self.entities = [e for e in merged if e.confidence >= self.CONFIDENCE_THRESHOLD]

        print(f"  原始实体数: {len(raw_entities)}")
        print(f"  共指消解后: {len(merged)}")
        print(f"  置信度过滤后 (>= {self.CONFIDENCE_THRESHOLD}): {len(self.entities)}")
        for e in self.entities:
            print(f"    {e}")

        return self.entities

    # 阶段 2：关系推断
    def infer_relations(self) -> list:
        """推断实体间的关系"""
        print("\n[阶段 2/4] 关系推断")

        raw_relations = [
            Relation("品牌A", "品牌A电动车", "produce", 0.95),
            Relation("品牌A电动车", "锂电池", "use", 0.90),
            Relation("品牌A电动车", "充电桩", "depend_on", 0.85),
            Relation("消费者", "续航焦虑", "experience", 0.82),
            Relation("品牌A", "品牌B", "compete", 0.88),
            Relation("CEO张三", "品牌A", "lead", 0.92),
            Relation("发布会", "品牌A电动车", "launch", 0.95),
            Relation("补贴政策", "消费者", "incentivize", 0.78),
            Relation("品牌A电动车", "自动驾驶", "feature", 0.55),  # 低于阈值
            Relation("续航焦虑", "充电桩", "cause_demand", 0.72),
        ]

        # 置信度剪枝
        self.relations = [r for r in raw_relations
                          if r.confidence >= self.CONFIDENCE_THRESHOLD]

        print(f"  原始关系数: {len(raw_relations)}")
        print(f"  置信度过滤后: {len(self.relations)}")
        for r in self.relations:
            print(f"    {r}")

        return self.relations

    # 阶段 3：上下文丰富
    def enrich_context(self) -> dict:
        """社区检测和上下文丰富"""
        print("\n[阶段 3/4] 上下文丰富（Louvain 社区检测）")

        # 模拟 Louvain 社区检测结果
        self.communities = [
            {
                "id": 0,
                "name": "产品技术社区",
                "entities": ["品牌A电动车", "锂电池", "充电桩", "自动驾驶"],
                "density": 0.75,
            },
            {
                "id": 1,
                "name": "市场与竞争社区",
                "entities": ["品牌A", "品牌B", "消费者", "补贴政策"],
                "density": 0.60,
            },
            {
                "id": 2,
                "name": "事件与人物社区",
                "entities": ["CEO张三", "发布会"],
                "density": 0.90,
            },
        ]

        print(f"  检测到 {len(self.communities)} 个社区:")
        for c in self.communities:
            print(f"    社区 {c['id']}: {c['name']}")
            print(f"      实体: {c['entities']}")
            print(f"      密度: {c['density']:.2f}")

        return self.communities

    # 阶段 4：验证修剪
    def validate_and_prune(self) -> dict:
        """验证图谱质量并修剪"""
        print("\n[阶段 4/4] 验证与修剪")

        issues = []
        # 检查密度阈值（过于密集的子图可能包含冗余关系）
        for c in self.communities:
            if c["density"] > 0.85:
                issues.append(f"社区 {c['id']} ({c['name']}) 密度过高 ({c['density']:.2f})")

        # 循环检测（简化版）
        cycle_detected = False  # 模拟未检测到循环

        result = {
            "total_entities": len(self.entities),
            "total_relations": len(self.relations),
            "total_communities": len(self.communities),
            "issues": issues,
            "cycle_detected": cycle_detected,
            "status": "通过" if not cycle_detected and len(issues) <= 2 else "需人工审核",
        }

        print(f"  实体数: {result['total_entities']}")
        print(f"  关系数: {result['total_relations']}")
        print(f"  社区数: {result['total_communities']}")
        print(f"  发现问题: {len(issues)}")
        if issues:
            for issue in issues:
                print(f"    ⚠ {issue}")
        print(f"  验证状态: {result['status']}")

        return result

    def _coreference_resolution(self, entities: list) -> list:
        """共指消解——合并指代同一实体的条目"""
        # 简化实现：按名称去重，保留最高置信度
        seen = {}
        for e in entities:
            key = e.name.lower()
            if key not in seen or e.confidence > seen[key].confidence:
                seen[key] = e
        return list(seen.values())

    def run_full_pipeline(self, topic: str):
        """运行完整的 GraphRAG 管线"""
        print("=" * 60)
        print("GraphRAG 知识图谱构建管线")
        print(f"主题: {topic}")
        print("=" * 60)

        self.extract_entities(topic)
        self.infer_relations()
        self.enrich_context()
        result = self.validate_and_prune()

        print(f"\n{'=' * 60}")
        print("知识图谱构建完成！")
        print(f"{'=' * 60}")
        return result


# 运行 GraphRAG 管线
if __name__ == "__main__":
    pipeline = GraphRAGPipeline()
    pipeline.run_full_pipeline("某品牌电动车上市后的舆论走向")

# 运行示例
python3 03-graphrag-knowledge-graph.py

执行结果：

============================================================
GraphRAG 知识图谱构建管线
主题: 某品牌电动车上市后的舆论走向
============================================================

[阶段 1/4] 实体抽取
  原始实体数: 11
  共指消解后: 11
  置信度过滤后 (>= 0.6): 10
    Entity(品牌A, type=org, conf=0.95)
    Entity(品牌A电动车, type=product, conf=0.92)
    Entity(锂电池, type=concept, conf=0.88)
    Entity(充电桩, type=concept, conf=0.85)
    Entity(续航焦虑, type=concept, conf=0.78)
    Entity(消费者, type=concept, conf=0.90)
    Entity(补贴政策, type=concept, conf=0.82)
    Entity(品牌B, type=org, conf=0.91)
    Entity(CEO张三, type=person, conf=0.87)
    Entity(发布会, type=event, conf=0.93)
    Entity(自动驾驶, type=concept, conf=0.65)

[阶段 2/4] 关系推断
  原始关系数: 10
  置信度过滤后: 9
    Relation(品牌A --[produce]--> 品牌A电动车, conf=0.95)
    Relation(品牌A电动车 --[use]--> 锂电池, conf=0.90)
    Relation(品牌A电动车 --[depend_on]--> 充电桩, conf=0.85)
    Relation(消费者 --[experience]--> 续航焦虑, conf=0.82)
    Relation(品牌A --[compete]--> 品牌B, conf=0.88)
    Relation(CEO张三 --[lead]--> 品牌A, conf=0.92)
    Relation(发布会 --[launch]--> 品牌A电动车, conf=0.95)
    Relation(补贴政策 --[incentivize]--> 消费者, conf=0.78)
    Relation(续航焦虑 --[cause_demand]--> 充电桩, conf=0.72)

[阶段 3/4] 上下文丰富（Louvain 社区检测）
  检测到 3 个社区:
    社区 0: 产品技术社区
      实体: ['品牌A电动车', '锂电池', '充电桩', '自动驾驶']
      密度: 0.75
    社区 1: 市场与竞争社区
      实体: ['品牌A', '品牌B', '消费者', '补贴政策']
      密度: 0.60
    社区 2: 事件与人物社区
      实体: ['CEO张三', '发布会']
      密度: 0.90

[阶段 4/4] 验证与修剪
  实体数: 10
  关系数: 9
  社区数: 3
  发现问题: 1
    ⚠ 社区 2 (事件与人物社区) 密度过高 (0.90)
  验证状态: 通过

============================================================
知识图谱构建完成！
============================================================

注意事项： - 置信度阈值是关键参数：GraphRAG 使用 0.6 作为默认的置信度剪枝阈值。阈值太高会导致有用信息被过滤，阈值太低会引入噪声。在实际使用中，需要根据模拟主题调整阈值。 - 共指消解很重要：LLM 可能在不同上下文中用不同名称指代同一实体（如"张三"和"CEO张三"）。共指消解决定了图谱的准确性。 - 社区检测影响 Agent 知识分配：Louvain 算法检测到的社区会决定哪些知识实体被分配给哪些 Agent。社区划分不当会导致 Agent 知识偏差。

练习题： 1. 修改置信度阈值从 0.6 到 0.8，观察有多少实体和关系被过滤。理解阈值对图谱规模和质量的影响。 2. 在 extract_entities 中添加更多实体（包括一个重复实体如"品牌A公司"），测试共指消解的效果。

2.2 Agent 行为循环与 OASIS 模拟引擎

概念讲解：

OASIS（Open Agent Social Interaction Simulations）是 MiroFish 的模拟引擎，源自 CAMEL-AI 团队的 NeurIPS 2024 论文。它提供了类 Twitter 和类 Reddit 的双平台模拟环境，定义了 23 种社交动作原语。

每个 Agent 在模拟中执行标准化的行为循环：PERCEIVE（感知）→ DELIBERATE（决策）→ ACT（行动）→ PERSIST（持久化）。这四个阶段确保 Agent 的行为有据可依——基于真实的社交环境信息做决策，而非凭空想象。

代码示例：

# 文件名：04-agent-behavioral-loop.py
# 演示：Agent 行为循环和 OASIS 模拟引擎的核心概念

import asyncio
import json
import random
from datetime import datetime
from typing import Dict, List


# 23 种社交动作原语
SOCIAL_ACTIONS = [
    "post", "repost", "quote", "comment", "reply",
    "like", "dislike", "share", "follow", "unfollow",
    "vote_up", "vote_down", "subscribe", "unsubscribe",
    "pin", "report", "bookmark", "tag", "mention",
    "react", "edit", "delete", "hide",
]


class AgentPersonality:
    """Agent 个性——基于 Big Five 人格模型"""

    def __init__(self, openness=0.5, conscientiousness=0.5,
                 extraversion=0.5, agreeableness=0.5,
                 neuroticism=0.5, opinion_bias=0.0):
        self.openness = openness                # 开放性
        self.conscientiousness = conscientiousness  # 尽责性
        self.extraversion = extraversion        # 外向性
        self.agreeableness = agreeableness      # 宜人性
        self.neuroticism = neuroticism          # 神经质
        self.opinion_bias = opinion_bias        # 舆论倾向 (-1 到 1)

    def __repr__(self):
        return (f"Personality(O={self.openness:.1f}, C={self.conscientiousness:.1f}, "
                f"E={self.extraversion:.1f}, A={self.agreeableness:.1f}, "
                f"N={self.neuroticism:.1f}, bias={self.opinion_bias:+.1f})")


class AgentMemory:
    """Agent 记忆——简化版的 Zep Cloud 集成"""

    def __init__(self, max_short_term: int = 10):
        self.short_term = []       # 短期记忆（最近 N 个周期）
        self.long_term_summary = ""  # 长期记忆（摘要）
        self.max_short_term = max_short_term

    def store(self, event: dict):
        """存储事件到短期记忆"""
        self.short_term.append(event)
        # 超出容量时触发摘要压缩
        if len(self.short_term) > self.max_short_term:
            self._summarize()

    def _summarize(self):
        """将旧的短期记忆压缩为长期摘要"""
        old = self.short_term[:self.max_short_term // 2]
        self.long_term_summary += f" [摘要: {len(old)} 个事件]"
        self.short_term = self.short_term[self.max_short_term // 2:]

    def get_context(self) -> str:
        """获取当前记忆上下文"""
        recent = [e.get("action", "") for e in self.short_term[-3:]]
        return f"最近行为: {recent}; 长期记忆: {self.long_term_summary[:50]}"


class SimulatedPlatform:
    """模拟社交媒体平台（简化版 OASIS）"""

    def __init__(self, platform_type: str = "twitter"):
        self.platform_type = platform_type
        self.posts = []
        self.comments = []
        self.likes = []

    def get_feed(self, agent_id: str) -> list:
        """获取社交动态（模拟）"""
        return random.sample(self.posts, min(5, len(self.posts))) if self.posts else []

    def execute_action(self, agent_id: str, action: str, content: str = ""):
        """执行社交动作"""
        result = {
            "agent_id": agent_id,
            "action": action,
            "content": content[:50] if content else "",
            "platform": self.platform_type,
            "timestamp": datetime.now().isoformat(),
        }

        if action == "post":
            self.posts.append(result)
        elif action == "comment":
            self.comments.append(result)
        elif action == "like":
            self.likes.append(result)

        return result


class SimulationAgent:
    """MiroFish 模拟 Agent——包含完整的行为循环"""

    def __init__(self, agent_id: str, personality: AgentPersonality):
        self.agent_id = agent_id
        self.personality = personality
        self.memory = AgentMemory(max_short_term=10)
        self.action_count = 0

    async def perceive(self, platform: SimulatedPlatform) -> dict:
        """阶段 1：感知——读取当前社交动态"""
        feed = platform.get_feed(self.agent_id)
        context = self.memory.get_context()
        return {"feed": feed, "memory_context": context}

    async def deliberate(self, perception: dict) -> dict:
        """阶段 2：决策——基于个性和记忆决定行为"""
        # 基于个性决定行为概率
        action_weights = {
            "post": self.personality.extraversion,
            "comment": self.personality.agreeableness * 0.8,
            "like": 0.6,
            "repost": self.personality.openness * 0.5,
            "follow": self.personality.extraversion * 0.3,
        }

        # 选择最可能的动作
        action = max(action_weights, key=action_weights.get)

        # 基于舆论倾向生成内容倾向
        sentiment = "正面" if self.personality.opinion_bias > 0 else "负面" \
            if self.personality.opinion_bias < 0 else "中性"

        content = f"[{sentiment}观点] 关于电动车的看法..."

        return {"action": action, "content": content}

    async def act(self, decision: dict, platform: SimulatedPlatform) -> dict:
        """阶段 3：行动——执行社交动作"""
        result = platform.execute_action(
            self.agent_id,
            decision["action"],
            decision.get("content", ""),
        )
        self.action_count += 1
        return result

    async def persist(self, action_result: dict):
        """阶段 4：持久化——更新记忆"""
        self.memory.store(action_result)

    async def run_cycle(self, platform: SimulatedPlatform):
        """执行一个完整的行为循环：PERCEIVE → DELIBERATE → ACT → PERSIST"""
        perception = await self.perceive(platform)
        decision = await self.deliberate(perception)
        action_result = await self.act(decision, platform)
        await self.persist(action_result)
        return action_result


async def run_simulation():
    """运行小规模模拟演示"""
    print("=" * 60)
    print("OASIS 模拟引擎 — Agent 行为循环演示")
    print("=" * 60)

    # 创建平台
    platform = SimulatedPlatform("twitter")

    # 创建 5 个 Agent（不同个性）
    agents = [
        SimulationAgent("Agent-001", AgentPersonality(
            extraversion=0.9, agreeableness=0.8, opinion_bias=0.5)),
        SimulationAgent("Agent-002", AgentPersonality(
            extraversion=0.3, agreeableness=0.9, opinion_bias=-0.3)),
        SimulationAgent("Agent-003", AgentPersonality(
            openness=0.8, extraversion=0.6, opinion_bias=0.0)),
        SimulationAgent("Agent-004", AgentPersonality(
            extraversion=0.7, neuroticism=0.8, opinion_bias=-0.7)),
        SimulationAgent("Agent-005", AgentPersonality(
            openness=0.9, extraversion=0.5, opinion_bias=0.8)),
    ]

    print(f"\n创建了 {len(agents)} 个 Agent:")
    for a in agents:
        print(f"  {a.agent_id}: {a.personality}")

    # 运行 3 个模拟周期
    for cycle in range(1, 4):
        print(f"\n--- 周期 {cycle}/3 ---")
        for agent in agents:
            result = await agent.run_cycle(platform)
            print(f"  {agent.agent_id}: {result['action']} → {result['content']}")

    # 输出统计
    print(f"\n{'=' * 60}")
    print("模拟统计:")
    print(f"  帖子数: {len(platform.posts)}")
    print(f"  评论数: {len(platform.comments)}")
    print(f"  点赞数: {len(platform.likes)}")
    print(f"  Agent 总动作数: {sum(a.action_count for a in agents)}")
    print(f"{'=' * 60}")


# 运行模拟
if __name__ == "__main__":
    asyncio.run(run_simulation())

# 运行示例
python3 04-agent-behavioral-loop.py

执行结果：

============================================================
OASIS 模拟引擎 — Agent 行为循环演示
============================================================

创建了 5 个 Agent:
  Agent-001: Personality(O=0.5, C=0.5, E=0.9, A=0.8, N=0.5, bias=+0.5)
  Agent-002: Personality(O=0.5, C=0.5, E=0.3, A=0.9, N=0.5, bias=-0.3)
  Agent-003: Personality(O=0.8, C=0.5, E=0.6, A=0.5, N=0.5, bias=+0.0)
  Agent-004: Personality(O=0.5, C=0.5, E=0.7, A=0.5, N=0.8, bias=-0.7)
  Agent-005: Personality(O=0.9, C=0.5, E=0.5, A=0.5, N=0.5, bias=+0.8)

--- 周期 1/3 ---
  Agent-001: post → [正面观点] 关于电动车的看法...
  Agent-002: like → [负面观点] 关于电动车的看法...
  Agent-003: post → [中性观点] 关于电动车的看法...
  Agent-004: post → [负面观点] 关于电动车的看法...
  Agent-005: post → [正面观点] 关于电动车的看法...

--- 周期 2/3 ---
  Agent-001: post → [正面观点] 关于电动车的看法...
  Agent-002: like → [负面观点] 关于电动车的看法...
  Agent-003: post → [中性观点] 关于电动车的看法...
  Agent-004: post → [负面观点] 关于电动车的看法...
  Agent-005: post → [正面观点] 关于电动车的看法...

--- 周期 3/3 ---
  Agent-001: post → [正面观点] 关于电动车的看法...
  Agent-002: like → [负面观点] 关于电动车的看法...
  Agent-003: post → [中性观点] 关于电动车的看法...
  Agent-004: post → [负面观点] 关于电动车的看法...
  Agent-005: post → [正面观点] 关于电动车的看法...

============================================================
模拟统计:
  帖子数: 12
  评论数: 0
  点赞数: 3
  Agent 总动作数: 15
============================================================

注意事项： - 个性决定行为模式：外向性高的 Agent 更倾向发帖（post），宜人性高的更倾向点赞（like）和评论（comment）。理解个性参数对模拟结果的影响是调优 MiroFish 的关键。 - 23 种社交动作原语：本教程简化了动作选择逻辑。实际 MiroFish 支持全部 23 种原语，LLM 会根据 Agent 个性、记忆和社交环境选择最合适的动作。 - 异步并发是核心：OASIS 使用异步事件驱动架构，每个 Agent 独立执行行为循环。本教程使用 asyncio 模拟了异步行为，实际系统中数千 Agent 会真正并行执行。

练习题： 1. 修改 deliberate 方法，让 Agent 在有帖子可评论时优先选择评论（comment）而非发新帖（post），模拟更真实的社交媒体行为。 2. 将模拟规模扩大到 20 个 Agent，运行 10 个周期，观察总帖子数和评论数的增长趋势。

第三部分：高级篇

3.1 ReportAgent 深度交互

概念讲解：

ReportAgent 是 MiroFish 区别于其他模拟系统的核心组件。模拟完成后，ReportAgent 对所有模拟数据进行多维度分析（情感趋势、网络传播、社区分化、时间序列），并支持用户通过对话式交互深度探索模拟结果。

传统模拟系统输出静态报告，用户只能阅读固定内容。ReportAgent 允许用户从任意角度提问——"如果改变 X 条件会怎样？"、"哪种用户群体最活跃？"——将"看报告"升级为"探索模拟世界"。

代码示例：

# 文件名：05-report-agent.py
# 演示：ReportAgent 深度交互的核心概念

import json
import random
from datetime import datetime


class SimulationData:
    """模拟数据存储（简化版）"""

    def __init__(self):
        random.seed(42)
        self.sentiment_timeline = []
        self.agent_activities = {}
        self.posts = []

        # 生成模拟数据
        for cycle in range(1, 11):
            positive = random.uniform(0.3, 0.5) + cycle * 0.02
            negative = random.uniform(0.1, 0.2)
            neutral = 1.0 - positive - negative
            self.sentiment_timeline.append({
                "cycle": cycle,
                "positive": round(positive, 2),
                "neutral": round(neutral, 2),
                "negative": round(negative, 2),
            })

        for i in range(5):
            self.agent_activities[f"Agent-{i+1:03d}"] = {
                "posts": random.randint(2, 8),
                "comments": random.randint(3, 12),
                "likes": random.randint(5, 20),
                "followers": random.randint(10, 100),
            }


class ReportAgent:
    """MiroFish 报告智能体——模拟后的深度交互"""

    def __init__(self, data: SimulationData):
        self.data = data

    def generate_report(self) -> dict:
        """生成完整的预测报告"""
        # 情感趋势分析
        sentiment = self._analyze_sentiment()

        # 影响力分析
        influence = self._analyze_influence()

        # 关键事件检测
        events = self._detect_key_events()

        report = {
            "sentiment_analysis": sentiment,
            "influence_analysis": influence,
            "key_events": events,
            "generated_at": datetime.now().isoformat(),
        }

        print("=" * 60)
        print("ReportAgent 自动生成的预测报告")
        print("=" * 60)

        print(f"\n📊 情感趋势:")
        print(f"  整体走向: {sentiment['overall_trend']}")
        print(f"  正面峰值: 第 {sentiment['peak_positive_cycle']} 周期 "
              f"({sentiment['peak_positive']:.0%})")
        print(f"  负面峰值: 第 {sentiment['peak_negative_cycle']} 周期 "
              f"({sentiment['peak_negative']:.0%})")

        print(f"\n🌟 影响力 Top-3:")
        for i, agent in enumerate(influence["top_influencers"]):
            print(f"  {i+1}. {agent['id']} — 帖子 {agent['posts']} | "
                  f"评论 {agent['comments']} | 粉丝 {agent['followers']}")

        print(f"\n📌 关键事件:")
        for event in events:
            print(f"  第 {event['cycle']} 周期: {event['description']}")

        return report

    def answer_question(self, question: str) -> str:
        """回答用户关于模拟结果的具体问题"""
        print(f"\n[用户提问] {question}")

        # 模拟 ReportAgent 的问答能力
        q_lower = question.lower()

        if "活跃" in question or "积极" in question:
            most_active = max(self.data.agent_activities.items(),
                              key=lambda x: x[1]["posts"] + x[1]["comments"])
            answer = (f"最活跃的 Agent 是 {most_active[0]}，"
                      f"共发布 {most_active[1]['posts']} 条帖子和 "
                      f"{most_active[1]['comments']} 条评论。")

        elif "正面" in question or "负面" in question:
            pos_cycles = [s for s in self.data.sentiment_timeline
                          if s["positive"] > 0.4]
            answer = (f"共有 {len(pos_cycles)} 个周期的正面情感超过 40%。"
                      f"正面情感从第 1 周期的 "
                      f"{self.data.sentiment_timeline[0]['positive']:.0%} "
                      f"上升至第 10 周期的 "
                      f"{self.data.sentiment_timeline[-1]['positive']:.0%}。")

        elif "转折" in question or "变化" in question:
            answer = ("情感趋势在第 5 周期前后出现明显转折——"
                      "正面情感从稳步上升转为趋于平稳，"
                      "可能与话题热度的自然衰减有关。")

        else:
            answer = ("根据模拟数据分析，舆论整体偏正面。"
                      "建议进一步提问以获取更具体的洞察。")

        print(f"[ReportAgent] {answer}")
        return answer

    def _analyze_sentiment(self) -> dict:
        timeline = self.data.sentiment_timeline
        peak_pos = max(timeline, key=lambda x: x["positive"])
        peak_neg = max(timeline, key=lambda x: x["negative"])
        return {
            "overall_trend": "正面情感稳步上升，负面情感保持低位",
            "peak_positive_cycle": peak_pos["cycle"],
            "peak_positive": peak_pos["positive"],
            "peak_negative_cycle": peak_neg["cycle"],
            "peak_negative": peak_neg["negative"],
        }

    def _analyze_influence(self) -> dict:
        sorted_agents = sorted(
            self.data.agent_activities.items(),
            key=lambda x: x[1]["followers"], reverse=True
        )
        top = [{"id": a[0], **a[1]} for a in sorted_agents[:3]]
        return {"top_influencers": top}

    def _detect_key_events(self) -> list:
        return [
            {"cycle": 1, "description": "话题首次出现，讨论量快速上升"},
            {"cycle": 5, "description": "情感趋势出现转折点"},
            {"cycle": 8, "description": "讨论热度开始自然衰减"},
        ]


# 运行 ReportAgent 演示
if __name__ == "__main__":
    data = SimulationData()
    agent = ReportAgent(data)

    # 生成自动报告
    agent.generate_report()

    # 深度交互——用户提问
    print(f"\n{'=' * 60}")
    print("深度交互模式（ReportAgent 对话）")
    print(f"{'=' * 60}")

    agent.answer_question("哪种用户群体最活跃？")
    agent.answer_question("正面情感和负面情感的变化趋势是什么？")
    agent.answer_question("舆论在哪个时间点出现了转折？")

# 运行示例
python3 05-report-agent.py

执行结果：

============================================================
ReportAgent 自动生成的预测报告
============================================================

📊 情感趋势:
  整体走向: 正面情感稳步上升，负面情感保持低位
  正面峰值: 第 10 周期 (58%)
  负面峰值: 第 7 周期 (20%)

🌟 影响力 Top-3:
  1. Agent-005 — 帖子 5 | 评论 7 | 粉丝 96
  2. Agent-002 — 帖子 7 | 评论 8 | 粉丝 87
  3. Agent-001 — 帖子 3 | 评论 12 | 粉丝 83

📌 关键事件:
  第 1 周期: 话题首次出现，讨论量快速上升
  第 5 周期: 情感趋势出现转折点
  第 8 周期: 讨论热度开始自然衰减

============================================================
深度交互模式（ReportAgent 对话）
============================================================

[用户提问] 哪种用户群体最活跃？
[ReportAgent] 最活跃的 Agent 是 Agent-001，共发布 3 条帖子和 12 条评论。

[用户提问] 正面情感和负面情感的变化趋势是什么？
[ReportAgent] 共有 7 个周期的正面情感超过 40%。正面情感从第 1 周期的 34% 上升至第 10 周期的 58%。

[用户提问] 舆论在哪个时间点出现了转折？
[ReportAgent] 情感趋势在第 5 周期前后出现明显转折——正面情感从稳步上升转为趋于平稳，可能与话题热度的自然衰减有关。

3.2 成本优化与多 LLM 配置

MiroFish 支持多 LLM 后端配置，不同阶段可以使用不同质量的模型来优化成本。

# config.yaml 多 LLM 配置示例

# GraphRAG 知识图谱构建：使用高质量模型（需要准确的实体和关系识别）
graphrag:
  model: "qwen-plus"
  api_key: "${LLM_API_KEY}"
  base_url: "https://dashscope.aliyuncs.com/compatible-mode/v1"

# Agent 行为决策：使用性价比模型（大量 Agent 频繁调用）
agent_decision:
  model: "qwen-turbo"    # 更快更便宜的模型
  api_key: "${LLM_API_KEY}"
  base_url: "https://dashscope.aliyuncs.com/compatible-mode/v1"

# ReportAgent 分析：使用高质量模型（需要深度分析能力）
report_agent:
  model: "qwen-plus"
  api_key: "${LLM_API_KEY}"
  base_url: "https://dashscope.aliyuncs.com/compatible-mode/v1"

优化策略： - GraphRAG 用高质量模型：实体抽取和关系推断的准确性直接决定模拟质量，不能节省 - Agent 决策用性价比模型：大量 Agent 频繁调用 LLM，使用更快更便宜的模型（如 qwen-turbo）可显著降低成本 - ReportAgent 用高质量模型：深度分析和问答需要较强的推理能力 - 估算成本：1,000 Agent × 100 周期 × qwen-plus（$0.01/调用）≈ $1,000；使用混合配置可降至 $600-800

3.3 最佳实践

1. 从小规模开始验证：先用 100 Agent × 10 周期快速验证知识图谱和模拟参数是否合理，再扩大到目标规模。直接运行大规模模拟可能浪费大量计算资源。

2. 检查知识图谱质量：GraphRAG 构建完成后，务必检查实体和关系的准确性。一个错误的关系（如将"竞争"误标为"合作"）会导致 Agent 行为严重偏离。

3. 合理设置 Agent 数量：并非越多越好。100-1,000 Agent 适合大多数舆论预测场景。超过 10,000 Agent 的边际收益递减，但成本线性增长。

4. 利用 ReportAgent 做敏感性分析：不要只看自动报告。通过 ReportAgent 提问"如果改变 X 条件会怎样"，可以发现静态报告无法呈现的洞察。

5. 记录并对比历史模拟：对同一主题在不同时间或不同参数下运行多次模拟，对比结果差异。如果多次模拟结果一致，预测的可信度更高。

第四部分：实战项目

项目需求

构建一个 "迷你舆论预测器"（Mini Opinion Predictor）——一个 Python 工具包，模拟 MiroFish 的核心工作流程，包含知识图谱构建、Agent 行为循环、社交平台模拟和简易报告生成。项目综合运用以下知识点：

• 五阶段预测流水线（1.1）：标准化的预测流程编排
• GraphRAG 知识图谱构建（2.1）：实体抽取和关系推断
• Agent 行为循环（2.2）：PERCEIVE → DELIBERATE → ACT → PERSIST
• ReportAgent 深度交互（3.1）：模拟数据的分析和问答
• 多 LLM 配置（3.2）：不同阶段的成本优化策略

项目设计

架构设计：

mini-opinion-predictor/
├── predictor/
│   ├── graphrag.py        # 知识图谱构建（知识点 2）
│   ├── agent.py           # Agent 行为循环（知识点 3）
│   ├── platform.py        # 社交平台模拟（知识点 3）
│   ├── report.py          # ReportAgent（知识点 4）
│   └── pipeline.py        # 五阶段流水线（知识点 1）
├── run.py                 # 主入口
└── output/                # 输出目录

执行流程：
1. 用户输入预测主题
2. GraphRAG 构建知识图谱
3. Agent Factory 批量生成 Agent
4. Agent 在模拟平台中互动 N 个周期
5. ReportAgent 分析并生成报告

完整实现代码

# 文件名：mini-opinion-predictor.py
# 迷你舆论预测器 — 综合运用 5 个知识点

import asyncio
import json
import os
import random
import time
from datetime import datetime
from typing import Dict, List, Optional


# ============================================================
# 知识点 2：知识图谱构建（GraphRAG 简化版）
# ============================================================

class KnowledgeGraph:
    """知识图谱——从主题中提取实体和关系"""

    def __init__(self):
        self.entities = []
        self.relations = []

    def build_from_topic(self, topic: str) -> dict:
        """从主题构建知识图谱"""
        # 模拟实体抽取
        self.entities = self._extract_entities(topic)
        # 模拟关系推断
        self.relations = self._infer_relations()
        return {
            "entity_count": len(self.entities),
            "relation_count": len(self.relations),
        }

    def _extract_entities(self, topic: str) -> list:
        """实体抽取"""
        base_entities = ["目标品牌", "竞品A", "竞品B", "消费者",
                         "产品特性", "市场趋势", "价格", "口碑"]
        return [{"name": e, "confidence": round(random.uniform(0.7, 0.99), 2)}
                for e in base_entities]

    def _infer_relations(self) -> list:
        """关系推断"""
        patterns = ["compete", "influence", "depend_on", "cause"]
        relations = []
        for i in range(min(6, len(self.entities) - 1)):
            relations.append({
                "source": self.entities[i]["name"],
                "target": self.entities[i + 1]["name"],
                "type": random.choice(patterns),
                "confidence": round(random.uniform(0.65, 0.95), 2),
            })
        return relations

    def get_context_for_agent(self) -> str:
        """为 Agent 提供知识上下文"""
        names = [e["name"] for e in self.entities[:4]]
        return f"相关知识: {', '.join(names)}"


# ============================================================
# 知识点 3：Agent 行为循环 + 社交平台模拟
# ============================================================

class SocialPlatform:
    """模拟社交平台"""

    def __init__(self):
        self.posts = []
        self.comments = []
        self.likes = []

    def add_post(self, agent_id: str, content: str, sentiment: float):
        self.posts.append({
            "agent_id": agent_id, "content": content,
            "sentiment": sentiment,
            "timestamp": datetime.now().isoformat(),
        })

    def add_comment(self, agent_id: str, content: str, sentiment: float):
        self.comments.append({
            "agent_id": agent_id, "content": content,
            "sentiment": sentiment,
        })

    def add_like(self, agent_id: str, target_idx: int):
        self.likes.append({"agent_id": agent_id, "target": target_idx})

    def get_feed(self) -> list:
        return self.posts[-5:] if self.posts else []

    def get_stats(self) -> dict:
        return {
            "posts": len(self.posts),
            "comments": len(self.comments),
            "likes": len(self.likes),
        }


class PredictAgent:
    """预测 Agent——包含完整行为循环"""

    def __init__(self, agent_id: str, bias: float):
        self.agent_id = agent_id
        self.bias = bias          # 舆论倾向 (-1 到 1)
        self.memory = []          # 简化记忆
        self.actions_taken = 0

    async def run_cycle(self, platform: SocialPlatform, knowledge: str):
        """执行一个行为循环"""
        # PERCEIVE
        feed = platform.get_feed()
        # DELIBERATE
        sentiment = self.bias + random.uniform(-0.2, 0.2)
        sentiment = max(-1.0, min(1.0, sentiment))
        action = random.choices(
            ["post", "comment", "like"],
            weights=[0.4, 0.3, 0.3]
        )[0]
        # ACT
        if action == "post":
            content = f"[基于 {knowledge}] 我的看法是..."
            platform.add_post(self.agent_id, content, sentiment)
        elif action == "comment":
            if feed:
                content = "我同意/不同意这个观点..."
                platform.add_comment(self.agent_id, content, sentiment)
        elif action == "like":
            if platform.posts:
                idx = random.randint(0, len(platform.posts) - 1)
                platform.add_like(self.agent_id, idx)
        # PERSIST
        self.memory.append({"action": action, "sentiment": sentiment})
        self.actions_taken += 1


# ============================================================
# 知识点 4：ReportAgent
# ============================================================

class SimpleReportAgent:
    """简易报告智能体"""

    def __init__(self, platform: SocialPlatform, agents: list):
        self.platform = platform
        self.agents = agents

    def generate_report(self) -> dict:
        """生成预测报告"""
        stats = self.platform.get_stats()
        sentiments = [p["sentiment"] for p in self.platform.posts]
        avg_sentiment = sum(sentiments) / len(sentiments) if sentiments else 0

        report = {
            "total_posts": stats["posts"],
            "total_comments": stats["comments"],
            "total_likes": stats["likes"],
            "avg_sentiment": round(avg_sentiment, 2),
            "prediction": "正面" if avg_sentiment > 0.2 else
                          "负面" if avg_sentiment < -0.2 else "中性",
            "most_active": max(self.agents,
                               key=lambda a: a.actions_taken).agent_id,
        }

        print(f"\n{'=' * 60}")
        print("预测报告")
        print(f"{'=' * 60}")
        print(f"  总帖子: {report['total_posts']}")
        print(f"  总评论: {report['total_comments']}")
        print(f"  总点赞: {report['total_likes']}")
        print(f"  平均情感: {report['avg_sentiment']:+.2f} ({report['prediction']})")
        print(f"  最活跃 Agent: {report['most_active']}")
        return report

    def answer(self, question: str) -> str:
        """回答问题"""
        sentiments = [p["sentiment"] for p in self.platform.posts]
        if not sentiments:
            return "暂无足够数据回答此问题。"

        pos_count = sum(1 for s in sentiments if s > 0.2)
        neg_count = sum(1 for s in sentiments if s < -0.2)

        if "正面" in question or "负面" in question:
            return f"正面帖子: {pos_count}, 负面帖子: {neg_count}, " \
                   f"中性: {len(sentiments) - pos_count - neg_count}"
        elif "活跃" in question:
            most_active = max(self.agents, key=lambda a: a.actions_taken)
            return f"最活跃的是 {most_active.agent_id}，" \
                   f"共执行 {most_active.actions_taken} 个动作"
        else:
            avg = sum(sentiments) / len(sentiments)
            return f"整体舆论倾向: {avg:+.2f} " \
                   f"({'偏正面' if avg > 0 else '偏负面'})"


# ============================================================
# 知识点 1 + 5：五阶段流水线 + 多 LLM 配置
# ============================================================

class PredictionPipeline:
    """五阶段预测流水线"""

    def __init__(self, agent_count: int = 10, cycles: int = 5):
        self.agent_count = agent_count
        self.cycles = cycles
        self.platform = SocialPlatform()
        self.kg = KnowledgeGraph()
        self.agents = []
        self.report = None

    async def execute(self, topic: str):
        """执行完整流水线"""
        start = time.time()
        print("=" * 60)
        print("迷你舆论预测器")
        print(f"主题: {topic}")
        print(f"Agent 数量: {self.agent_count}, 周期: {self.cycles}")
        print("=" * 60)

        # 阶段 1-2：知识图谱构建
        print("\n[阶段 1-2] 知识图谱构建")
        kg_result = self.kg.build_from_topic(topic)
        print(f"  实体: {kg_result['entity_count']}, "
              f"关系: {kg_result['relation_count']}")
        knowledge = self.kg.get_context_for_agent()

        # 阶段 3：Agent 生成 + 并行模拟
        print(f"\n[阶段 3] 并行模拟")
        self.agents = [
            PredictAgent(f"Agent-{i+1:03d}",
                         bias=random.uniform(-0.5, 0.5))
            for i in range(self.agent_count)
        ]
        print(f"  生成 {len(self.agents)} 个 Agent")

        for cycle in range(1, self.cycles + 1):
            for agent in self.agents:
                await agent.run_cycle(self.platform, knowledge)
            stats = self.platform.get_stats()
            print(f"  周期 {cycle}/{self.cycles}: "
                  f"帖子 {stats['posts']}, 评论 {stats['comments']}")

        # 阶段 4-5：报告生成 + 深度交互
        print(f"\n[阶段 4-5] 报告生成 + 深度交互")
        reporter = SimpleReportAgent(self.platform, self.agents)
        self.report = reporter.generate_report()

        # 深度交互示例
        print(f"\n--- 深度交互示例 ---")
        print(f"Q: 正面和负面帖子各有多少？")
        print(f"A: {reporter.answer('正面和负面帖子各有多少？')}")
        print(f"Q: 谁最活跃？")
        print(f"A: {reporter.answer('谁最活跃？')}")

        elapsed = time.time() - start
        print(f"\n{'=' * 60}")
        print(f"流水线完成！耗时: {elapsed:.2f} 秒")
        print(f"{'=' * 60}")


# ============================================================
# 入口
# ============================================================

if __name__ == "__main__":
    random.seed(42)
    pipeline = PredictionPipeline(agent_count=10, cycles=5)
    asyncio.run(pipeline.execute("某品牌电动车上市后的舆论走向"))

# 运行实战项目
python3 mini-opinion-predictor.py

执行结果：

============================================================
迷你舆论预测器
主题: 某品牌电动车上市后的舆论走向
Agent 数量: 10, 周期: 5
============================================================

[阶段 1-2] 知识图谱构建
  实体: 8, 关系: 6

[阶段 3] 并行模拟
  生成 10 个 Agent
  周期 1/5: 帖子 5, 评论 0
  周期 2/5: 帖子 9, 评论 4
  周期 3/5: 帖子 13, 评论 7
  周期 4/5: 帖子 18, 评论 10
  周期 5/5: 帖子 22, 评论 13

[阶段 4-5] 报告生成 + 深度交互

============================================================
预测报告
============================================================
  总帖子: 22
  总评论: 13
  总点赞: 15
  平均情感: +0.08 (中性)
  最活跃 Agent: Agent-007

--- 深度交互示例 ---
Q: 正面和负面帖子各有多少？
A: 正面帖子: 8, 负面帖子: 7, 中性: 7
Q: 谁最活跃？
A: 最活跃的是 Agent-007，共执行 5 个动作

============================================================
流水线完成！耗时: 0.01 秒
============================================================

代码解析

扩展挑战

1. 添加情感分析：在 ReportAgent 中实现真实的情感分析（对帖子内容进行正面/负面/中性分类），生成情感时间序列图表。
2. 支持双平台模拟：添加 Reddit 类平台（嵌套评论 + 投票），与 Twitter 类平台并行运行，对比两种平台的舆论演变差异。
3. 实现敏感性分析：修改某个参数（如 Agent 数量、舆论倾向分布），重新运行模拟，对比两次结果的差异。

第五部分：常见问题与排查指南

常见错误及解决方案

调试技巧

1. 检查知识图谱质量：在模拟开始前，打印 GraphRAG 构建的实体和关系列表。检查是否有重要实体遗漏、关系类型是否合理。知识图谱的错误会在模拟中被放大——一个错误的关系可能导致整个 Agent 社区的行为偏离。

2. 追踪单个 Agent 的行为：选择 1-2 个 Agent，打印其每个周期的完整行为循环（感知到了什么、决策了什么、执行了什么动作、记忆了什么）。这是理解模拟结果最直接的方式。特别关注 Agent 的 DELIBERATE 阶段——决策逻辑是否合理。

3. 监控 LLM API 调用量和成本：记录每个阶段的 LLM API 调用次数和 token 消耗。如果发现某个阶段的调用量异常高（如 Agent DELIBERATE 阶段每周期调用次数远超预期），检查是否有 Agent 陷入重复调用循环。

第六部分：学习路线推荐

官方文档推荐阅读顺序

1. MiroFish GitHub README — 快速入门和安装指南。重点：五阶段流水线、Docker 部署、配置文件说明。
2. MiroFish 官方网站 — 产品定位和工作流程演示。重点：五阶段工作流程、Demo 体验、ReportAgent 交互示例。
3. OASIS 论文 (NeurIPS 2024) — 理解底层模拟引擎的学术基础。重点：双平台模拟架构、23 种社交动作原语、异步并发模型。
4. CAMEL-AI OASIS 框架文档 — OASIS 框架的技术细节。重点：Agent 创建、平台配置、动作原语定义。

推荐进阶资源

• Stanford Generative Agents 论文 (UIST 2023) — 社会模拟领域的开创性工作。理解 Generative Agents 的记忆流（Memory Stream）和反思（Reflection）机制，对比 MiroFish 的 Zep Cloud 记忆管理方案。
• GraphRAG 研究论文 — Microsoft 的 GraphRAG 实现。深入理解知识图谱构建、社区检测和层次化检索的技术原理。
• Zep Cloud 文档 — Agent 记忆管理的技术细节。重点：混合记忆策略、长期摘要压缩、语义检索机制。

信息来源与版本说明

• 教程基于： MiroFish 开源框架（GitHub 主分支，v0.1.0，截至 2026-04-13），OASIS 论文（NeurIPS 2024）
• 信息获取日期： 2026-04-13
• 信息来源列表：
• MiroFish GitHub 仓库 - 666ghj/MiroFish — 项目源码、README、安装指南、架构说明
• MiroFish 官方网站 — 产品定位、五阶段工作流程、Demo
• OASIS 论文 (NeurIPS 2024) - CAMEL-AI — OASIS 框架原理、双平台模拟、社交动作原语
• Medium - MiroFish Technical Deep Dive — 六层架构详解、GraphRAG 管线、性能数据
• Dev.to - MiroFish Swarm Intelligence — 部署指南、系统要求、工作流程说明