import asyncio import logging import random from typing import Optional from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models.orm import Character, Message from app.services.hermes_client import hermes_client logger = logging.getLogger(__name__) class AgentScheduler: def __init__(self): self._running = False self._agents: dict[str, dict] = {} self._speaker_queue: list[str] = [] self._current_speaker_index: int = 0 def init_agents(self, characters: list[Character]): self._agents = {} self._speaker_queue = [] for char in characters: if char.hermes_profile: self._agents[char.id] = { "name": char.name, "profile": char.hermes_profile, "character_id": char.id, } self._speaker_queue.append(char.id) async def ask_agent( self, name: str, message: str, character_id: Optional[str] = None, ) -> str: profile_name = name.lower().replace(" ", "_").replace("·", "_") try: return await hermes_client.chat(profile_name, message) except Exception as e: logger.error(f"ask_agent failed for '{name}': {e}") return f"（{name}暂时无法回应）" async def dm_speak(self, content: str) -> str: return f"【主持人】：{content}" def get_next_speaker(self) -> Optional[str]: if not self._speaker_queue: return None idx = self._current_speaker_index % len(self._speaker_queue) self._current_speaker_index += 1 return self._speaker_queue[idx] def reset_speakers(self): self._current_speaker_index = 0 def set_speaker_order(self, order: list[str]): self._speaker_queue = [cid for cid in order if cid in self._agents] self._current_speaker_index = 0 async def ask_current_speaker( self, session: AsyncSession, script_id: str, context: str, ) -> Optional[dict]: speaker_id = self.get_next_speaker() if not speaker_id or speaker_id not in self._agents: return None agent = self._agents[speaker_id] message = f"""当前讨论内容： {context} 请以{agent['name']}的身份发言。保持在角色中，用中文回复2-4句话。""" try: response = await self.ask_agent(agent["name"], message, speaker_id) msg = Message( character_id=speaker_id, game_phase="round1_speak", msg_type="character", content=response.strip(), ) session.add(msg) await session.commit() await session.refresh(msg) return { "id": msg.id, "character_id": speaker_id, "character_name": agent["name"], "content": response.strip(), "msg_type": "character", "game_phase": "round1_speak", } except Exception as e: logger.error(f"Speaker error for {agent['name']}: {e}") return None async def run_auto_speaking( self, session: AsyncSession, script_id: str, interval: float = 8.0, rounds: int = 1, socket_emit=None, ): self._running = True characters_result = await session.execute( select(Character).where(Character.script_id == script_id) ) characters = list(characters_result.scalars().all()) self.init_agents(characters) for _ in range(rounds): if not self._running: break self.reset_speakers() for _ in range(len(self._speaker_queue)): if not self._running: break speaker_id = self.get_next_speaker() agent = self._agents.get(speaker_id) if not agent: continue msgs_result = await session.execute( select(Message).order_by(Message.created_at.desc()).limit(20) ) recent = list(msgs_result.scalars().all()) context = "\n".join([ f"{m.character.name if m.character else '系统'}: {m.content}" for m in reversed(recent) ]) try: response = await self.ask_agent(agent["name"], context, speaker_id) msg = Message( character_id=speaker_id, game_phase="round1_speak", msg_type="character", content=response.strip(), ) session.add(msg) await session.commit() await session.refresh(msg) if socket_emit: await socket_emit("new_message", { "id": msg.id, "character_id": speaker_id, "character_name": agent["name"], "content": response.strip(), "msg_type": "character", "game_phase": "round1_speak", }) await socket_emit("speaker_change", { "current_speaker_id": self._speaker_queue[ self._current_speaker_index % len(self._speaker_queue) ] if self._speaker_queue else None, }) except Exception as e: logger.error(f"Auto speak error for {agent['name']}: {e}") await asyncio.sleep(interval) async def trigger_voting( self, session: AsyncSession, script_id: str, socket_emit=None, ) -> list[dict]: characters_result = await session.execute( select(Character).where(Character.script_id == script_id) ) characters = list(characters_result.scalars().all()) msgs_result = await session.execute( select(Message).order_by(Message.created_at.desc()).limit(50) ) messages = list(msgs_result.scalars().all()) summary = "\n".join([ f"{m.character.name if m.character else '系统'}: {m.content[:100]}" for m in reversed(messages) ]) character_names = [c.name for c in characters] votes = [] for character in characters: if not character.hermes_profile: continue vote_prompt = f"""当前是投票阶段。以下是讨论摘要： {summary} 可用角色：{', '.join(n for n in character_names if n != character.name)} 请以{character.name}的身份投票选出你认为的凶手。只回复JSON格式： {{"target": "角色名", "reason": "投票理由"}}""" try: response = await self.ask_agent(character.name, vote_prompt, character.id) import json try: decision = json.loads(response) except json.JSONDecodeError: decision = {"target": "", "reason": response} target_name = decision.get("target", "") reason = decision.get("reason", "") target_char = next((c for c in characters if c.name == target_name), None) if not target_char: others = [c for c in characters if c.id != character.id] if others: target_char = random.choice(others) reason = "随机投票" if target_char: votes.append({ "voter_id": character.id, "voter_name": character.name, "target_id": target_char.id, "target_name": target_char.name, "reason": reason, }) if socket_emit: await socket_emit("vote_cast", { "voter_id": character.id, "voter_name": character.name, "target_id": target_char.id, "target_name": target_char.name, "reason": reason, }) except Exception as e: logger.error(f"Voting error for {character.name}: {e}") return votes def stop(self): self._running = False agent_scheduler = AgentScheduler()