Introduction
Configuring hundreds of AI agents for a social media simulation is difficult: every agent needs an activity schedule, posting frequency, response delay, influence weight, and topic stance.
MiroFish automates this work with LLM-powered configuration generation. It analyzes documents, a knowledge graph, and simulation requirements, then generates detailed settings for every agent.
The main implementation challenge is reliability. LLM responses can be truncated, malformed, or limited by context windows. A production pipeline needs batching, validation, JSON repair, retries, and deterministic fallbacks.
This guide implements:
- Staged generation: time β events β agents β platforms
- Batch processing for large agent sets
- JSON repair for malformed or truncated LLM output
- Rule-based fallback configs
- Type-specific activity patterns for students, officials, and media
- Validation and correction before saving the final config
π‘ The configuration pipeline processes 100+ agents through a series of API calls. Apidog was used to validate request/response schemas at each stage, catch JSON format errors before they reached production, and generate test cases for edge cases such as truncated LLM outputs.
All code comes from production use in MiroFish.
Architecture Overview
The generator is a pipeline. Each stage receives only the context it needs, which keeps requests within token limits and makes failures easier to isolate.
βββββββββββββββββββ βββββββββββββββββββ βββββββββββββββββββ
β Context β βββΊ β Time Config β βββΊ β Event Config β
β Builder β β Generator β β Generator β
β β β β β β
β - Simulation β β - Total hours β β - Initial posts β
β requirement β β - Minutes/round β β - Hot topics β
β - Entity summaryβ β - Peak hours β β - Narrative β
β - Document text β β - Activity mult β β direction β
βββββββββββββββββββ βββββββββββββββββββ βββββββββββββββββββ
β
βΌ
βββββββββββββββββββ βββββββββββββββββββ βββββββββββββββββββ
β Final Config β βββ β Platform β βββ β Agent Config β
β Assembly β β Config β β Batches β
β β β β β β
β - Merge all β β - Twitter paramsβ β - 15 agents β
β - Validate β β - Reddit params β β per batch β
β - Save JSON β β - Viral thresholdβ β - N batches β
βββββββββββββββββββ βββββββββββββββββββ βββββββββββββββββββ
A corresponding backend layout:
backend/app/services/
βββ simulation_config_generator.py # Main config generation logic
βββ ontology_generator.py # Ontology generation (shared)
βββ zep_entity_reader.py # Entity filtering
backend/app/models/
βββ task.py # Task tracking
βββ project.py # Project state
Generate Configurations in Stages
Do not ask an LLM to configure every simulation component and every agent in one request. Instead:
- Build bounded context.
- Generate global time settings.
- Generate event settings and initial posts.
- Generate agent settings in batches of 15.
- Assign initial posts to generated agents.
- Assemble platform and simulation settings.
class SimulationConfigGenerator:
# Each batch generates configs for 15 agents
AGENTS_PER_BATCH = 15
# Context limits
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
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 = []
# Step 1: global timeline and activity windows
report_progress(1, "Generating time configuration...")
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: {time_config_result.get('reasoning', 'Success')}"
)
# Step 2: topics, narrative, and initial posts
report_progress(2, "Generating event config and hot topics...")
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: {event_config_result.get('reasoning', 'Success')}"
)
# Step 3 through N: per-agent behavior settings
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"Generating agent config ({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"Agent config: Generated {len(all_agent_configs)} agents"
)
# Initial post publisher IDs can only be assigned after agents exist.
event_config = self._assign_initial_post_agents(
event_config,
all_agent_configs,
)
# Final step: platform configuration
report_progress(total_steps, "Generating platform configuration...")
twitter_config = (
PlatformConfig(platform="twitter", ...)
if enable_twitter
else None
)
reddit_config = (
PlatformConfig(platform="reddit", ...)
if enable_reddit
else None
)
return 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),
)
This structure provides practical benefits:
- Each request has a narrow, explicit responsibility.
- Progress reporting is straightforward.
- A failed agent batch can fall back to rules without invalidating time or event settings.
- Context size stays bounded as the number of entities grows.
Build Context Within a Fixed Budget
The context should include the simulation requirement, a compact entity summary, and as much source document text as the budget permits.
def _build_context(
self,
simulation_requirement: str,
document_text: str,
entities: List[EntityNode],
) -> str:
entity_summary = self._summarize_entities(entities)
context_parts = [
f"## Simulation Requirement\n{simulation_requirement}",
f"\n## Entity Information ({len(entities)} entities)\n{entity_summary}",
]
current_length = sum(len(part) for part in context_parts)
remaining_length = self.MAX_CONTEXT_LENGTH - current_length - 500
if remaining_length > 0 and document_text:
doc_text = document_text[:remaining_length]
if len(document_text) > remaining_length:
doc_text += "\n...(document truncated)"
context_parts.append(f"\n## Original Document\n{doc_text}")
return "\n".join(context_parts)
Summarize Entities by Type
Grouping entities by type gives the model useful population-level context without including every full entity record.
def _summarize_entities(self, entities: List[EntityNode]) -> str:
lines = []
by_type: Dict[str, List[EntityNode]] = {}
for entity in entities:
entity_type = entity.get_entity_type() or "Unknown"
by_type.setdefault(entity_type, []).append(entity)
for entity_type, type_entities in by_type.items():
lines.append(f"\n### {entity_type} ({len(type_entities)} entities)")
for entity in type_entities[:self.ENTITIES_PER_TYPE_DISPLAY]:
summary = entity.summary or ""
summary_preview = (
summary[:self.ENTITY_SUMMARY_LENGTH] + "..."
if len(summary) > self.ENTITY_SUMMARY_LENGTH
else summary
)
lines.append(f"- {entity.name}: {summary_preview}")
if len(type_entities) > self.ENTITIES_PER_TYPE_DISPLAY:
remaining = len(type_entities) - self.ENTITIES_PER_TYPE_DISPLAY
lines.append(f" ... and {remaining} more")
return "\n".join(lines)
Example output:
### Student (45 entities)
- Zhang Wei: Active in student union, frequently posts about campus events and academic pressure...
- Li Ming: Graduate student researching AI ethics, often shares technology news...
... and 43 more
### University (3 entities)
- Wuhan University: Official account, posts announcements and news...
Generate and Validate Time Configuration
Time configuration controls the duration of the simulation and the volume of activity per hour.
Keep the time-generation context shorter than the full context. This reduces prompt size for a task that only needs high-level information.
def _generate_time_config(
self,
context: str,
num_entities: int,
) -> Dict[str, Any]:
context_truncated = context[:self.TIME_CONFIG_CONTEXT_LENGTH]
# Cap hourly activation below the total number of available agents.
max_agents_allowed = max(1, int(num_entities * 0.9))
prompt = f"""Based on the following simulation requirements, generate time configuration.
{context_truncated}
## Task
Generate time configuration JSON.
### Basic Principles (adjust based on event type and participant groups):
- User base is Chinese, must follow Beijing timezone habits
- 0-5 AM: Almost no activity (coefficient 0.05)
- 6-8 AM: Gradually waking up (coefficient 0.4)
- 9-18 PM: Work hours, moderate activity (coefficient 0.7)
- 19-22 PM: Evening peak, most active (coefficient 1.5)
- 23 PM: Activity declining (coefficient 0.5)
### Return JSON format (no markdown):
Example:
{{
"total_simulation_hours": 72,
"minutes_per_round": 60,
"agents_per_hour_min": 5,
"agents_per_hour_max": 50,
"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": "Time configuration explanation"
}}
Field descriptions:
- total_simulation_hours (int): 24-168 hours, shorter for breaking news, longer for ongoing topics
- minutes_per_round (int): 30-120 minutes, recommend 60
- agents_per_hour_min (int): Range 1-{max_agents_allowed}
- agents_per_hour_max (int): Range 1-{max_agents_allowed}
- peak_hours (int array): Adjust based on participant groups
- off_peak_hours (int array): Usually late night/early morning
- morning_hours (int array): Morning hours
- work_hours (int array): Work hours
- reasoning (string): Brief explanation"""
system_prompt = (
"You are a social media simulation expert. Return pure JSON format."
)
try:
return self._call_llm_with_retry(prompt, system_prompt)
except Exception as error:
logger.warning(
f"Time config LLM generation failed: {error}, using default"
)
return self._get_default_time_config(num_entities)
Correct Invalid LLM Values
Never directly trust generated numeric values. Validate relationships between fields before building your domain object.
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(
"agents_per_hour_min (%s) exceeds total agents (%s), corrected",
agents_per_hour_min,
num_entities,
)
agents_per_hour_min = max(1, num_entities // 10)
if agents_per_hour_max > num_entities:
logger.warning(
"agents_per_hour_max (%s) exceeds total agents (%s), corrected",
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(
"agents_per_hour_min >= agents_per_hour_max, corrected to %s",
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,
)
Use a Deterministic Default
If the time-config request fails completely, continue the pipeline with known-safe defaults.
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": "Using default Chinese timezone configuration",
}
Generate Event Configuration
The event configuration establishes:
- Hot-topic keywords
- Narrative direction
- Initial post content
- The intended entity type for each initial post
The poster_type field is important because it lets the system map generated posts to actual agents later.
def _generate_event_config(
self,
context: str,
simulation_requirement: str,
entities: List[EntityNode],
) -> Dict[str, Any]:
entity_types_available = list(
set(entity.get_entity_type() or "Unknown" for entity in entities)
)
type_examples: Dict[str, List[str]] = {}
for entity in entities:
entity_type = entity.get_entity_type() or "Unknown"
type_examples.setdefault(entity_type, [])
if len(type_examples[entity_type]) < 3:
type_examples[entity_type].append(entity.name)
type_info = "\n".join(
f"- {entity_type}: {', '.join(examples)}"
for entity_type, examples in type_examples.items()
)
context_truncated = context[:self.EVENT_CONFIG_CONTEXT_LENGTH]
prompt = f"""Based on the following simulation requirements, generate event configuration.
Simulation Requirement: {simulation_requirement}
{context_truncated}
## Available Entity Types and Examples
{type_info}
## Task
Generate event configuration JSON:
- Extract hot topic keywords
- Describe narrative direction
- Design initial posts, each post must specify poster_type
Important: poster_type must be selected from "Available Entity Types" above, so
initial posts can be assigned to appropriate agents.
For example: Official statements should be posted by Official/University types,
news by MediaOutlet, student opinions by Student.
Return JSON format (no markdown):
{{
"hot_topics": ["keyword1", "keyword2"],
"narrative_direction": "<narrative direction description>",
"initial_posts": [
{{
"content": "Post content",
"poster_type": "Entity Type (must match available types)"
}}
],
"reasoning": "<brief explanation>"
}}"""
system_prompt = (
"You are an opinion analysis expert. Return pure JSON format."
)
try:
return self._call_llm_with_retry(prompt, system_prompt)
except Exception as error:
logger.warning(
f"Event config LLM generation failed: {error}, using default"
)
return {
"hot_topics": [],
"narrative_direction": "",
"initial_posts": [],
"reasoning": "Using default configuration",
}
Assign Initial Posts to Real Agents
The LLM generates a type such as Student, University, or MediaOutlet, not a concrete agent ID. Resolve that type after all agent configurations have been created.
Use three matching levels:
- Exact entity-type match
- Alias match for likely LLM variations
- Highest-influence agent fallback
def _assign_initial_post_agents(
self,
event_config: EventConfig,
agent_configs: List[AgentActivityConfig],
) -> EventConfig:
if not event_config.initial_posts:
return event_config
agents_by_type: Dict[str, List[AgentActivityConfig]] = {}
for agent in agent_configs:
entity_type = agent.entity_type.lower()
agents_by_type.setdefault(entity_type, []).append(agent)
type_aliases = {
"official": [
"official",
"university",
"governmentagency",
"government",
],
"university": ["university", "official"],
"mediaoutlet": ["mediaoutlet", "media"],
"student": ["student", "person"],
"professor": ["professor", "expert", "teacher"],
"alumni": ["alumni", "person"],
"organization": ["organization", "ngo", "company", "group"],
"person": ["person", "student", "alumni"],
}
# Rotate through matching agents instead of assigning every post to one ID.
used_indices: Dict[str, int] = {}
updated_posts = []
for post in event_config.initial_posts:
poster_type = post.get("poster_type", "").lower()
content = post.get("content", "")
matched_agent_id = None
# 1. Exact type match
if poster_type in agents_by_type:
agents = agents_by_type[poster_type]
index = used_indices.get(poster_type, 0) % len(agents)
matched_agent_id = agents[index].agent_id
used_indices[poster_type] = index + 1
# 2. Alias match
else:
for alias_key, aliases in type_aliases.items():
if poster_type in aliases or alias_key == poster_type:
for alias in aliases:
if alias in agents_by_type:
agents = agents_by_type[alias]
index = used_indices.get(alias, 0) % len(agents)
matched_agent_id = agents[index].agent_id
used_indices[alias] = index + 1
break
if matched_agent_id is not None:
break
# 3. Highest-influence fallback
if matched_agent_id is None:
logger.warning(
"No matching agent for type '%s', using highest influence agent",
poster_type,
)
if agent_configs:
matched_agent_id = max(
agent_configs,
key=lambda agent: agent.influence_weight,
).agent_id
else:
matched_agent_id = 0
updated_posts.append(
{
"content": content,
"poster_type": post.get("poster_type", "Unknown"),
"poster_agent_id": matched_agent_id,
}
)
logger.info(
"Initial post assignment: poster_type='%s' -> agent_id=%s",
poster_type,
matched_agent_id,
)
event_config.initial_posts = updated_posts
return event_config
Generate Agent Configurations in Batches
For hundreds of agents, avoid a single large response. Generate configurations in batches of 15 agents.
Each batch includes only the information needed for those agents: ID, name, type, and a shortened entity summary.
def _generate_agent_configs_batch(
self,
context: str,
entities: List[EntityNode],
start_idx: int,
simulation_requirement: str,
) -> List[AgentActivityConfig]:
entity_list = []
for index, entity in enumerate(entities):
entity_list.append(
{
"agent_id": start_idx + index,
"entity_name": entity.name,
"entity_type": entity.get_entity_type() or "Unknown",
"summary": (
entity.summary[:self.AGENT_SUMMARY_LENGTH]
if entity.summary
else ""
),
}
)
prompt = f"""Based on the following information, generate social media activity configuration for each entity.
Simulation Requirement: {simulation_requirement}
## Entity List
json
{json.dumps(entity_list, ensure_ascii=False, indent=2)}
## Task
Generate activity configuration for each entity. Note:
- Time must follow Chinese habits: 0-5 AM almost no activity, 19-22 PM most active
- Official institutions (University/GovernmentAgency): Low activity (0.1-0.3), work hours (9-17), slow response (60-240 min), high influence (2.5-3.0)
- Media (MediaOutlet): Moderate activity (0.4-0.6), all-day activity (8-23), fast response (5-30 min), high influence (2.0-2.5)
- Individuals (Student/Person/Alumni): High activity (0.6-0.9), mainly evening (18-23), fast response (1-15 min), low influence (0.8-1.2)
- Public figures/Experts: Moderate activity (0.4-0.6), medium-high influence (1.5-2.0)
Return JSON only:
{{
"agent_configs": [
{{
"agent_id": 0,
"activity_level": 0.5,
"posts_per_hour": 0.5,
"comments_per_hour": 1.0,
"active_hours": [9, 10, 11],
"response_delay_min": 5,
"response_delay_max": 60,
"sentiment_bias": 0.0,
"stance": "neutral",
"influence_weight": 1.0
}}
]
}}"""
system_prompt = (
"You are a social media behavior analysis expert. Return pure JSON format."
)
try:
result = self._call_llm_with_retry(prompt, system_prompt)
llm_configs = {
config["agent_id"]: config
for config in result.get("agent_configs", [])
}
except Exception as error:
logger.warning(
f"Agent config batch LLM generation failed: {error}, "
"using rule-based generation"
)
llm_configs = {}
configs = []
for index, entity in enumerate(entities):
agent_id = start_idx + index
config_data = llm_configs.get(agent_id, {})
if not config_data:
config_data = self._generate_agent_config_by_rule(entity)
configs.append(
AgentActivityConfig(
agent_id=agent_id,
entity_uuid=entity.uuid,
entity_name=entity.name,
entity_type=entity.get_entity_type() or "Unknown",
activity_level=config_data.get("activity_level", 0.5),
posts_per_hour=config_data.get("posts_per_hour", 0.5),
comments_per_hour=config_data.get("comments_per_hour", 1.0),
active_hours=config_data.get(
"active_hours",
list(range(9, 23)),
),
response_delay_min=config_data.get(
"response_delay_min",
5,
),
response_delay_max=config_data.get(
"response_delay_max",
60,
),
sentiment_bias=config_data.get("sentiment_bias", 0.0),
stance=config_data.get("stance", "neutral"),
influence_weight=config_data.get(
"influence_weight",
1.0,
),
)
)
return configs
python
Add Rule-Based Agent Fallbacks
A fallback should produce valid behavior even when an LLM request fails, returns malformed JSON, or omits an agent.
These defaults encode the same type-specific behavior expected in the LLM prompt.
def _generate_agent_config_by_rule(
self,
entity: EntityNode,
) -> Dict[str, Any]:
entity_type = (entity.get_entity_type() or "Unknown").lower()
if entity_type in ["university", "governmentagency", "ngo"]:
# Official institution: work hours, low frequency, high influence
return {
"activity_level": 0.2,
"posts_per_hour": 0.1,
"comments_per_hour": 0.05,
"active_hours": list(range(9, 18)),
"response_delay_min": 60,
"response_delay_max": 240,
"sentiment_bias": 0.0,
"stance": "neutral",
"influence_weight": 3.0,
}
if entity_type in ["mediaoutlet"]:
# Media: all-day activity, moderate frequency, high influence
return {
"activity_level": 0.5,
"posts_per_hour": 0.8,
"comments_per_hour": 0.3,
"active_hours": list(range(7, 24)),
"response_delay_min": 5,
"response_delay_max": 30,
"sentiment_bias": 0.0,
"stance": "observer",
"influence_weight": 2.5,
}
if entity_type in ["professor", "expert", "official"]:
# Expert or professor: work hours plus evening activity
return {
"activity_level": 0.4,
"posts_per_hour": 0.3,
"comments_per_hour": 0.5,
"active_hours": list(range(8, 22)),
"response_delay_min": 15,
"response_delay_max": 90,
"sentiment_bias": 0.0,
"stance": "neutral",
"influence_weight": 2.0,
}
if entity_type in ["student"]:
# Student: evening peak and high interaction frequency
return {
"activity_level": 0.8,
"posts_per_hour": 0.6,
"comments_per_hour": 1.5,
"active_hours": [
8, 9, 10, 11, 12, 13,
18, 19, 20, 21, 22, 23,
],
"response_delay_min": 1,
"response_delay_max": 15,
"sentiment_bias": 0.0,
"stance": "neutral",
"influence_weight": 0.8,
}
if entity_type in ["alumni"]:
# Alumni: lunch and evening activity
return {
"activity_level": 0.6,
"posts_per_hour": 0.4,
"comments_per_hour": 0.8,
"active_hours": [12, 13, 19, 20, 21, 22, 23],
"response_delay_min": 5,
"response_delay_max": 30,
"sentiment_bias": 0.0,
"stance": "neutral",
"influence_weight": 1.0,
}
# Default person: daytime plus evening activity
return {
"activity_level": 0.7,
"posts_per_hour": 0.5,
"comments_per_hour": 1.2,
"active_hours": [9, 10, 11, 12, 13, 18, 19, 20, 21, 22, 23],
"response_delay_min": 2,
"response_delay_max": 20,
"sentiment_bias": 0.0,
"stance": "neutral",
"influence_weight": 1.0,
}
Retry LLM Calls and Repair JSON
LLM output handling should have multiple layers:
- Request JSON output from the model.
- Detect length-based truncation.
- Attempt a normal JSON parse.
- Repair common formatting issues.
- Retry with lower temperature.
- Fall back to deterministic settings when all retries fail.
def _call_llm_with_retry(
self,
prompt: str,
system_prompt: str,
) -> Dict[str, Any]:
max_attempts = 3
last_error = None
for attempt in range(max_attempts):
try:
response = self.client.chat.completions.create(
model=self.model_name,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt},
],
response_format={"type": "json_object"},
temperature=0.7 - (attempt * 0.1),
)
content = response.choices[0].message.content
finish_reason = response.choices[0].finish_reason
if finish_reason == "length":
logger.warning(
"LLM output truncated (attempt %s)",
attempt + 1,
)
content = self._fix_truncated_json(content)
try:
return json.loads(content)
except json.JSONDecodeError as error:
logger.warning(
"JSON parse failed (attempt %s): %s",
attempt + 1,
str(error)[:80],
)
fixed = self._try_fix_config_json(content)
if fixed:
return fixed
last_error = error
except Exception as error:
logger.warning(
"LLM call failed (attempt %s): %s",
attempt + 1,
str(error)[:80],
)
last_error = error
import time
time.sleep(2 * (attempt + 1))
raise last_error or Exception("LLM call failed")
Repair Truncated JSON
For truncated JSON, first balance braces and brackets. If the output appears to end in an incomplete string, close the string before closing containers.
def _fix_truncated_json(self, content: str) -> str:
content = content.strip()
open_braces = content.count("{") - content.count("}")
open_brackets = content.count("[") - content.count("]")
if content and content[-1] not in '",}]':
content += '"'
content += "]" * open_brackets
content += "}" * open_braces
return content
Apply Additional JSON Repairs
When bracket balancing is not enough, extract the JSON body and normalize problematic string content or control characters.
def _try_fix_config_json(
self,
content: str,
) -> Optional[Dict[str, Any]]:
content = self._fix_truncated_json(content)
json_match = re.search(r"\{[\s\S]*\}", content)
if not json_match:
return None
json_str = json_match.group()
def fix_string(match):
value = match.group(0)
value = value.replace("\n", " ").replace("\r", " ")
return re.sub(r"\s+", " ", value)
json_str = re.sub(
r'"[^"\\]*(?:\\.[^"\\]*)*"',
fix_string,
json_str,
)
try:
return json.loads(json_str)
except json.JSONDecodeError:
json_str = re.sub(r"[\x00-\x1f\x7f-\x9f]", " ", json_str)
json_str = re.sub(r"\s+", " ", json_str)
try:
return json.loads(json_str)
except json.JSONDecodeError:
return None
Define Explicit Configuration Models
Typed configuration structures make validation, serialization, and debugging easier.
Agent Activity Config
@dataclass
class AgentActivityConfig:
"""Single agent activity configuration."""
agent_id: int
entity_uuid: str
entity_name: str
entity_type: str
# Activity level (0.0-1.0)
activity_level: float = 0.5
# Posting frequency (per hour)
posts_per_hour: float = 1.0
comments_per_hour: float = 2.0
# Active hours (24-hour format, 0-23)
active_hours: List[int] = field(
default_factory=lambda: list(range(8, 23))
)
# Response speed in simulated minutes
response_delay_min: int = 5
response_delay_max: int = 60
# Sentiment tendency (-1.0 to 1.0)
sentiment_bias: float = 0.0
# supportive, opposing, neutral, observer
stance: str = "neutral"
# Affects probability of being seen
influence_weight: float = 1.0
Time Simulation Config
@dataclass
class TimeSimulationConfig:
"""Time simulation configuration (Chinese timezone)."""
total_simulation_hours: int = 72
minutes_per_round: int = 60
# Agents activated per hour
agents_per_hour_min: int = 5
agents_per_hour_max: int = 20
# Evening peak
peak_hours: List[int] = field(
default_factory=lambda: [19, 20, 21, 22]
)
peak_activity_multiplier: float = 1.5
# Early-morning low activity
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
Complete Simulation Parameters
@dataclass
class SimulationParameters:
"""Complete simulation parameter configuration."""
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]:
return {
"simulation_id": self.simulation_id,
"project_id": self.project_id,
"graph_id": self.graph_id,
"simulation_requirement": self.simulation_requirement,
"time_config": asdict(self.time_config),
"agent_configs": [
asdict(agent)
for agent 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,
}
Agent Type Defaults
Use these patterns as both prompt guidance and deterministic fallback behavior.
| Agent Type | Activity | Active Hours | Posts/Hour | Comments/Hour | Response (min) | Influence |
|---|---|---|---|---|---|---|
| University | 0.2 | 9-17 | 0.1 | 0.05 | 60-240 | 3.0 |
| GovernmentAgency | 0.2 | 9-17 | 0.1 | 0.05 | 60-240 | 3.0 |
| MediaOutlet | 0.5 | 7-23 | 0.8 | 0.3 | 5-30 | 2.5 |
| Professor | 0.4 | 8-21 | 0.3 | 0.5 | 15-90 | 2.0 |
| Student | 0.8 | 8-12, 18-23 | 0.6 | 1.5 | 1-15 | 0.8 |
| Alumni | 0.6 | 12-13, 19-23 | 0.4 | 0.8 | 5-30 | 1.0 |
| Person (default) | 0.7 | 9-13, 18-23 | 0.5 | 1.2 | 2-20 | 1.0 |
Conclusion
Reliable LLM-powered configuration generation is mostly a systems problem, not only a prompting problem.
Build the pipeline around these implementation rules:
- Generate configuration in stages: time β events β agents β platforms.
- Process agents in batches of 15 to control context size and output length.
- Validate generated values before constructing domain objects.
- Repair malformed and truncated JSON before retrying.
- Lower temperature on retries to reduce output variability.
- Use rule-based defaults so a failed LLM call does not stop the simulation.
- Apply type-specific activity patterns to keep behavior internally consistent.
- Correct impossible values, such as
agents_per_hour_maxexceeding the total agent count.
With these guardrails, the system can generate configurations for 100+ agents while remaining recoverable when individual LLM calls fail.
Top comments (0)