はじめに
ソーシャルメディアシミュレーションのために数百ものAIエージェントを構成するのは、膨大な作業に見えます。エージェントごとにアクティビティスケジュール、投稿頻度、応答遅延、影響力ウェイト、スタンスなど細かい調整が必要です。手動では現実的ではありません。
MiroFishは、LLMを活用した設定自動生成でこの作業を一括自動化します。システムはドキュメント、ナレッジグラフ、シミュレーション要件を解析し、各エージェントの詳細な設定を自動生成します。
ただし、LLMは出力が途中で切れる・JSONが壊れる・トークン制限にかかる等の失敗も発生します。
このガイドでは、以下の実装戦略を解説します：
- ステップバイステップの生成（時間 → イベント → エージェント → プラットフォーム）
- コンテキスト制限対策のバッチ処理
- 切り詰め出力のためのJSON修復
- LLM失敗時のルールベース・フォールバック
- タイプ別エージェントアクティビティパターン（学生/公式/メディアなど）
- 検証・修正ロジック
💡 構成生成パイプラインは100以上のエージェントをAPI経由で処理します。Apidogは各段階でのリクエスト/レスポンススキーマ検証、JSONエラーの早期検出、LLMの切り詰め対応テストに活用されています。
すべてのコードはMiroFishプロダクション環境の実例です。
アーキテクチャの概要
構成ジェネレータはパイプライン方式です。
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ コンテキスト │ ──► │ 時間設定 │ ──► │ イベント設定 │
│ ビルダー │ │ ジェネレータ │ │ ジェネレータ │
│ │ │ │ │ │
│ - シミュレーション │ │ - 総時間 │ │ - 初期投稿 │
│ 要件 │ │ - ラウンドあたりの分数 │ │ - ホットトピック │
│ - エンティティ概要 │ │ - ピーク時間 │ │ - 物語の方向性 │
│ - ドキュメントテキスト │ │ - 活動乗数 │ │ │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│
▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ 最終設定 │ ◄── │ プラットフォーム │ ◄── │ エージェント設定 │
│ アセンブリ │ │ 設定 │ │ バッチ │
│ │ │ │ │ │
│ - すべてマージ │ │ - Twitterパラメータ│ │ - バッチあたり15エージェント│
│ - 検証 │ │ - Redditパラメータ │ │ - Nバッチ │
│ - JSONを保存 │ │ - バイラルしきい値│ │ │
└─────────────────┘ └─────────────────┘ └─────────────────┘
ファイル構造
backend/app/services/
├── simulation_config_generator.py # 主要な設定生成ロジック
├── ontology_generator.py # オントロジー生成（共有）
└── zep_entity_reader.py # エンティティフィルタリング
backend/app/models/
├── task.py # タスクトラッキング
└── project.py # プロジェクト状態
ステップバイステップの生成戦略
全設定を一度に生成するとトークン制限を超えます。段階ごと・バッチごとにLLMを呼び出します。
class SimulationConfigGenerator:
# 各バッチは15エージェントの構成を生成します
AGENTS_PER_BATCH = 15
# コンテキスト制限
MAX_CONTEXT_LENGTH = 50000
TIME_CONFIG_CONTEXT_LENGTH = 10000
EVENT_CONFIG_CONTEXT_LENGTH = 8000
ENTITY_SUMMARY_LENGTH = 300
AGENT_SUMMARY_LENGTH = 300
ENTITIES_PER_TYPE_DISPLAY = 20
def generate_config(
self,
simulation_id: str,
project_id: str,
graph_id: str,
simulation_requirement: str,
document_text: str,
entities: List[EntityNode],
enable_twitter: bool = True,
enable_reddit: bool = True,
progress_callback: Optional[Callable[[int, int, str], None]] = None,
) -> SimulationParameters:
# 総ステップ数を計算
num_batches = math.ceil(len(entities) / self.AGENTS_PER_BATCH)
total_steps = 3 + num_batches # 時間 + イベント + N エージェントバッチ + プラットフォーム
current_step = 0
def report_progress(step: int, message: str):
nonlocal current_step
current_step = step
if progress_callback:
progress_callback(step, total_steps, message)
logger.info(f"[{step}/{total_steps}] {message}")
# コンテキストを構築
context = self._build_context(
simulation_requirement=simulation_requirement,
document_text=document_text,
entities=entities
)
reasoning_parts = []
# ステップ1: 時間設定を生成
report_progress(1, "時間設定を生成中...")
time_config_result = self._generate_time_config(context, len(entities))
time_config = self._parse_time_config(time_config_result, len(entities))
reasoning_parts.append(f"時間設定: {time_config_result.get('reasoning', '成功')}")
# ステップ2: イベント設定を生成
report_progress(2, "イベント設定とホットトピックを生成中...")
event_config_result = self._generate_event_config(context, simulation_requirement, entities)
event_config = self._parse_event_config(event_config_result)
reasoning_parts.append(f"イベント設定: {event_config_result.get('reasoning', '成功')}")
# ステップ3-N: エージェント設定をバッチで生成
all_agent_configs = []
for batch_idx in range(num_batches):
start_idx = batch_idx * self.AGENTS_PER_BATCH
end_idx = min(start_idx + self.AGENTS_PER_BATCH, len(entities))
batch_entities = entities[start_idx:end_idx]
report_progress(
3 + batch_idx,
f"エージェント設定を生成中 ({start_idx + 1}-{end_idx}/{len(entities)})..."
)
batch_configs = self._generate_agent_configs_batch(
context=context,
entities=batch_entities,
start_idx=start_idx,
simulation_requirement=simulation_requirement
)
all_agent_configs.extend(batch_configs)
reasoning_parts.append(f"エージェント設定: {len(all_agent_configs)}のエージェントを生成")
# 初期投稿の投稿者を割り当てる
event_config = self._assign_initial_post_agents(event_config, all_agent_configs)
# 最終ステップ: プラットフォーム設定
report_progress(total_steps, "プラットフォーム設定を生成中...")
twitter_config = PlatformConfig(platform="twitter", ...) if enable_twitter else None
reddit_config = PlatformConfig(platform="reddit", ...) if enable_reddit else None
# 最終構成を組み立てる
params = SimulationParameters(
simulation_id=simulation_id,
project_id=project_id,
graph_id=graph_id,
simulation_requirement=simulation_requirement,
time_config=time_config,
agent_configs=all_agent_configs,
event_config=event_config,
twitter_config=twitter_config,
reddit_config=reddit_config,
generation_reasoning=" | ".join(reasoning_parts)
)
return params
この分割アプローチにより：
- LLM呼び出しごとに集中した処理ができる
- 進捗状況の可視化が可能
- 途中エラーが出ても部分的に再実行・回復できる
コンテキストの構築
コンテキストビルダーはトークン制限内で必要情報を構成します。
def _build_context(
self,
simulation_requirement: str,
document_text: str,
entities: List[EntityNode]
) -> str:
# エンティティの要約
entity_summary = self._summarize_entities(entities)
context_parts = [
f"## シミュレーション要件\n{simulation_requirement}",
f"\n## エンティティ情報 ({len(entities)}エンティティ)\n{entity_summary}",
]
# スペースが許せばドキュメントテキストを追加
current_length = sum(len(p) for p in context_parts)
remaining_length = self.MAX_CONTEXT_LENGTH - current_length - 500 # 500文字のバッファ
if remaining_length > 0 and document_text:
doc_text = document_text[:remaining_length]
if len(document_text) > remaining_length:
doc_text += "\n...(ドキュメントが切り詰められました)"
context_parts.append(f"\n## 元のドキュメント\n{doc_text}")
return "\n".join(context_parts)
エンティティの要約
エンティティはタイプ別に要約してコンテキストに組み込みます。
def _summarize_entities(self, entities: List[EntityNode]) -> str:
lines = []
# タイプ別にグループ化
by_type: Dict[str, List[EntityNode]] = {}
for e in entities:
t = e.get_entity_type() or "不明"
if t not in by_type:
by_type[t] = []
by_type[t].append(e)
for entity_type, type_entities in by_type.items():
lines.append(f"\n### {entity_type} ({len(type_entities)}エンティティ)")
# 限られた要約長で限られた数を表示
display_count = self.ENTITIES_PER_TYPE_DISPLAY
summary_len = self.ENTITY_SUMMARY_LENGTH
for e in type_entities[:display_count]:
summary_preview = (e.summary[:summary_len] + "...") if len(e.summary) > summary_len else e.summary
lines.append(f"- {e.name}: {summary_preview}")
if len(type_entities) > display_count:
lines.append(f" ...その他 {len(type_entities) - display_count}件")
return "\n".join(lines)
例：
### 学生 (45エンティティ)
- Zhang Wei: 学生組合で活動し、キャンパスのイベントや学業のプレッシャーについて頻繁に投稿...
- Li Ming: AI倫理を研究している大学院生で、テクノロジーニュースをよく共有...
...その他 43件
### 大学 (3エンティティ)
- Wuhan University: 公式アカウント、お知らせやニュースを投稿...
時間設定の生成
シミュレーション全体の期間・アクティビティパターンを自動生成します。
def _generate_time_config(self, context: str, num_entities: int) -> Dict[str, Any]:
# この特定のステップのためにコンテキストを切り詰める
context_truncated = context[:self.TIME_CONFIG_CONTEXT_LENGTH]
max_agents_allowed = max(1, int(num_entities * 0.9))
prompt = f"""...（省略。詳細は上記コードの通り）..."""
system_prompt = "あなたはソーシャルメディアシミュレーションの専門家です。純粋なJSON形式で返してください。"
try:
return self._call_llm_with_retry(prompt, system_prompt)
except Exception as e:
logger.warning(f"時間設定LLM生成失敗: {e}、デフォルトを使用します")
return self._get_default_time_config(num_entities)
時間設定の解析と検証
生成値が矛盾しないか検証・修正します。
def _parse_time_config(self, result: Dict[str, Any], num_entities: int) -> TimeSimulationConfig:
agents_per_hour_min = result.get("agents_per_hour_min", max(1, num_entities // 15))
agents_per_hour_max = result.get("agents_per_hour_max", max(5, num_entities // 5))
if agents_per_hour_min > num_entities:
logger.warning(f"agents_per_hour_min ({agents_per_hour_min}) が総エージェント数 ({num_entities}) を超えています。修正済み")
agents_per_hour_min = max(1, num_entities // 10)
if agents_per_hour_max > num_entities:
logger.warning(f"agents_per_hour_max ({agents_per_hour_max}) が総エージェント数 ({num_entities}) を超えています。修正済み")
agents_per_hour_max = max(agents_per_hour_min + 1, num_entities // 2)
if agents_per_hour_min >= agents_per_hour_max:
agents_per_hour_min = max(1, agents_per_hour_max // 2)
logger.warning(f"agents_per_hour_min >= max です。{agents_per_hour_min}に修正済み")
return TimeSimulationConfig(
total_simulation_hours=result.get("total_simulation_hours", 72),
minutes_per_round=result.get("minutes_per_round", 60),
agents_per_hour_min=agents_per_hour_min,
agents_per_hour_max=agents_per_hour_max,
peak_hours=result.get("peak_hours", [19, 20, 21, 22]),
off_peak_hours=result.get("off_peak_hours", [0, 1, 2, 3, 4, 5]),
off_peak_activity_multiplier=0.05,
morning_activity_multiplier=0.4,
work_activity_multiplier=0.7,
peak_activity_multiplier=1.5
)
デフォルトの時間設定（中国タイムゾーン）
def _get_default_time_config(self, num_entities: int) -> Dict[str, Any]:
return {
"total_simulation_hours": 72,
"minutes_per_round": 60,
"agents_per_hour_min": max(1, num_entities // 15),
"agents_per_hour_max": max(5, num_entities // 5),
"peak_hours": [19, 20, 21, 22],
"off_peak_hours": [0, 1, 2, 3, 4, 5],
"morning_hours": [6, 7, 8],
"work_hours": [9, 10, 11, 12, 13, 14, 15, 16, 17, 18],
"reasoning": "デフォルトの中国タイムゾーン設定を使用しています"
}
イベント設定の生成
初期投稿・ホットトピック・物語方向性などイベント構成を自動生成します。
def _generate_event_config(
self,
context: str,
simulation_requirement: str,
entities: List[EntityNode]
) -> Dict[str, Any]:
... # 詳細は上記コードの通り
初期投稿の投稿者の割り当て
初期投稿ごとに実際のエージェントIDを割り当てます。
def _assign_initial_post_agents(
self,
event_config: EventConfig,
agent_configs: List[AgentActivityConfig]
) -> EventConfig:
... # 詳細は上記コードの通り
バッチエージェント設定の生成
エージェント設定は15件ずつバッチでLLM生成・ルールベース生成を併用します。
def _generate_agent_configs_batch(
self,
context: str,
entities: List[EntityNode],
start_idx: int,
simulation_requirement: str
) -> List[AgentActivityConfig]:
... # 詳細は上記コードの通り
ルールベースのフォールバック設定
LLM失敗時は下記パターンで自動補完します。
def _generate_agent_config_by_rule(self, entity: EntityNode) -> Dict[str, Any]:
... # 詳細は上記コードの通り
リトライとJSON修復を伴うLLM呼び出し
LLMの出力エラー（切り詰め・JSON壊れ）を自動処理します。
def _call_llm_with_retry(self, prompt: str, system_prompt: str) -> Dict[str, Any]:
... # 詳細は上記コードの通り
切り詰められたJSONの修正
def _fix_truncated_json(self, content: str) -> str:
... # 詳細は上記コードの通り
高度なJSON修復
def _try_fix_config_json(self, content: str) -> Optional[Dict[str, Any]]:
... # 詳細は上記コードの通り
設定データ構造
エージェントアクティビティ設定
@dataclass
class AgentActivityConfig:
"""単一エージェントのアクティビティ設定"""
agent_id: int
entity_uuid: str
entity_name: str
entity_type: str
activity_level: float = 0.5
posts_per_hour: float = 1.0
comments_per_hour: float = 2.0
active_hours: List[int] = field(default_factory=lambda: list(range(8, 23)))
response_delay_min: int = 5
response_delay_max: int = 60
sentiment_bias: float = 0.0
stance: str = "neutral"
influence_weight: float = 1.0
時間シミュレーション設定
@dataclass
class TimeSimulationConfig:
"""時間シミュレーション設定（中国タイムゾーン）"""
total_simulation_hours: int = 72
minutes_per_round: int = 60
agents_per_hour_min: int = 5
agents_per_hour_max: int = 20
peak_hours: List[int] = field(default_factory=lambda: [19, 20, 21, 22])
peak_activity_multiplier: float = 1.5
off_peak_hours: List[int] = field(default_factory=lambda: [0, 1, 2, 3, 4, 5])
off_peak_activity_multiplier: float = 0.05
morning_hours: List[int] = field(default_factory=lambda: [6, 7, 8])
morning_activity_multiplier: float = 0.4
work_hours: List[int] = field(default_factory=lambda: [9, 10, 11, 12, 13, 14, 15, 16, 17, 18])
work_activity_multiplier: float = 0.7
完全なシミュレーションパラメータ
@dataclass
class SimulationParameters:
"""完全なシミュレーションパラメータ構成"""
simulation_id: str
project_id: str
graph_id: str
simulation_requirement: str
time_config: TimeSimulationConfig = field(default_factory=TimeSimulationConfig)
agent_configs: List[AgentActivityConfig] = field(default_factory=list)
event_config: EventConfig = field(default_factory=EventConfig)
twitter_config: Optional[PlatformConfig] = None
reddit_config: Optional[PlatformConfig] = None
llm_model: str = ""
llm_base_url: str = ""
generated_at: str = field(default_factory=lambda: datetime.now().isoformat())
generation_reasoning: str = ""
def to_dict(self) -> Dict[str, Any]:
time_dict = asdict(self.time_config)
return {
"simulation_id": self.simulation_id,
"project_id": self.project_id,
"graph_id": self.graph_id,
"simulation_requirement": self.simulation_requirement,
"time_config": time_dict,
"agent_configs": [asdict(a) for a in self.agent_configs],
"event_config": asdict(self.event_config),
"twitter_config": asdict(self.twitter_config) if self.twitter_config else None,
"reddit_config": asdict(self.reddit_config) if self.reddit_config else None,
"llm_model": self.llm_model,
"llm_base_url": self.llm_base_url,
"generated_at": self.generated_at,
"generation_reasoning": self.generation_reasoning,
}
要約表：エージェントタイプ別パターン
| エージェントタイプ | 活動レベル | 活動時間 | 投稿数/時間 | コメント数/時間 | 応答時間 (分) | 影響力 |
|---|---|---|---|---|---|---|
| 大学 | 0.2 | 9-17 | 0.1 | 0.05 | 60-240 | 3.0 |
| 政府機関 | 0.2 | 9-17 | 0.1 | 0.05 | 60-240 | 3.0 |
| メディアアウトレット | 0.5 | 7-23 | 0.8 | 0.3 | 5-30 | 2.5 |
| 教授 | 0.4 | 8-21 | 0.3 | 0.5 | 15-90 | 2.0 |
| 学生 | 0.8 | 8-12, 18-23 | 0.6 | 1.5 | 1-15 | 0.8 |
| 卒業生 | 0.6 | 12-13, 19-23 | 0.4 | 0.8 | 5-30 | 1.0 |
| 個人 (デフォルト) | 0.7 | 9-13, 18-23 | 0.5 | 1.2 | 2-20 | 1.0 |
結論
LLMを用いた構成自動化では、以下のアプローチが実用的です：
- ステップバイステップ生成：各構成（時間→イベント→エージェント→プラットフォーム）を分割
- バッチ処理：15件単位でエージェントを生成し、トークン制限を回避
- JSON修復：括弧・エスケープ等の修正で壊れた出力も受容
- ルールベース・フォールバック：LLM失敗時も確実に構成を生成
- タイプ別パターン：エージェントタイプごとの活動傾向を明確化
- 検証と修正：生成値を必ずチェック・問題があれば自動修正
この構成パイプラインを活用することで、大規模エージェントシミュレーションも短時間で実装可能です。
Top comments (0)