Giriş
Bir sosyal medya simülasyonu için yüzlerce yapay zeka ajanını yapılandırmak göz korkutucu gelebilir. Her ajanın etkinlik çizelgelerine, gönderi sıklıklarına, yanıt gecikmelerine, etki ağırlıklarına ve duruş pozisyonlarına ihtiyacı vardır. Bunu manuel olarak yapmak saatler sürerdi.
MiroFish bunu LLM destekli yapılandırma oluşturma ile otomatikleştirir. Sistem, belgelerinizi, bilgi grafiğinizi ve simülasyon gereksinimlerinizi analiz eder, ardından her ajan için ayrıntılı yapılandırmalar oluşturur.
Zorluk: LLM'ler başarısız olabilir. Çıktılar kesilebilir. JSON bozulabilir. Token limitleri sorun yaratabilir.
Bu kılavuz, eksiksiz uygulamayı kapsar:
- Adım adım oluşturma (zaman → olaylar → ajanlar → platformlar)
- Bağlam sınırlamalarından kaçınmak için toplu işleme
- Kesilmiş çıktılar için JSON onarım stratejileri
- LLM başarısız olduğunda kural tabanlı yedek yapılandırmalar
- Türe göre ajan etkinlik desenleri (Öğrenciye Karşı Yetkiliye Karşı Medya)
- Doğrulama ve düzeltme mantığı
💡
Yapılandırma oluşturma işlem hattı, bir dizi API çağrısı aracılığıyla 100'den fazla ajanı işler. Apidog, her aşamada istek/yanıt şemalarını doğrulamak, üretim ortamına ulaşmadan önce JSON formatı hatalarını yakalamak ve kesilmiş LLM çıktıları gibi uç senaryolar için test durumları oluşturmak için kullanıldı.
Tüm kod, MiroFish'deki üretim kullanımından alınmıştır.
Mimariye Genel Bakış
Yapılandırma oluşturucu, boru hattı tabanlı bir yaklaşım kullanır:
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Bağlam │ ──► │ Zaman Yapıl. │ ──► │ Olay Yapıl. │
│ Oluşturucu │ │ Oluşturucu │ │ Oluşturucu │
│ │ │ │ │ │
│ - Simülasyon │ │ - Toplam saat │ │ - İlk gönderiler│
│ gereksinimi │ │ - Dakika/tur │ │ - Gündemler │
│ - Varlık özeti │ │ - Yoğun saatler │ │ - Anlatı │
│ - Belge metni │ │ - Etkinlik çar. │ │ yönü │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│
▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Son Yapıl. │ ◄── │ Platform │ ◄── │ Ajan Yapıl. │
│ Birleştirme │ │ Yapıl. │ │ Gruplar │
│ │ │ │ │ │
│ - Hepsini bir. │ │ - Twitter para. │ │ - 15 ajan │
│ - Doğrula │ │ - Reddit para. │ │ her grup │
│ - JSON kaydet │ │ - Viral eşik │ │ - N grup │
└─────────────────┘ └─────────────────┘ └─────────────────┘
Dosya Yapısı
backend/app/services/
├── simulation_config_generator.py # Ana yapılandırma oluşturma mantığı
├── ontology_generator.py # Ontoloji oluşturma (paylaşımlı)
└── zep_entity_reader.py # Varlık filtreleme
backend/app/models/
├── task.py # Görev takibi
└── project.py # Proje durumu
Adım Adım Oluşturma Stratejisi
Tüm yapılandırmaları aynı anda oluşturmak, token limitlerini aşar. Bunun yerine, sistem aşamalı olarak oluşturur:
class SimulationConfigGenerator:
# Her parti 15 ajan için yapılandırma oluşturur
AGENTS_PER_BATCH = 15
# Bağlam limitleri
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:
# Toplam adım sayısını hesapla
num_batches = math.ceil(len(entities) / self.AGENTS_PER_BATCH)
total_steps = 3 + num_batches # Zaman + Olaylar + N Ajan Grupları + Platform
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}")
# Bağlam oluştur
context = self._build_context(
simulation_requirement=simulation_requirement,
document_text=document_text,
entities=entities
)
reasoning_parts = []
# Adım 1: Zaman yapılandırması oluştur
report_progress(1, "Zaman yapılandırması oluşturuluyor...")
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"Zaman yapılandırması: {time_config_result.get('reasoning', 'Başarılı')}")
# Adım 2: Olay yapılandırması ve gündemler oluştur
report_progress(2, "Olay yapılandırması ve gündemler oluşturuluyor...")
event_config_result = self._generate_event_config(context, simulation_requirement, entities)
event_config = self._parse_event_config(event_config_result)
reasoning_parts.append(f"Olay yapılandırması: {event_config_result.get('reasoning', 'Başarılı')}")
# 3-N Adımları: Ajan yapılandırmalarını gruplar halinde oluştur
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"Ajan yapılandırması oluşturuluyor ({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"Ajan yapılandırması: {len(all_agent_configs)} ajan oluşturuldu")
# İlk gönderi yayıncılarını ata
event_config = self._assign_initial_post_agents(event_config, all_agent_configs)
# Son adım: Platform yapılandırması
report_progress(total_steps, "Platform yapılandırması oluşturuluyor...")
twitter_config = PlatformConfig(platform="twitter", ...) if enable_twitter else None
reddit_config = PlatformConfig(platform="reddit", ...) if enable_reddit else None
# Son yapılandırmayı birleştir
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
Bu aşamalı yaklaşım:
- Her LLM çağrısını odaklanmış ve yönetilebilir tutar
- Kullanıcıya ilerleme güncellemeleri sağlar
- Bir aşama başarısız olursa kısmi kurtarmaya izin verir
Bağlam Oluşturma
Bağlam oluşturucu, token limitlerine saygı duyarak ilgili bilgileri bir araya getirir:
def _build_context(
self,
simulation_requirement: str,
document_text: str,
entities: List[EntityNode]
) -> str:
# Varlık özeti
entity_summary = self._summarize_entities(entities)
context_parts = [
f"## Simülasyon Gereksinimi\n{simulation_requirement}",
f"\n## Varlık Bilgileri ({len(entities)} varlık)\n{entity_summary}",
]
# Yer varsa belge metnini ekle
current_length = sum(len(p) for p in context_parts)
remaining_length = self.MAX_CONTEXT_LENGTH - current_length - 500 # 500 karakterlik tampon
if remaining_length > 0 and document_text:
doc_text = document_text[:remaining_length]
if len(document_text) > remaining_length:
doc_text += "\n...(belge kesildi)"
context_parts.append(f"\n## Orijinal Belge\n{doc_text}")
return "\n".join(context_parts)
Varlık Özetleme
Varlıklar türe göre özetlenir:
def _summarize_entities(self, entities: List[EntityNode]) -> str:
lines = []
# Türe göre grupla
by_type: Dict[str, List[EntityNode]] = {}
for e in entities:
t = e.get_entity_type() or "Unknown"
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)} varlık)")
# Sınırlı sayıda varlığı, sınırlı özet uzunluğuyla göster
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" ... ve {len(type_entities) - display_count} tane daha")
return "\n".join(lines)
Bu, aşağıdaki gibi bir çıktı üretir:
### Öğrenci (45 varlık)
- Zhang Wei: Öğrenci birliğinde aktif, sık sık kampüs etkinlikleri ve akademik baskı hakkında gönderi paylaşıyor...
- Li Ming: Yapay zeka etiği üzerine araştırma yapan lisansüstü öğrenci, genellikle teknoloji haberleri paylaşıyor...
... ve 43 tane daha
### Üniversite (3 varlık)
- Wuhan Üniversitesi: Resmi hesap, duyurular ve haberler paylaşıyor...
Zaman Yapılandırması Oluşturma
Zaman yapılandırması, simülasyon süresini ve etkinlik desenlerini belirler:
def _generate_time_config(self, context: str, num_entities: int) -> Dict[str, Any]:
# Bu özel adım için bağlamı kırp
context_truncated = context[:self.TIME_CONFIG_CONTEXT_LENGTH]
# İzin verilen maksimum değeri hesapla (ajan sayısının %90'ı)
max_agents_allowed = max(1, int(num_entities * 0.9))
prompt = f"""Aşağıdaki simülasyon gereksinimlerine dayanarak, zaman yapılandırmasını oluştur.
{context_truncated}
## Görev
Zaman yapılandırma JSON'ını oluştur.
### Temel İlkeler (olay türüne ve katılımcı gruplarına göre ayarla):
- Kullanıcı tabanı Çinli, Pekin saat dilimi alışkanlıklarına uymalıdır
- 0-5 AM: Neredeyse hiç etkinlik yok (katsayı 0.05)
- 6-8 AM: Yavaş yavaş uyanma (katsayı 0.4)
- 9-18 PM: Çalışma saatleri, orta düzey etkinlik (katsayı 0.7)
- 19-22 PM: Akşam zirvesi, en aktif (katsayı 1.5)
- 23 PM: Etkinlik azalıyor (katsayı 0.5)
### JSON formatında döndür (markdown yok):
Örnek:
{{
"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": "Zaman yapılandırma açıklaması"
}}
Alan açıklamaları:
- total_simulation_hours (int): 24-168 saat, son dakika haberleri için daha kısa, devam eden konular için daha uzun
- minutes_per_round (int): 30-120 dakika, 60 önerilir
- agents_per_hour_min (int): Aralığı 1-{max_agents_allowed}
- agents_per_hour_max (int): Aralığı 1-{max_agents_allowed}
- peak_hours (int array): Katılımcı gruplarına göre ayarla
- off_peak_hours (int array): Genellikle geç gece/erken sabah
- morning_hours (int array): Sabah saatleri
- work_hours (int array): Çalışma saatleri
- reasoning (string): Kısa açıklama"""
system_prompt = "Sosyal medya simülasyonu uzmanısınız. Saf JSON formatında döndürün."
try:
return self._call_llm_with_retry(prompt, system_prompt)
except Exception as e:
logger.warning(f"Zaman yapılandırması LLM oluşturma başarısız oldu: {e}, varsayılan kullanılıyor")
return self._get_default_time_config(num_entities)
Zaman Yapılandırmasını Ayrıştırma ve Doğrulama
def _parse_time_config(self, result: Dict[str, Any], num_entities: int) -> TimeSimulationConfig:
# Ham değerleri al
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))
# Doğrula ve düzelt: toplam ajan sayısını aşmadığından emin ol
if agents_per_hour_min > num_entities:
logger.warning(f"agents_per_hour_min ({agents_per_hour_min}) toplam ajanı ({num_entities}) aşıyor, düzeltildi")
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}) toplam ajanı ({num_entities}) aşıyor, düzeltildi")
agents_per_hour_max = max(agents_per_hour_min + 1, num_entities // 2)
# Min < max olduğundan emin ol
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} olarak düzeltildi")
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
)
Varsayılan Zaman Yapılandırması (Çin Saat Dilimi)
def _get_default_time_config(self, num_entities: int) -> Dict[str, Any]:
return {
"total_simulation_hours": 72,
"minutes_per_round": 60, # Tur başına 1 saat
"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": "Varsayılan Çin saat dilimi yapılandırması kullanılıyor"
}
Olay Yapılandırması Oluşturma
Olay yapılandırması, ilk gönderileri, gündemleri ve anlatı yönünü tanımlar:
def _generate_event_config(
self,
context: str,
simulation_requirement: str,
entities: List[EntityNode]
) -> Dict[str, Any]:
# LLM referansı için mevcut varlık türlerini al
entity_types_available = list(set(
e.get_entity_type() or "Unknown" for e in entities
))
# Türe göre örnekleri göster
type_examples = {}
for e in entities:
etype = e.get_entity_type() or "Unknown"
if etype not in type_examples:
type_examples[etype] = []
if len(type_examples[etype]) < 3:
type_examples[etype].append(e.name)
type_info = "\n".join([
f"- {t}: {', '.join(examples)}"
for t, examples in type_examples.items()
])
context_truncated = context[:self.EVENT_CONFIG_CONTEXT_LENGTH]
prompt = f"""Aşağıdaki simülasyon gereksinimlerine dayanarak, olay yapılandırmasını oluştur.
Simülasyon Gereksinimi: {simulation_requirement}
{context_truncated}
## Mevcut Varlık Türleri ve Örnekleri
{type_info}
## Görev
Olay yapılandırma JSON'ını oluştur:
- Gündem anahtar kelimelerini çıkar
- Anlatı yönünü açıkla
- İlk gönderileri tasarla, **her gönderi poster_type'ı belirtmeli**
**Önemli**: poster_type, yukarıdaki "Mevcut Varlık Türleri" arasından seçilmelidir, böylece ilk gönderiler uygun ajanlara atanabilir.
Örneğin: Resmi açıklamalar Yetkili/Üniversite türleri tarafından, haberler MedyaKuruluşu tarafından, öğrenci görüşleri Öğrenci tarafından yayınlanmalıdır.
JSON formatında döndür (markdown yok):
{{
"hot_topics": ["anahtarkelime1", "anahtarkelime2", ...],
"narrative_direction": "<anlatı yönü açıklaması>",
"initial_posts": [
{{"content": "Gönderi içeriği", "poster_type": "Varlık Türü (mevcut türlerle eşleşmeli)"}},
...
],
"reasoning": "<kısa açıklama>"
}}"""
system_prompt = "Bir fikir analizi uzmanısınız. Saf JSON formatında döndürün."
try:
return self._call_llm_with_retry(prompt, system_prompt)
except Exception as e:
logger.warning(f"Olay yapılandırması LLM oluşturma başarısız oldu: {e}, varsayılan kullanılıyor")
return {
"hot_topics": [],
"narrative_direction": "",
"initial_posts": [],
"reasoning": "Varsayılan yapılandırma kullanılıyor"
}
İlk Gönderi Yayıncılarını Atama
İlk gönderileri oluşturduktan sonra, bunları gerçek ajanlarla eşleştir:
def _assign_initial_post_agents(
self,
event_config: EventConfig,
agent_configs: List[AgentActivityConfig]
) -> EventConfig:
if not event_config.initial_posts:
return event_config
# Ajanları türe göre indeksle
agents_by_type: Dict[str, List[AgentActivityConfig]] = {}
for agent in agent_configs:
etype = agent.entity_type.lower()
if etype not in agents_by_type:
agents_by_type[etype] = []
agents_by_type[etype].append(agent)
# Tip takma adı eşlemesi (LLM varyasyonlarını işler)
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"],
}
# Aynı ajanı tekrar kullanmamak için kullanılan indeksleri takip et
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. Doğrudan eşleşme
if poster_type in agents_by_type:
agents = agents_by_type[poster_type]
idx = used_indices.get(poster_type, 0) % len(agents)
matched_agent_id = agents[idx].agent_id
used_indices[poster_type] = idx + 1
else:
# 2. Takma ad eşleşmesi
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]
idx = used_indices.get(alias, 0) % len(agents)
matched_agent_id = agents[idx].agent_id
used_indices[alias] = idx + 1
break
if matched_agent_id is not None:
break
# 3. Geri dönüş: en yüksek etkiye sahip ajanı kullan
if matched_agent_id is None:
logger.warning(f"'{poster_type}' türü için eşleşen ajan yok, en yüksek etkiye sahip ajan kullanılıyor")
if agent_configs:
sorted_agents = sorted(agent_configs, key=lambda a: a.influence_weight, reverse=True)
matched_agent_id = sorted_agents[0].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(f"İlk gönderi ataması: poster_type='{poster_type}' -> agent_id={matched_agent_id}")
event_config.initial_posts = updated_posts
return event_config
Toplu Ajan Yapılandırması Oluşturma
Yüzlerce ajan için yapılandırmaları aynı anda oluşturmak, token limitlerini aşar. Sistem 15'erli gruplar halinde işler:
def _generate_agent_configs_batch(
self,
context: str,
entities: List[EntityNode],
start_idx: int,
simulation_requirement: str
) -> List[AgentActivityConfig]:
# Sınırlı özet uzunluğuyla varlık bilgisini oluştur
entity_list = []
summary_len = self.AGENT_SUMMARY_LENGTH
for i, e in enumerate(entities):
entity_list.append({
"agent_id": start_idx + i,
"entity_name": e.name,
"entity_type": e.get_entity_type() or "Unknown",
"summary": e.summary[:summary_len] if e.summary else ""
})
prompt = f"""Aşağıdaki bilgilere dayanarak, her varlık için sosyal medya etkinlik yapılandırmasını oluştur.
Simülasyon Gereksinimi: {simulation_requirement}
## Varlık Listesi
```json
{json.dumps(entity_list, ensure_ascii=False, indent=2)}
Görev
Her varlık için etkinlik yapılandırması oluştur. Not:
- Zaman Çin alışkanlıklarına uymalıdır: 0-5 AM neredeyse hiç etkinlik yok, 19-22 PM en aktif
- Resmi kurumlar (Üniversite/DevletKurumu): Düşük etkinlik (0.1-0.3), çalışma saatleri (9-17), yavaş yanıt (60-240 dk), yüksek etki (2.5-3.0)
- Medya (MedyaKuruluşu): Orta düzey etkinlik (0.4-0.6), tüm gün etkinlik (8-23), hızlı yanıt (5-30 dk), yüksek etki (2.0-2.5)
- Bireyler (Öğrenci/Kişi/Mezun): Yüksek etkinlik (0.6-0.9), ağırlıklı olarak akşam (18-23), hızlı yanıt (1-15 dk), düşük etki (0.8-1.2)
- Tanınmış kişiler/Uzmanlar: Orta düzey etkinlik (0.4-0.6), orta-yüksek etki (1.5-2.0)
system_prompt = "Sosyal medya davranış analizi uzmanısınız. Saf JSON formatında döndürün."
try:
result = self._call_llm_with_retry(prompt, system_prompt)
llm_configs = {cfg["agent_id"]: cfg for cfg in result.get("agent_configs", [])}
except Exception as e:
logger.warning(f"Ajan yapılandırma grubu LLM oluşturma başarısız oldu: {e}, kural tabanlı oluşturma kullanılıyor")
llm_configs = {}
# AgentActivityConfig nesneleri oluştur
configs = []
for i, entity in enumerate(entities):
agent_id = start_idx + i
cfg = llm_configs.get(agent_id, {})
# LLM başarısız olursa kural tabanlı geri dönüş kullan
if not cfg:
cfg = self._generate_agent_config_by_rule(entity)
config = AgentActivityConfig(
agent_id=agent_id,
entity_uuid=entity.uuid,
entity_name=entity.name,
entity_type=entity.get_entity_type() or "Unknown",
activity_level=cfg.get("activity_level", 0.5),
posts_per_hour=cfg.get("posts_per_hour", 0.5),
comments_per_hour=cfg.get("comments_per_hour", 1.0),
active_hours=cfg.get("active_hours", list(range(9, 23))),
response_delay_min=cfg.get("response_delay_min", 5),
response_delay_max=cfg.get("response_delay_max", 60),
sentiment_bias=cfg.get("sentiment_bias", 0.0),
stance=cfg.get("stance", "neutral"),
influence_weight=cfg.get("influence_weight", 1.0)
)
configs.append(config)
return configs
### Kural Tabanlı Geri Dönüş Yapılandırmaları
LLM başarısız olduğunda, önceden tanımlanmış desenleri kullan:
```python
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"]:
# Resmi kurum: çalışma saatleri, düşük sıklık, yüksek etki
return {
"activity_level": 0.2,
"posts_per_hour": 0.1,
"comments_per_hour": 0.05,
"active_hours": list(range(9, 18)), # 9:00-17:59
"response_delay_min": 60,
"response_delay_max": 240,
"sentiment_bias": 0.0,
"stance": "neutral",
"influence_weight": 3.0
}
elif entity_type in ["mediaoutlet"]:
# Medya: tüm gün etkinlik, orta sıklık, yüksek etki
return {
"activity_level": 0.5,
"posts_per_hour": 0.8,
"comments_per_hour": 0.3,
"active_hours": list(range(7, 24)), # 7:00-23:59
"response_delay_min": 5,
"response_delay_max": 30,
"sentiment_bias": 0.0,
"stance": "observer",
"influence_weight": 2.5
}
elif entity_type in ["professor", "expert", "official"]:
# Uzman/Profesör: çalışma + akşam, orta sıklık
return {
"activity_level": 0.4,
"posts_per_hour": 0.3,
"comments_per_hour": 0.5,
"active_hours": list(range(8, 22)), # 8:00-21:59
"response_delay_min": 15,
"response_delay_max": 90,
"sentiment_bias": 0.0,
"stance": "neutral",
"influence_weight": 2.0
}
elif entity_type in ["student"]:
# Öğrenci: akşam zirvesi, yüksek sıklık
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
}
elif entity_type in ["alumni"]:
# Mezun: akşam odaklı
return {
"activity_level": 0.6,
"posts_per_hour": 0.4,
"comments_per_hour": 0.8,
"active_hours": [12, 13, 19, 20, 21, 22, 23], # Öğle + akşam
"response_delay_min": 5,
"response_delay_max": 30,
"sentiment_bias": 0.0,
"stance": "neutral",
"influence_weight": 1.0
}
else:
# Varsayılan kişi: akşam zirvesi
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
}
Yeniden Denemeli ve JSON Onarımlı LLM Çağrısı
LLM çağrıları başarısız olur. Çıktılar kesilir. JSON bozulur. Sistem tüm bunları ele alır:
def _call_llm_with_retry(self, prompt: str, system_prompt: str) -> Dict[str, Any]:
import re
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) # Yeniden denemede sıcaklığı düşür
)
content = response.choices[0].message.content
finish_reason = response.choices[0].finish_reason
# Kesilip kesilmediğini kontrol et
if finish_reason == 'length':
logger.warning(f"LLM çıktısı kesildi (deneme {attempt+1})")
content = self._fix_truncated_json(content)
# JSON ayrıştırmayı dene
try:
return json.loads(content)
except json.JSONDecodeError as e:
logger.warning(f"JSON ayrıştırma başarısız oldu (deneme {attempt+1}): {str(e)[:80]}")
# JSON'ı onarmayı dene
fixed = self._try_fix_config_json(content)
if fixed:
return fixed
last_error = e
except Exception as e:
logger.warning(f"LLM çağrısı başarısız oldu (deneme {attempt+1}): {str(e)[:80]}")
last_error = e
import time
time.sleep(2 * (attempt + 1))
raise last_error or Exception("LLM çağrısı başarısız oldu")
Kesilmiş JSON'ı Düzeltme
def _fix_truncated_json(self, content: str) -> str:
content = content.strip()
# Kapanmamış parantezleri say
open_braces = content.count('{') - content.count('}')
open_brackets = content.count('[') - content.count(']')
# Kapanmamış dizeyi kontrol et
if content and content[-1] not in '",}]':
content += '"'
# Parantezleri kapat
content += ']' * open_brackets
content += '}' * open_braces
return content
Gelişmiş JSON Onarımı
def _try_fix_config_json(self, content: str) -> Optional[Dict[str, Any]]:
import re
# Kesmeyi düzelt
content = self._fix_truncated_json(content)
# JSON kısmını çıkar
json_match = re.search(r'\{[\s\S]*\}', content)
if json_match:
json_str = json_match.group()
# Dizelerdeki yeni satırları kaldır
def fix_string(match):
s = match.group(0)
s = s.replace('\n', ' ').replace('\r', ' ')
s = re.sub(r'\s+', ' ', s)
return s
json_str = re.sub(r'"[^"\\]*(?:\\.[^"\\]*)*"', fix_string, json_str)
try:
return json.loads(json_str)
except:
# Kontrol karakterlerini kaldırmayı dene
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:
pass
return None
Yapılandırma Veri Yapıları
Ajan Etkinlik Yapılandırması
@dataclass
class AgentActivityConfig:
"""Tek ajan etkinlik yapılandırması"""
agent_id: int
entity_uuid: str
entity_name: str
entity_type: str
# Etkinlik seviyesi (0.0-1.0)
activity_level: float = 0.5
# Gönderi sıklığı (saat başına)
posts_per_hour: float = 1.0
comments_per_hour: float = 2.0
# Aktif saatler (24 saat formatı, 0-23)
active_hours: List[int] = field(default_factory=lambda: list(range(8, 23)))
# Yanıt hızı (simüle edilmiş dakikalar cinsinden tepki gecikmesi)
response_delay_min: int = 5
response_delay_max: int = 60
# Duygu eğilimi (-1.0 ila 1.0, negatiften pozitife)
sentiment_bias: float = 0.0
# Belirli konulara karşı duruş
stance: str = "neutral" # destekleyici, karşıt, tarafsız, gözlemci
# Etki ağırlığı (görülme olasılığını etkiler)
influence_weight: float = 1.0
Zaman Simülasyon Yapılandırması
@dataclass
class TimeSimulationConfig:
"""Zaman simülasyonu yapılandırması (Çin saat dilimi)"""
total_simulation_hours: int = 72 # Varsayılan 72 saat (3 gün)
minutes_per_round: int = 60 # Tur başına 60 dakika
# Saat başına aktif olan ajanlar
agents_per_hour_min: int = 5
agents_per_hour_max: int = 20
# Yoğun saatler (akşam 19-22, Çinliler en aktif)
peak_hours: List[int] = field(default_factory=lambda: [19, 20, 21, 22])
peak_activity_multiplier: float = 1.5
# Yoğun olmayan saatler (sabah erken 0-5, neredeyse hiç etkinlik yok)
off_peak_hours: List[int] = field(default_factory=lambda: [0, 1, 2, 3, 4, 5])
off_peak_activity_multiplier: float = 0.05
# Sabah saatleri
morning_hours: List[int] = field(default_factory=lambda: [6, 7, 8])
morning_activity_multiplier: float = 0.4
# Çalışma saatleri
work_hours: List[int] = field(default_factory=lambda: [9, 10, 11, 12, 13, 14, 15, 16, 17, 18])
work_activity_multiplier: float = 0.7
Eksiksiz Simülasyon Parametreleri
@dataclass
class SimulationParameters:
"""Eksiksiz simülasyon parametre yapılandırması"""
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,
}
Özet Tablosu: Ajan Türü Desenleri
| Ajan Türü | Etkinlik | Aktif Saatler | Gönderi/Saat | Yorum/Saat | Yanıt (dk) | Etki |
|---|---|---|---|---|---|---|
| Üniversite | 0.2 | 9-17 | 0.1 | 0.05 | 60-240 | 3.0 |
| DevletKurumu | 0.2 | 9-17 | 0.1 | 0.05 | 60-240 | 3.0 |
| MedyaKuruluşu | 0.5 | 7-23 | 0.8 | 0.3 | 5-30 | 2.5 |
| Profesör | 0.4 | 8-21 | 0.3 | 0.5 | 15-90 | 2.0 |
| Öğrenci | 0.8 | 8-12, 18-23 | 0.6 | 1.5 | 1-15 | 0.8 |
| Mezun | 0.6 | 12-13, 19-23 | 0.4 | 0.8 | 5-30 | 1.0 |
| Kişi (varsayılan) | 0.7 | 9-13, 18-23 | 0.5 | 1.2 | 2-20 | 1.0 |
Sonuç
LLM destekli yapılandırma oluşturma, aşağıdakilerin dikkatli bir şekilde ele alınmasını gerektirir:
- Adım adım oluşturma: Yönetilebilir aşamalara ayır (zaman → olaylar → ajanlar → platformlar)
- Toplu işleme: Bağlam sınırlamalarını önlemek için her grupta 15 ajanı işle
- JSON onarımı: Köşeli parantez eşleştirmesi ve dize kaçış karakterleriyle kesmeyi ele al
- Kural tabanlı geri dönüşler: LLM başarısız olduğunda mantıklı varsayılanlar sağla
- Türe özgü desenler: Farklı ajan türlerinin farklı etkinlik desenleri vardır
- Doğrulama ve düzeltme: Oluşturulan değerleri kontrol et ve sorunları düzelt (örn. agents_per_hour > total_agents)