forecast-runtime

name	forecast-runtime
description	基于校正后状态和历史经验的滚动舆情预测引擎。编排 EvoSim 轻量仿真循环, 为 Leader Actor 注入因果经验辅助 LLM 行为决策,通过可选干预 Agent 实时稳定仿真轨迹, 最终由 Report Agent 输出结构化 JSON + 人类可读 Markdown 双格式预测报告。 当需要执行以下任务时使用此 Skill: (1) 在校正后的仿真快照上继续向未来推演 (2) 输出立场分布、极化、关键人物活跃度、传播结构的变化趋势 (3) 基于历史因果经验(原因导致结果)辅助预测并稳定仿真 (4) 生成供人类阅读的预测报告和供下游 Skill 消费的结构化数据 此 Skill 为 EvoPalantir 闭环中的 forecast_runtime 模块, 上游为 calibration_engine 输出的校正状态,下游为 calibration_memory_engine 的经验沉淀。
compatibility	Requires Python 3.10+, EvoSim project, OpenAI-compatible LLM API, and optionally a vector database (Qdrant/Milvus/pgvector).
metadata	{"author":"MaYiding","version":"1.0"}

Forecast Runtime

Overview

在校正后的仿真状态上执行滚动预测。

核心架构决策: 本 Skill 使用外部逐 tick 编排模式,不调用 EvoSim 的 sim.run()。 原因: 需要在每个 tick 的 Leader 行为决策前注入经验到 prompt,这要求拦截 tick 内部步骤, sim.run() 作为黑盒无法实现此拦截。因此本 Skill 直接调用 EvoSim 的底层原语 (AgentUser.get_feed(), AgentUser.react_to_feed(), FeedPipeline, SnapshotManager 等)。

数据流:

校正后 Snapshot + ActorStates
  -> 恢复 EvoSim DB + 初始化组件
  -> 外部 tick loop:
       获取经验 -> Leader: 经验增强 prompt + LLM 决策
       Crowd: 规则决策 -> 推荐器分发 -> 状态更新
       -> 指标计算 -> [干预检查] -> [快照]
  -> Report Agent (ReACT) 生成报告
  -> 输出: JSON + Markdown + DB

Input Parameters

调用时传入 ForecastInput (完整 schema 见 data-schema.md):

参数	必填	默认值	说明
event_id	是	-	目标事件 ID
db_url	是	-	数据库连接串, 如 sqlite:///path/to/sim.db
project_root	是	-	EvoSim 项目根目录, SnapshotManager 需要
forecast_horizon_ticks	是	-	预测跨度 (tick 数)
tick_interval_minutes	是	-	每 tick 代表的真实分钟数
llm_config	是	-	LLM API 配置 (leader/intervention/report 三模型可分开指定)
output_dir	是	-	输出目录路径
parent_session_id	否	自动取最新	EvoSim 快照 session ID
vector_db_config	否	None	向量 DB 配置
intervention_enabled	否	true	是否启用干预 Agent
intervention_check_interval_ticks	否	3	干预检查间隔
snapshot_interval_ticks	否	5	快照保存间隔
experience_refresh_strategy	否	auto	auto/fixed/on_demand
experience_cache_ttl_ticks	否	5	经验缓存基础 TTL
recommender_config	否	见 Config 节	推荐权重
output_to_db	否	true	是否写入数据库

Orchestration Workflow

Phase 1: 初始化

import json, os, sys, uuid, random, asyncio
from datetime import datetime
from openai import OpenAI
from scripts.state_loader import StateLoader
from scripts.experience_retriever import (
    ExperienceRetriever, VectorDBConfig, SQLConfig,
    build_experience_enhanced_prompt, query_user_actions,
    query_user_actions_max_id, _ACTION_MAP,
)
from scripts.experience_cache import ExperienceCache, TickMetricsSnapshot
from scripts.metrics_calculator import MetricsCalculator
from scripts.forecast_writer import ForecastWriter, build_summary_from_timeline

# ── 0. 设置 EvoSim 导入路径 ──
# EvoSim 内部使用裸导入 (如 "import control_flags"), 需将 src/ 加入 sys.path
evosim_root = forecast_input["project_root"]
evosim_src = os.path.join(evosim_root, "src")
for p in [evosim_root, evosim_src]:
    if p not in sys.path:
        sys.path.insert(0, p)

# ── 1. 加载校正后状态 ──
loader = StateLoader(db_url=forecast_input["db_url"])
state = loader.load_all(forecast_input["event_id"])
# state: LoadedState(event, actor_states, snapshot, start_tick, metadata)
if len(state.actor_states) == 0:
    raise ValueError("No actor states found. Ensure calibration_engine has run.")
event = state.event
start_tick = state.start_tick

# ── 2. 分离 Leader 和 Crowd ──
leader_actors = [a for a in state.actor_states if a.get("role") == "leader"]
crowd_actors = [a for a in state.actor_states if a.get("role") == "crowd"]
# leader_actors: 8 个, crowd_actors: 22 个

# ── 3. 禁用 EvoSim 攻防模块 ──
# control_flags 是模块级变量 (不是 class), 使用 "import control_flags" 后直接赋值
import control_flags
control_flags.attack_enabled = False
control_flags.aftercare_enabled = False
control_flags.moderation_enabled = False

# ── 4. 恢复 EvoSim 快照 ──
from snapshot_manager import create_snapshot_manager
db_path = forecast_input["db_url"].replace("sqlite:///", "")  # 提取文件路径
snapshot_mgr = create_snapshot_manager(forecast_input["project_root"], db_path)
session_id = forecast_input.get("parent_session_id")
if session_id is None:
    sessions = snapshot_mgr.list_sessions()
    session_id = sessions[0]["session_id"] if sessions else None
if session_id:
    snapshot_mgr.restore_from_tick(tick=start_tick, session_id=session_id)
# 重要: restore_from_tick 不设置 session_id, 必须显式创建新 session 才能保存快照
forecast_session_id = snapshot_mgr.create_session(
    parent_session_id=session_id, parent_tick=start_tick
)

# ── 5. 初始化 EvoSim 底层组件 (不调用 Simulation()) ──
# EvoSim 有两个 DatabaseManager:
#   - src/database_manager.py: UserManager 用, 有 get_connection()
#   - src/database/database_manager.py: AgentUser 内部单例, 有 set_database_path()
# UserManager.__init__ 调用 db_manager.get_connection(), 必须用 src/database_manager 的实例

# (a) 为 UserManager 创建直连 DatabaseManager
from database_manager import DatabaseManager
db_mgr = DatabaseManager(db_path, reset_db=False, use_service=False)

# (b) 设置 AgentUser 内部单例的路径 (AgentUser 构造时会用到)
from database.database_manager import get_db_manager as get_singleton_db
_singleton = get_singleton_db()
_singleton.set_database_path(db_path)

from user_manager import UserManager
user_mgr = UserManager(
    config={"num_users": len(state.actor_states), "temperature": 0.8,
            "agent_config_path": "separate"},
    db_manager=db_mgr,          # src/database_manager.DatabaseManager 实例
    restore_existing=True        # 从 DB 加载已有 AgentUser 对象
)
# user_mgr.users: List[AgentUser], 已恢复的用户对象

# ── 6. 创建 OpenAI 客户端 ──
# react_to_feed(openai_client, engine, feed) 需要这两个参数
llm_cfg = forecast_input["llm_config"]
# api_key: 优先读取 llm_config.api_key_env 指定的环境变量, 回退到 LLM_API_KEY
api_key_env = llm_cfg.get("api_key_env", "LLM_API_KEY")
openai_client = OpenAI(
    api_key=os.environ.get(api_key_env, ""),
    base_url=os.environ.get("LLM_API_BASE", llm_cfg.get("api_base", ""))
)
leader_engine = llm_cfg.get("leader_model", "gpt-4")

# ── 7. 初始化经验系统 ──
retriever = ExperienceRetriever(
    vector_config=VectorDBConfig(**forecast_input["vector_db_config"])
        if forecast_input.get("vector_db_config") else None,
    sql_config=SQLConfig(db_url=forecast_input["db_url"])
)
cache = ExperienceCache(
    retriever, base_ttl_ticks=forecast_input.get("experience_cache_ttl_ticks", 5)
)

# ── 8. 初始化指标和输出 ──
calc = MetricsCalculator()
writer = ForecastWriter(
    output_dir=forecast_input["output_dir"],
    db_url=forecast_input["db_url"] if forecast_input.get("output_to_db", True) else None
)

# ── 9. 确保 EvoPalantir 扩展表存在 ──
import sqlite3
_conn = sqlite3.connect(db_path)
_conn.execute("""CREATE TABLE IF NOT EXISTS actor_states (
    actor_id TEXT PRIMARY KEY, event_id TEXT NOT NULL, role TEXT,
    stance_posterior TEXT, activity_hazard REAL DEFAULT 0.5,
    bridge_score REAL DEFAULT 0.0, topic_mix TEXT,
    uncertainty REAL DEFAULT 0.5, last_observed_at TEXT,
    state_version INTEGER DEFAULT 0, updated_at_tick INTEGER DEFAULT 0)""")
_conn.execute("""CREATE TABLE IF NOT EXISTS intervention_records (
    intervention_id TEXT PRIMARY KEY, event_id TEXT NOT NULL,
    tick INTEGER, trigger_reason TEXT, deviation_score REAL,
    severity TEXT, adjustments_json TEXT, created_at TEXT)""")
_conn.commit()
_conn.close()

# ── 10. 初始化 forecast_output 对象 ──
end_tick = start_tick + forecast_input["forecast_horizon_ticks"]
forecast_output = {
    "forecast_id": str(uuid.uuid4()),
    "event_id": forecast_input["event_id"],
    "event_title": event.get("title", forecast_input["event_id"]),
    "generated_at": datetime.now().isoformat(),
    "forecast_horizon_ticks": forecast_input["forecast_horizon_ticks"],
    "tick_interval_minutes": forecast_input["tick_interval_minutes"],
    "start_tick": start_tick,
    "end_tick": end_tick,
    "timeline": [],            # Phase 2 每 tick 追加
    "summary": {},             # Phase 3 Report Agent 填充
    "experience_used": [],     # Phase 2 收集
}
intervention_records = []      # Phase 2 收集
experience_used_set = {}       # experience_id -> {experience_id, event_similarity, ...}
prev_phase = event.get("phase", "")
intervention_count = 0
max_interventions = forecast_input["forecast_horizon_ticks"] // 3

Phase 2: Tick 循环

整个 tick 循环包裹在 async 函数中,因为 react_to_feed() 是 async 方法。 外层使用 asyncio.run(run_forecast_loop()) 驱动。

async def run_forecast_loop():
  """async 包装: react_to_feed 是 async 方法, 必须在 async 上下文中调用。"""

  for tick in range(start_tick + 1, end_tick + 1):
    tick_actions = []  # 本 tick 所有 action, 格式见下方

    # ── Step 1: 获取/刷新经验 ──
    # 缓存返回 top_k=5, 但 Leader prompt 仅注入前 2 条, 干预 Agent 用前 3 条
    # (截断在 build_experience_enhanced_prompt 内部完成)
    experiences = cache.get_or_fetch(
        event=event, current_tick=tick,
        actors=leader_actors,  # 仅传入 leader, 用于计算 actor_memory_similarity
        top_k=5
    )
    for exp in experiences:
        if exp.experience_id not in experience_used_set:
            experience_used_set[exp.experience_id] = {
                "experience_id": exp.experience_id,
                "event_type": exp.event_type,
                "event_similarity": exp.event_similarity,
                "influence_on_forecast": exp.effect_description[:100]
            }

    # ── Step 2: Leader 行为决策 (LLM + 经验注入) ──
    # 所有 8 个 Leader 每 tick 强制参与, 不做概率采样
    for user in user_mgr.users:
        actor_state = next((a for a in leader_actors if a["actor_id"] == user.user_id), None)
        if actor_state is None:
            continue  # 不是 leader, 跳过 (crowd 在 Step 3 处理)

        # 获取 feed (复用 EvoSim 7 阶段 FeedPipeline)
        feed = user.get_feed(experiment_config={}, time_step=tick)

        # 临时注入经验到 persona (详见 experience-system.md#injection-format)
        original_persona = user.persona
        user.persona = build_experience_enhanced_prompt(
            persona=original_persona, feed=feed,
            experiences=experiences, actor_state=actor_state
        )

        # 记录 user_actions 表当前最大 ID, 用于后续提取本次行动
        _max_id_before = query_user_actions_max_id(user.user_id, db_path)

        # LLM 决策 -> FeedReaction(actions: List[FeedAction])
        try:
            await user.react_to_feed(openai_client, leader_engine, feed)
        except Exception:
            pass  # fallback: 本 tick 该 leader 无行动

        user.persona = original_persona  # 恢复

        # 从 user_actions 表提取本次调用新增的行动 (id > _max_id_before)
        # 统一 action dict 格式 (MetricsCalculator 期望):
        # {"actor_id": str, "action_type": "comment|repost|like|silence",
        #  "stance_expressed": "support|oppose|neutral|unclear",
        #  "target_content_id": str|None, "target_actor_id": str|None}
        user_tick_actions = query_user_actions(user.user_id, db_path,
                                               after_id=_max_id_before)
        tick_actions.extend(user_tick_actions)

    # ── Step 3: Crowd 行为决策 (规则驱动) ──
    # 完整决策逻辑详见 simulation-guide.md#crowd-behavior-rules
    # 注意: 活跃度概率采样仅在此处执行一次, crowd_decide 内部不再重复检查
    for actor_state in crowd_actors:
        if random.random() > actor_state.get("activity_hazard", 0.3):
            continue  # 本 tick 不活跃

        # Crowd 也需要 feed 来选择目标内容
        crowd_user = next(
            (u for u in user_mgr.users if u.user_id == actor_state["actor_id"]), None
        )
        crowd_feed = crowd_user.get_feed({}, time_step=tick) if crowd_user else []

        if not crowd_feed:
            continue  # 无内容可交互

        # 规则决策: 根据立场对齐度选择行动类型
        # stance_posterior 是 dict: {"support": 0.4, "oppose": 0.3, ...}
        stance = actor_state.get("stance_posterior", {"neutral": 1.0})
        dominant_stance = max(stance, key=stance.get)

        target = crowd_feed[0] if crowd_feed else None  # 取 feed 首条
        target_id = target.get("post_id") if isinstance(target, dict) else getattr(target, "post_id", None)

        # 根据立场对齐度决定行为 (简化规则)
        alignment = stance.get("support", 0) - stance.get("oppose", 0)
        if alignment > 0.3:
            action_type = random.choices(
                ["comment", "repost", "like"], weights=[0.4, 0.3, 0.3]
            )[0]
        elif alignment < -0.3:
            action_type = random.choices(
                ["comment", "silence"], weights=[0.3, 0.7]
            )[0]
        else:
            action_type = random.choices(
                ["like", "silence"], weights=[0.3, 0.7]
            )[0]

        tick_actions.append({
            "actor_id": actor_state["actor_id"],
            "action_type": action_type,
            "stance_expressed": dominant_stance,
            "target_content_id": target_id,
            "target_actor_id": None,
        })

    # ── Step 4: 更新 Actor 状态 ──
    # 根据本 tick 行动更新 stance_posterior (渐进混合)
    # Leader: alpha=0.15, Crowd: alpha=0.08
    for actor_list, alpha in [(leader_actors, 0.15), (crowd_actors, 0.08)]:
        for actor in actor_list:
            actor_actions = [a for a in tick_actions if a["actor_id"] == actor["actor_id"]]
            if not actor_actions:
                continue
            # 取本 tick 最显著的 stance 表达
            expressed = actor_actions[0].get("stance_expressed", "neutral")
            old_stance = actor.get("stance_posterior", {"neutral": 1.0})
            # 渐进混合: new = (1-alpha)*old + alpha*one_hot(expressed)
            new_stance = {}
            for k in ["support", "oppose", "neutral", "unclear"]:
                one_hot = 1.0 if k == expressed else 0.0
                new_stance[k] = (1 - alpha) * old_stance.get(k, 0.0) + alpha * one_hot
            # 归一化
            total = sum(new_stance.values()) or 1.0
            new_stance = {k: v / total for k, v in new_stance.items()}
            actor["stance_posterior"] = new_stance
            actor["activity_hazard"] = max(0.05, actor.get("activity_hazard", 0.3) * 0.95)
            actor["state_version"] = actor.get("state_version", 0) + 1
            actor["updated_at_tick"] = tick

    # ── Step 5: 计算 tick 指标 ──
    all_actors = leader_actors + crowd_actors
    metrics = calc.calculate(tick, all_actors, tick_actions)

    # 构建 TickMetricsSnapshot 传给缓存 (类型转换)
    metrics_snapshot = TickMetricsSnapshot(
        tick=tick,
        stance_distribution=metrics.stance_distribution,
        polarization_index=metrics.polarization_index,
        activity_rate=metrics.activity_rate,
        has_new_injection=False  # 如果本 tick 有外部注入, 设为 True
    )
    cache.record_tick_metrics(metrics_snapshot)

    # ── Step 6: 检查 event.phase 变化 ──
    # event.phase 可能被外部进程 (如 real_world_observer) 修改
    # 若外部无修改, 此处不会触发
    current_phase = event.get("phase", "")
    if current_phase != prev_phase:
        cache.invalidate_on_phase_change()
        prev_phase = current_phase

    # ── Step 7: 干预检查 ──
    intervention_applied = {"applied": False, "reason": "", "adjustments": {}}
    # 使用 (tick - start_tick) 偏移, 确保首次检查在积累足够 tick 后
    if (forecast_input.get("intervention_enabled", True)
        and (tick - start_tick) % forecast_input.get("intervention_check_interval_ticks", 3) == 0
        and (tick - start_tick) > 0
        and intervention_count < max_interventions):
        # 详见 intervention-agent.md
        # 计算偏差: calculate_deviation(current_metrics, expected_from_experience)
        # 构建 expected: 取经验中 effect_stance_shift 叠加到当前基准
        # 偏差 >= 0.30 OR 存在趋势异常 -> LLM 判断
        # 如果干预: 调整推荐权重/活跃度/立场, 记录 InterventionRecord
        # 干预后: cache.invalidate_on_intervention()
        pass  # 具体实现见 intervention-agent.md

    # ── Step 8: 保存快照 ──
    if tick % forecast_input.get("snapshot_interval_ticks", 5) == 0:
        snapshot_mgr.save_tick_snapshot(tick, {
            "tick": tick, "timestamp": datetime.now().isoformat(),
            "user_count": len(state.actor_states)
        })

    # ── Step 9: 记录 tick 到 timeline ──
    forecast_output["timeline"].append({
        "tick": tick,
        "simulated_time": datetime.now().isoformat(),
        "metrics": metrics.to_dict(),
        "notable_actions": tick_actions[:10],  # 取前10个
        "intervention_applied": intervention_applied
    })

# 驱动 async 循环 (带异常保护, 确保部分 timeline 可进入 Phase 3)
try:
    asyncio.run(run_forecast_loop())
except Exception as e:
    forecast_output.setdefault("summary", {}).setdefault("risk_signals", []).append(
        "simulation_incomplete"
    )
    # 记录异常但不中断, 继续使用已收集的 timeline 进入 Phase 3

action dict 转换说明: EvoSim 的 react_to_feed() 将行为写入 user_actions 表时 使用 action.replace('-', '_') (如 "like_post", "comment_post")。 query_user_actions() 通过 _ACTION_MAP 转换为 MetricsCalculator 统一格式 (如 "like", "comment", "repost", "silence")。

Phase 3: 报告生成

Report Agent 使用 ReACT 模式 (详见 report-guide.md):

# 1. 规划大纲 (必含: 总体走势概览, 关键人物动态, 经验对比与验证)
# 2. 逐 section ReACT 循环 (3-5 轮工具调用)
#    工具均为本地函数, 操作内存中的 forecast_output 和 experience_used_set:
#      QueryTimeline(start_tick, end_tick, fields) -> timeline 子集
#      QueryActorHistory(actor_id, fields) -> 该 actor 的 action 序列
#      QueryMetricsTrend(metric, start_tick, end_tick) -> 指标时间序列
#      SearchExperience(query, top_k) -> 经验条目
#      CompareWithExperience(experience_id, metrics) -> 对比表
#      GetInterventionLog() -> intervention_records 列表
#    工具签名详见 report-guide.md#available-tools
# 3. 生成 summary JSON
# 4. 组装 report_sections

report_sections = []  # list[{"title": str, "content": str}]
# 每个 section 由 Report Agent ReACT 循环产出:
# report_sections.append({"title": "总体走势概览", "content": "..."})

# 生成 summary (如果 Report Agent 未成功填充, 从 timeline 数据回退构建):
# 注意: summary 初始化为 {}, 是 falsy; 需检查是否包含实际内容
if "stance_trend" not in forecast_output.get("summary", {}):
    # 回退: 从 timeline 提取 stance/polarization/activity 趋势
    forecast_output["summary"] = build_summary_from_timeline(
        forecast_output["timeline"]
    )
    forecast_output["summary"].setdefault("risk_signals", []).append(
        "summary_from_fallback"
    )

experience_used_list = list(experience_used_set.values())
forecast_output["experience_used"] = experience_used_list

Phase 4: 输出

# 写入 JSON
json_path = writer.write_json(forecast_output)

# 写入 Markdown 报告
md_path = writer.write_markdown(
    forecast_output, report_sections,
    experience_used=experience_used_list,
    intervention_log=intervention_records
)

# 写入数据库
if forecast_input.get("output_to_db", True):
    writer.write_to_db(forecast_output)

# 干预记录写入 intervention_records 表 (通过直接 SQL)
# per-tick 简要信息在 timeline[].intervention_applied
# 完整 InterventionRecord 在 intervention_records 表

输出文件:

• {output_dir}/forecast_output_{event_id}_{timestamp}.json
• {output_dir}/forecast_report_{event_id}_{timestamp}.md

Configuration Quick Reference

推荐器权重

ForecastInput.recommender_config 使用 _weight 后缀 key:

similar_feed_weight: 0.5
cross_cutting_weight: 0.2
trending_weight: 0.3
leader_comment_share: 0.7
crowd_comment_share: 0.3

干预 Agent 的 InterventionRecord.adjustments.recommender_weights 使用短 key:

similar_feed, cross_cutting, trending

翻译映射: similar_feed -> similar_feed_weight, cross_cutting -> cross_cutting_weight, trending -> trending_weight。干预后需将短 key 翻译回 _weight key 更新推荐器配置。

LLM Temperature

Leader Actor: 0.7     # 保留行为多样性
Intervention Agent: 0.3  # 保守/确定
Report Agent: 0.4     # 稳定/可复现

干预阈值

偏差 >= 0.30 或存在趋势异常 -> 触发 LLM 判断 (OR 关系)。 过度干预保护: intervention_count >= forecast_horizon_ticks // 3 时停止。

经验缓存

base_ttl: 5 ticks (高波动减半, 低波动加倍)
立即失效: 注入新内容 / 干预执行 / event.phase 变更 / on_demand 策略

Scripts

脚本	用途	何时执行
state_loader.py	加载校正后状态	Phase 1
experience_retriever.py	混合检索经验 (向量+SQL+RRF)	Phase 2 Step 1
experience_cache.py	经验缓存与失效管理	Phase 2 全程
metrics_calculator.py	计算 tick 指标	Phase 2 Step 5
forecast_writer.py	输出 JSON + Markdown + DB	Phase 4

References

文档	内容	何时查阅
data-schema.md	所有数据结构定义	需要了解字段含义时
experience-system.md	经验检索、缓存、注入 prompt 格式	Phase 2 Step 1-2
simulation-guide.md	EvoSim API、Leader/Crowd 行为、推荐管线	Phase 2 全程
intervention-agent.md	干预监控、手柄、策略、prompt	Phase 2 Step 7
report-guide.md	Report Agent ReACT、工具、prompt	Phase 3

Error Handling

场景	处理方式
actor_states 为空	Phase 1 抛出 ValueError 终止
EvoSim import 失败	捕获 ImportError, 报告缺失模块名
经验库不可用	Leader prompt 省略经验 section, 干预仅用硬性指标
LLM 调用失败	Leader 本 tick 无行动 (try/except), 不中断
tick 循环异常	用已收集 timeline 进入 Phase 3, risk_signals 标记 simulation_incomplete
Phase 3 summary LLM 失败	从 timeline 数据回退构建, risk_signals 标记 summary_from_fallback
过度干预	超过 horizon//3 次停止, risk_signals 标记 excessive_intervention
输出写入失败	备用路径 {output_dir}/fallback/, 仍失败返回内存结果

Environment Variables

LLM_API_KEY        # OpenAI 兼容 API key
LLM_API_BASE       # API base URL
VECTOR_DB_API_KEY  # 向量数据库 API key (如需要)
SQL_DB_URL         # 经验库 SQL 连接串