From fb1228a679e1a01bd7d68770f31b2dda61e55bdd Mon Sep 17 00:00:00 2001 From: Atharva-1512 Date: Sat, 21 Mar 2026 14:05:16 +0530 Subject: [PATCH 1/2] Fix: prevent false keyword matches using regex word boundaries --- .history/docs/ARCHITECTURE_20260314000717.md | 86 ++++++++ .history/docs/ARCHITECTURE_20260314001429.md | 96 +++++++++ .history/rooms/session_20260314000717.py | 213 +++++++++++++++++++ .history/rooms/session_20260321140457.py | 213 +++++++++++++++++++ rooms/session.py | 2 +- 5 files changed, 609 insertions(+), 1 deletion(-) create mode 100644 .history/docs/ARCHITECTURE_20260314000717.md create mode 100644 .history/docs/ARCHITECTURE_20260314001429.md create mode 100644 .history/rooms/session_20260314000717.py create mode 100644 .history/rooms/session_20260321140457.py diff --git a/.history/docs/ARCHITECTURE_20260314000717.md b/.history/docs/ARCHITECTURE_20260314000717.md new file mode 100644 index 0000000..f9f60ca --- /dev/null +++ b/.history/docs/ARCHITECTURE_20260314000717.md @@ -0,0 +1,86 @@ +# Multi-Agent Rooms Architecture + +## How LiteLLM Works +**LiteLLM is a universal routing library, not an AI model or an API endpoint itself.** + +The Multi-Agent Rooms framework uses LiteLLM as an adapter. Different AI providers (OpenAI, Anthropic, local applications like Ollama) expect code formatted in entirely different ways. LiteLLM acts as a universal translator. The Rooms framework sends standard "messages" to LiteLLM, and LiteLLM translates them and routes them to the correct backend. + +### Is it Free? Does it require API Keys? +LiteLLM is a completely free, open-source python package. Because it's just a translator, it does not charge money or process your data externally. + +When you configure an agent's `model` to a string like `ollama/llama3`, LiteLLM recognizes the `ollama/` prefix. It explicitly **does not** send the request over the internet to a commercial provider. Instead, it routes the HTTP request to `http://localhost:11434`, which is the default port Ollama uses on your local machine. + +Because the request never leaves your computer, **there are no API keys required, no rate limits, and zero costs.** + +When the Agents reply to you in the terminal, it means your computer's local CPU/GPU is quietly processing the inference via the Ollama application running in your background! + +## Session Memory & Timestamps +All conversation history is held in RAM for the duration of the session. Each entry — agent turn, user message, and system introduction — is tagged with a `timestamp` (format: `YYYY-MM-DD HH:MM:SS`). This makes transcripts auditable without any external database. + +The history is accessible internally to all agents via `Session.get_agent_context()`, which formats it into the LLM-standard `role/content` format before passing it to each model. + +## Saving Transcripts +At the end of a session, the user is prompted to optionally save. Two formats are available: + +- **Markdown (`.md`)**: Human-readable transcript with speaker headings and timestamps. +- **CSV (`.csv`)**: Machine-parseable table with columns: `Timestamp`, `Speaker`, `Message`. + +The filename is auto-suggested from a short slug of the session topic (e.g. `bioethics_of_designer_babies.md`). System bootstrap messages are excluded from saved output. + +## User Profile & Participant Identity +The wizard captures a **user name and background** before the session starts. This is injected into the global session introduction, so all agents are explicitly informed of who the human participant is and can treat them as an equal voice in the room. + +## Smart Agent Selection + +The framework goes beyond simple rotation with an intelligent selection system for `dynamic` sessions. + +### Expertise-Weighted Scoring +Each `AgentConfig` has an `expertise` list (e.g., `["law", "contracts", "compliance"]`). In `dynamic` mode, before each turn the session scores all available agents against the last 5 messages in the history. The agent with the most matching keywords in the live conversation context is selected to speak next. + +This means a room with a lawyer and an engineer will naturally shift voice depending on whether the topic drifts toward legal frameworks or technical implementation. + +### User-Directed Addressing (`@AgentName`) +At any human input prompt, the user may type `@AgentName` to lock the next response to that specific agent. The `add_user_message()` method parses the message for `@mentions` as well as natural patterns (e.g., *"What does Elena think?"*) and stores the forced agent. It is consumed on the next `generate_next_turn()` call and then cleared. + +### PASS Mechanic +Agents can respond with just `PASS` if they genuinely have nothing meaningful to add. The session detects this, increments the turn count, marks the turn as `skipped: True`, and does not add it to the visible history. The CLI renders these silently. This prevents verbose filler text that degrades the quality of the transcript. + +### Early Human-In-The-Loop Trigger +Beyond the configured turn interval, the `needs_human_input()` method performs an additional check: if the **last agent message explicitly addresses the user by name**, the HITL prompt fires immediately. This ensures the conversation never inadvertently "speaks for" the human participant. + +**Decision Flow:** +``` +generate_next_turn() + ↓ +Orchestrator due? → speak (or PASS → skip) + ↓ +_forced_next_agent set? → use it, clear it + ↓ +session_type = DYNAMIC? + → @mention in last message? → forced agent + → Score all agents by expertise → pick best + → fallback: round robin + ↓ +agent.generate_response() + ↓ +response == "PASS"? → skip, return {skipped: True} + ↓ +append to history with timestamp, return turn_data +``` + +## Custom Model Integrations (Bring Your Own Code) + +If you do not want to use LiteLLM at all, the framework allows you to inject arbitrary Python scripts as the "brain" for an agent. + +1. Create a python file (e.g. `my_model.py`). +2. Write a function that accepts a `List[Dict[str, str]]` (the conversation history) and returns a `str` (the agent's reply). +3. In the CLI wizard, select `custom_function` as the Model Type. +4. Provide the path to `my_model.py` and the exact name of the function you wrote. + +The framework will dynamically import your file at runtime and use it exclusively for that agent's turns. + +## CI/CD and Robustness +To ensure the framework remains stable as it grows, we maintain a comprehensive CI/CD pipeline using **GitHub Actions**. Every contribution is automatically tested against Python 3.13 for: +- **Linting**: High-standard code hygiene via `flake8`. +- **Logic Robustness**: Detailed edge-case testing including word-boundary expertise matching and non-repeating HITL triggers. +- **Regression Testing**: Ensuring core orchestration types (Round Robin, Dynamic, Argumentative) remain deterministic. diff --git a/.history/docs/ARCHITECTURE_20260314001429.md b/.history/docs/ARCHITECTURE_20260314001429.md new file mode 100644 index 0000000..085254c --- /dev/null +++ b/.history/docs/ARCHITECTURE_20260314001429.md @@ -0,0 +1,96 @@ +# Multi-Agent Rooms Architecture + +## How LiteLLM Works +**LiteLLM is a universal routing library, not an AI model or an API endpoint itself.** + +The Multi-Agent Rooms framework uses LiteLLM as an adapter. Different AI providers (OpenAI, Anthropic, local applications like Ollama) expect code formatted in entirely different ways. LiteLLM acts as a universal translator. The Rooms framework sends standard "messages" to LiteLLM, and LiteLLM translates them and routes them to the correct backend. + +### Is it Free? Does it require API Keys? +LiteLLM is a completely free, open-source python package. Because it's just a translator, it does not charge money or process your data externally. + +When you configure an agent's `model` to a string like `ollama/llama3`, LiteLLM recognizes the `ollama/` prefix. It explicitly **does not** send the request over the internet to a commercial provider. Instead, it routes the HTTP request to `http://localhost:11434`, which is the default port Ollama uses on your local machine. + +Because the request never leaves your computer, **there are no API keys required, no rate limits, and zero costs.** + +When the Agents reply to you in the terminal, it means your computer's local CPU/GPU is quietly processing the inference via the Ollama application running in your background! + +## Session Memory & Timestamps +All conversation history is held in RAM for the duration of the session. Each entry — agent turn, user message, and system introduction — is tagged with a `timestamp` (format: `YYYY-MM-DD HH:MM:SS`). This makes transcripts auditable without any external database. + +The history is accessible internally to all agents via `Session.get_agent_context()`, which formats it into the LLM-standard `role/content` format before passing it to each model. + +## Saving Transcripts +At the end of a session, the user is prompted to optionally save. Two formats are available: + +- **Markdown (`.md`)**: Human-readable transcript with speaker headings and timestamps. +- **CSV (`.csv`)**: Machine-parseable table with columns: `Timestamp`, `Speaker`, `Message`. + +The filename is auto-suggested from a short slug of the session topic (e.g. `bioethics_of_designer_babies.md`). System bootstrap messages are excluded from saved output. + +## User Profile & Participant Identity +The wizard captures a **user name and background** before the session starts. This is injected into the global session introduction, so all agents are explicitly informed of who the human participant is and can treat them as an equal voice in the room. + +## Smart Agent Selection + +The framework goes beyond simple rotation with an intelligent selection system for `dynamic` sessions. + +### Expertise-Weighted Scoring +Each `AgentConfig` has an `expertise` list (e.g., `["law", "contracts", "compliance"]`). In `dynamic` mode, before each turn the session scores all available agents against the last 5 messages in the history. The agent with the most matching keywords in the live conversation context is selected to speak next. + +This means a room with a lawyer and an engineer will naturally shift voice depending on whether the topic drifts toward legal frameworks or technical implementation. + +### User-Directed Addressing (`@AgentName`) +At any human input prompt, the user may type `@AgentName` to lock the next response to that specific agent. The `add_user_message()` method parses the message for `@mentions` as well as natural patterns (e.g., *"What does Elena think?"*) and stores the forced agent. It is consumed on the next `generate_next_turn()` call and then cleared. + +### PASS Mechanic +Agents can respond with just `PASS` if they genuinely have nothing meaningful to add. The session detects this, increments the turn count, marks the turn as `skipped: True`, and does not add it to the visible history. The CLI renders these silently. This prevents verbose filler text that degrades the quality of the transcript. + +### Early Human-In-The-Loop Trigger +Beyond the configured turn interval, the `needs_human_input()` method performs an additional check: if the **last agent message explicitly addresses the user by name**, the HITL prompt fires immediately. This ensures the conversation never inadvertently "speaks for" the human participant. + +**Decision Flow:** +```mermaid +flowchart TD + +A[generate_next_turn()] --> B{Orchestrator due?} +B -->|Yes| C[Speak or PASS → skip] +B -->|No| D{_forced_next_agent set?} + +D -->|Yes| E[Use forced agent and clear flag] +D -->|No| F{session_type = DYNAMIC?} + +F -->|Yes| G{@mention in last message?} +G -->|Yes| H[Use mentioned agent] +G -->|No| I[Score agents by expertise] + +I --> J[Pick best agent] +J --> K[Fallback: round robin] + +F -->|No| K + +K --> L[agent.generate_response()] + +L --> M{response == PASS?} + +M -->|Yes| N[Skip turn return skipped true] +M -->|No| O[Append to history with timestamp] + +O --> P[Return turn_data] +``` + +## Custom Model Integrations (Bring Your Own Code) + +If you do not want to use LiteLLM at all, the framework allows you to inject arbitrary Python scripts as the "brain" for an agent. + +1. Create a python file (e.g. `my_model.py`). +2. Write a function that accepts a `List[Dict[str, str]]` (the conversation history) and returns a `str` (the agent's reply). +3. In the CLI wizard, select `custom_function` as the Model Type. +4. Provide the path to `my_model.py` and the exact name of the function you wrote. + +The framework will dynamically import your file at runtime and use it exclusively for that agent's turns. + +## CI/CD and Robustness +To ensure the framework remains stable as it grows, we maintain a comprehensive CI/CD pipeline using **GitHub Actions**. Every contribution is automatically tested against Python 3.13 for: +- **Linting**: High-standard code hygiene via `flake8`. +- **Logic Robustness**: Detailed edge-case testing including word-boundary expertise matching and non-repeating HITL triggers. +- **Regression Testing**: Ensuring core orchestration types (Round Robin, Dynamic, Argumentative) remain deterministic. diff --git a/.history/rooms/session_20260314000717.py b/.history/rooms/session_20260314000717.py new file mode 100644 index 0000000..8c2b7ce --- /dev/null +++ b/.history/rooms/session_20260314000717.py @@ -0,0 +1,213 @@ +import re +from datetime import datetime +from typing import List, Dict, Any, Optional + +from .config import SessionConfig, SessionType +from .agent import Agent + +def _now() -> str: + return datetime.now().strftime("%Y-%m-%d %H:%M:%S") + +def _score_agent_expertise(agent: Agent, context_text: str) -> int: + """Score an agent's expertise keywords against the context text. + Returns the number of matching keywords found.""" + if not agent.config.expertise: + return 0 + context_lower = context_text.lower() + score = 0 + for kw in agent.config.expertise: + if re.search(rf"\b{re.escape(kw.lower())}\b", context_lower): + score += 1 + return score + +def _find_forced_agent(message: str, agents: List[Agent]) -> Optional[Agent]: + """Detect if user explicitly addresses an agent by name via @Name or 'I want to hear X'. + Returns the matching Agent or None.""" + message_lower = message.lower() + + # Check for @AgentName shorthand + at_mentions = re.findall(r"@(\w[\w\s]*)", message, re.IGNORECASE) + for mention in at_mentions: + for agent in agents: + # Match on first name or full name (case-insensitive) + first_name = agent.name.split("(")[0].strip().lower() + if mention.strip().lower() in (agent.name.lower(), first_name): + return agent + + # Check for "I want to hear [Name]" or "What does [Name] think" patterns + for agent in agents: + first_name = agent.name.split("(")[0].strip() + if re.search(rf"\b{re.escape(first_name.lower())}\b", message_lower): + return agent + + return None + +def _user_is_addressed(message: str, user_profile: Optional[Dict[str, str]]) -> bool: + """Detect if the agent's response is directly addressing the user by name.""" + if not user_profile or not user_profile.get("name"): + return False + user_name = user_profile["name"].lower() + return bool(re.search(rf"\b{re.escape(user_name)}\b", message.lower())) + + +class Session: + def __init__(self, config: SessionConfig, agents: List[Agent], user_profile: Optional[Dict[str, str]] = None): + self.config = config + self.agents = agents + self.user_profile = user_profile # {"name": "...", "background": "..."} + self.history: List[Dict[str, str]] = [] + self.turn_count = 0 + self._last_orchestrator_turn = -1 + self._forced_next_agent: Optional[Agent] = None # Locked next agent from @mention or user direction + self._hitl_triggered = False # Track if early HITL was already triggered for current turn + + # Build global introduction including user if provided + agent_names = ', '.join([a.name for a in self.agents]) + user_intro = "" + if self.user_profile: + user_intro = f"\nUser Participant: {self.user_profile['name']} - {self.user_profile.get('background', '')}" + + self.global_intro = ( + f"The room is active. Topic: {self.config.topic}\n" + f"Participating Agents: {agent_names}{user_intro}\n" + f"Note: Agents may respond with 'PASS' if they have nothing meaningful to add." + ) + self.history.append({"role": "system", "content": self.global_intro, "timestamp": _now()}) + self._agent_index = 0 + + def add_user_message(self, username: str, message: str): + """Inject a human message into the session history, and detect @mention agent forcing.""" + self.history.append({"role": username, "content": message, "timestamp": _now()}) + + # Check if the user addressed a specific agent + forced = _find_forced_agent(message, self.agents) + if forced: + self._forced_next_agent = forced + + # Reset HITL trigger on user input + self._hitl_triggered = False + + def get_agent_context(self, current_agent: Agent) -> List[Dict[str, str]]: + """Format history into an LLM context including system prompt.""" + context = [] + for msg in self.history: + role = "user" + if msg["role"] == current_agent.name: + role = "assistant" + elif msg["role"] == "system": + role = "system" + + content = msg["content"] + if msg["role"] not in (current_agent.name, "system"): + content = f"[{msg['role']} said]: {content}" + + context.append({"role": role, "content": content}) + return context + + def generate_next_turn(self) -> Optional[Dict[str, str]]: + """Determine next agent, get response, and log it.""" + if self.turn_count >= self.config.max_turns: + return None + + # Optional Orchestrator Intervention (every 3 turns, once per interval) + if ( + self.config.orchestrator + and self.turn_count > 0 + and self.turn_count % 3 == 0 + and self._last_orchestrator_turn != self.turn_count + ): + self._last_orchestrator_turn = self.turn_count + orch_agent = Agent(self.config.orchestrator) + context = self.get_agent_context(orch_agent) + msg = orch_agent.generate_response(context) + self.turn_count += 1 + if "PASS" not in msg: + turn_data = { + "role": f"Orchestrator ({orch_agent.name})", + "content": msg, + "color": orch_agent.config.color, + "timestamp": _now() + } + self.history.append(turn_data) + return turn_data + # Orchestrator passed — fall through to normal turn + + # Resolve agent: forced @mention > smart selection + if self._forced_next_agent: + agent = self._forced_next_agent + self._forced_next_agent = None + else: + agent = self._select_next_agent() + + context = self.get_agent_context(agent) + response_text = agent.generate_response(context) + + # Handle PASS: agent has nothing to add — silently skip turn + if response_text.strip().upper() == "PASS": + self.turn_count += 1 + return {"role": agent.name, "content": "PASS", "color": agent.config.color, "timestamp": _now(), "skipped": True} + + turn_data = { + "role": agent.name, + "content": response_text, + "color": agent.config.color, + "timestamp": _now() + } + self.history.append(turn_data) + self.turn_count += 1 + return turn_data + + def _select_next_agent(self) -> Agent: + """Select who talks next based on SessionType and expertise scoring.""" + if self.config.session_type == SessionType.ROUND_ROBIN: + agent = self.agents[self._agent_index] + self._agent_index = (self._agent_index + 1) % len(self.agents) + return agent + + elif self.config.session_type == SessionType.ARGUMENTATIVE: + if len(self.agents) >= 2: + return self.agents[self.turn_count % 2] + return self.agents[0] + + elif self.config.session_type == SessionType.DYNAMIC: + # Build context text from recent history for scoring + recent = " ".join(m["content"] for m in self.history[-5:]) + + # 1. Check for @mention or name reference in last user/agent message + if self.history: + last_content = self.history[-1].get("content", "") + forced = _find_forced_agent(last_content, self.agents) + if forced: + return forced + + # 2. Score all agents by expertise relevance to recent context + scored = [(agent, _score_agent_expertise(agent, recent)) for agent in self.agents] + scored.sort(key=lambda x: x[1], reverse=True) + + # If top scorer has meaningful score AND is NOT the agent that just spoke, pick them + if scored and scored[0][1] > 0: + last_speaker = self.history[-1].get("role", "") if self.history else "" + for candidate, score in scored: + if candidate.name != last_speaker: + return candidate + + # 3. Fallback to round-robin + agent = self.agents[self._agent_index] + self._agent_index = (self._agent_index + 1) % len(self.agents) + return agent + + return self.agents[0] + + def needs_human_input(self) -> bool: + """Check if it's time for human input, including early trigger if user is directly addressed.""" + if self.config.human_in_the_loop_turns <= 0: + return False + + # Early HITL: if the last agent's message addressed the user by name + if not self._hitl_triggered and len(self.history) > 1: + last = self.history[-1] + if last.get("role") not in ("system",) and _user_is_addressed(last.get("content", ""), self.user_profile): + self._hitl_triggered = True + return True + + return self.turn_count > 0 and self.turn_count % self.config.human_in_the_loop_turns == 0 diff --git a/.history/rooms/session_20260321140457.py b/.history/rooms/session_20260321140457.py new file mode 100644 index 0000000..0cb0ffc --- /dev/null +++ b/.history/rooms/session_20260321140457.py @@ -0,0 +1,213 @@ +import re +from datetime import datetime +from typing import List, Dict, Any, Optional + +from .config import SessionConfig, SessionType +from .agent import Agent + +def _now() -> str: + return datetime.now().strftime("%Y-%m-%d %H:%M:%S") + +def _score_agent_expertise(agent: Agent, context_text: str) -> int: + """Score an agent's expertise keywords against the context text. + Returns the number of matching keywords found.""" + if not agent.config.expertise: + return 0 + context_lower = context_text.lower() + score = 0 + for kw in agent.config.expertise: + if re.search(rf"\b{re.escape(kw)}\b", context_lower, re.IGNORECASE): + score += 1 + return score + +def _find_forced_agent(message: str, agents: List[Agent]) -> Optional[Agent]: + """Detect if user explicitly addresses an agent by name via @Name or 'I want to hear X'. + Returns the matching Agent or None.""" + message_lower = message.lower() + + # Check for @AgentName shorthand + at_mentions = re.findall(r"@(\w[\w\s]*)", message, re.IGNORECASE) + for mention in at_mentions: + for agent in agents: + # Match on first name or full name (case-insensitive) + first_name = agent.name.split("(")[0].strip().lower() + if mention.strip().lower() in (agent.name.lower(), first_name): + return agent + + # Check for "I want to hear [Name]" or "What does [Name] think" patterns + for agent in agents: + first_name = agent.name.split("(")[0].strip() + if re.search(rf"\b{re.escape(first_name.lower())}\b", message_lower): + return agent + + return None + +def _user_is_addressed(message: str, user_profile: Optional[Dict[str, str]]) -> bool: + """Detect if the agent's response is directly addressing the user by name.""" + if not user_profile or not user_profile.get("name"): + return False + user_name = user_profile["name"].lower() + return bool(re.search(rf"\b{re.escape(user_name)}\b", message.lower())) + + +class Session: + def __init__(self, config: SessionConfig, agents: List[Agent], user_profile: Optional[Dict[str, str]] = None): + self.config = config + self.agents = agents + self.user_profile = user_profile # {"name": "...", "background": "..."} + self.history: List[Dict[str, str]] = [] + self.turn_count = 0 + self._last_orchestrator_turn = -1 + self._forced_next_agent: Optional[Agent] = None # Locked next agent from @mention or user direction + self._hitl_triggered = False # Track if early HITL was already triggered for current turn + + # Build global introduction including user if provided + agent_names = ', '.join([a.name for a in self.agents]) + user_intro = "" + if self.user_profile: + user_intro = f"\nUser Participant: {self.user_profile['name']} - {self.user_profile.get('background', '')}" + + self.global_intro = ( + f"The room is active. Topic: {self.config.topic}\n" + f"Participating Agents: {agent_names}{user_intro}\n" + f"Note: Agents may respond with 'PASS' if they have nothing meaningful to add." + ) + self.history.append({"role": "system", "content": self.global_intro, "timestamp": _now()}) + self._agent_index = 0 + + def add_user_message(self, username: str, message: str): + """Inject a human message into the session history, and detect @mention agent forcing.""" + self.history.append({"role": username, "content": message, "timestamp": _now()}) + + # Check if the user addressed a specific agent + forced = _find_forced_agent(message, self.agents) + if forced: + self._forced_next_agent = forced + + # Reset HITL trigger on user input + self._hitl_triggered = False + + def get_agent_context(self, current_agent: Agent) -> List[Dict[str, str]]: + """Format history into an LLM context including system prompt.""" + context = [] + for msg in self.history: + role = "user" + if msg["role"] == current_agent.name: + role = "assistant" + elif msg["role"] == "system": + role = "system" + + content = msg["content"] + if msg["role"] not in (current_agent.name, "system"): + content = f"[{msg['role']} said]: {content}" + + context.append({"role": role, "content": content}) + return context + + def generate_next_turn(self) -> Optional[Dict[str, str]]: + """Determine next agent, get response, and log it.""" + if self.turn_count >= self.config.max_turns: + return None + + # Optional Orchestrator Intervention (every 3 turns, once per interval) + if ( + self.config.orchestrator + and self.turn_count > 0 + and self.turn_count % 3 == 0 + and self._last_orchestrator_turn != self.turn_count + ): + self._last_orchestrator_turn = self.turn_count + orch_agent = Agent(self.config.orchestrator) + context = self.get_agent_context(orch_agent) + msg = orch_agent.generate_response(context) + self.turn_count += 1 + if "PASS" not in msg: + turn_data = { + "role": f"Orchestrator ({orch_agent.name})", + "content": msg, + "color": orch_agent.config.color, + "timestamp": _now() + } + self.history.append(turn_data) + return turn_data + # Orchestrator passed — fall through to normal turn + + # Resolve agent: forced @mention > smart selection + if self._forced_next_agent: + agent = self._forced_next_agent + self._forced_next_agent = None + else: + agent = self._select_next_agent() + + context = self.get_agent_context(agent) + response_text = agent.generate_response(context) + + # Handle PASS: agent has nothing to add — silently skip turn + if response_text.strip().upper() == "PASS": + self.turn_count += 1 + return {"role": agent.name, "content": "PASS", "color": agent.config.color, "timestamp": _now(), "skipped": True} + + turn_data = { + "role": agent.name, + "content": response_text, + "color": agent.config.color, + "timestamp": _now() + } + self.history.append(turn_data) + self.turn_count += 1 + return turn_data + + def _select_next_agent(self) -> Agent: + """Select who talks next based on SessionType and expertise scoring.""" + if self.config.session_type == SessionType.ROUND_ROBIN: + agent = self.agents[self._agent_index] + self._agent_index = (self._agent_index + 1) % len(self.agents) + return agent + + elif self.config.session_type == SessionType.ARGUMENTATIVE: + if len(self.agents) >= 2: + return self.agents[self.turn_count % 2] + return self.agents[0] + + elif self.config.session_type == SessionType.DYNAMIC: + # Build context text from recent history for scoring + recent = " ".join(m["content"] for m in self.history[-5:]) + + # 1. Check for @mention or name reference in last user/agent message + if self.history: + last_content = self.history[-1].get("content", "") + forced = _find_forced_agent(last_content, self.agents) + if forced: + return forced + + # 2. Score all agents by expertise relevance to recent context + scored = [(agent, _score_agent_expertise(agent, recent)) for agent in self.agents] + scored.sort(key=lambda x: x[1], reverse=True) + + # If top scorer has meaningful score AND is NOT the agent that just spoke, pick them + if scored and scored[0][1] > 0: + last_speaker = self.history[-1].get("role", "") if self.history else "" + for candidate, score in scored: + if candidate.name != last_speaker: + return candidate + + # 3. Fallback to round-robin + agent = self.agents[self._agent_index] + self._agent_index = (self._agent_index + 1) % len(self.agents) + return agent + + return self.agents[0] + + def needs_human_input(self) -> bool: + """Check if it's time for human input, including early trigger if user is directly addressed.""" + if self.config.human_in_the_loop_turns <= 0: + return False + + # Early HITL: if the last agent's message addressed the user by name + if not self._hitl_triggered and len(self.history) > 1: + last = self.history[-1] + if last.get("role") not in ("system",) and _user_is_addressed(last.get("content", ""), self.user_profile): + self._hitl_triggered = True + return True + + return self.turn_count > 0 and self.turn_count % self.config.human_in_the_loop_turns == 0 diff --git a/rooms/session.py b/rooms/session.py index 8c2b7ce..0cb0ffc 100644 --- a/rooms/session.py +++ b/rooms/session.py @@ -16,7 +16,7 @@ def _score_agent_expertise(agent: Agent, context_text: str) -> int: context_lower = context_text.lower() score = 0 for kw in agent.config.expertise: - if re.search(rf"\b{re.escape(kw.lower())}\b", context_lower): + if re.search(rf"\b{re.escape(kw)}\b", context_lower, re.IGNORECASE): score += 1 return score From 3e87212ecbc433ad64fc7bb6e1f3cb815c82baf9 Mon Sep 17 00:00:00 2001 From: Atharva-1512 Date: Sat, 21 Mar 2026 14:07:08 +0530 Subject: [PATCH 2/2] Cleanup: remove unnecessary history files --- .gitignore | 4 +- .history/docs/ARCHITECTURE_20260314000717.md | 86 -------- .history/docs/ARCHITECTURE_20260314001429.md | 96 --------- .history/rooms/session_20260314000717.py | 213 ------------------- .history/rooms/session_20260321140457.py | 213 ------------------- 5 files changed, 3 insertions(+), 609 deletions(-) delete mode 100644 .history/docs/ARCHITECTURE_20260314000717.md delete mode 100644 .history/docs/ARCHITECTURE_20260314001429.md delete mode 100644 .history/rooms/session_20260314000717.py delete mode 100644 .history/rooms/session_20260321140457.py diff --git a/.gitignore b/.gitignore index 3a56287..1dae7bb 100644 --- a/.gitignore +++ b/.gitignore @@ -20,4 +20,6 @@ Thumbs.db # Project Outputs outputs/* -!outputs/.gitkeep \ No newline at end of file +!outputs/.gitkeep + +.history/ \ No newline at end of file diff --git a/.history/docs/ARCHITECTURE_20260314000717.md b/.history/docs/ARCHITECTURE_20260314000717.md deleted file mode 100644 index f9f60ca..0000000 --- a/.history/docs/ARCHITECTURE_20260314000717.md +++ /dev/null @@ -1,86 +0,0 @@ -# Multi-Agent Rooms Architecture - -## How LiteLLM Works -**LiteLLM is a universal routing library, not an AI model or an API endpoint itself.** - -The Multi-Agent Rooms framework uses LiteLLM as an adapter. Different AI providers (OpenAI, Anthropic, local applications like Ollama) expect code formatted in entirely different ways. LiteLLM acts as a universal translator. The Rooms framework sends standard "messages" to LiteLLM, and LiteLLM translates them and routes them to the correct backend. - -### Is it Free? Does it require API Keys? -LiteLLM is a completely free, open-source python package. Because it's just a translator, it does not charge money or process your data externally. - -When you configure an agent's `model` to a string like `ollama/llama3`, LiteLLM recognizes the `ollama/` prefix. It explicitly **does not** send the request over the internet to a commercial provider. Instead, it routes the HTTP request to `http://localhost:11434`, which is the default port Ollama uses on your local machine. - -Because the request never leaves your computer, **there are no API keys required, no rate limits, and zero costs.** - -When the Agents reply to you in the terminal, it means your computer's local CPU/GPU is quietly processing the inference via the Ollama application running in your background! - -## Session Memory & Timestamps -All conversation history is held in RAM for the duration of the session. Each entry — agent turn, user message, and system introduction — is tagged with a `timestamp` (format: `YYYY-MM-DD HH:MM:SS`). This makes transcripts auditable without any external database. - -The history is accessible internally to all agents via `Session.get_agent_context()`, which formats it into the LLM-standard `role/content` format before passing it to each model. - -## Saving Transcripts -At the end of a session, the user is prompted to optionally save. Two formats are available: - -- **Markdown (`.md`)**: Human-readable transcript with speaker headings and timestamps. -- **CSV (`.csv`)**: Machine-parseable table with columns: `Timestamp`, `Speaker`, `Message`. - -The filename is auto-suggested from a short slug of the session topic (e.g. `bioethics_of_designer_babies.md`). System bootstrap messages are excluded from saved output. - -## User Profile & Participant Identity -The wizard captures a **user name and background** before the session starts. This is injected into the global session introduction, so all agents are explicitly informed of who the human participant is and can treat them as an equal voice in the room. - -## Smart Agent Selection - -The framework goes beyond simple rotation with an intelligent selection system for `dynamic` sessions. - -### Expertise-Weighted Scoring -Each `AgentConfig` has an `expertise` list (e.g., `["law", "contracts", "compliance"]`). In `dynamic` mode, before each turn the session scores all available agents against the last 5 messages in the history. The agent with the most matching keywords in the live conversation context is selected to speak next. - -This means a room with a lawyer and an engineer will naturally shift voice depending on whether the topic drifts toward legal frameworks or technical implementation. - -### User-Directed Addressing (`@AgentName`) -At any human input prompt, the user may type `@AgentName` to lock the next response to that specific agent. The `add_user_message()` method parses the message for `@mentions` as well as natural patterns (e.g., *"What does Elena think?"*) and stores the forced agent. It is consumed on the next `generate_next_turn()` call and then cleared. - -### PASS Mechanic -Agents can respond with just `PASS` if they genuinely have nothing meaningful to add. The session detects this, increments the turn count, marks the turn as `skipped: True`, and does not add it to the visible history. The CLI renders these silently. This prevents verbose filler text that degrades the quality of the transcript. - -### Early Human-In-The-Loop Trigger -Beyond the configured turn interval, the `needs_human_input()` method performs an additional check: if the **last agent message explicitly addresses the user by name**, the HITL prompt fires immediately. This ensures the conversation never inadvertently "speaks for" the human participant. - -**Decision Flow:** -``` -generate_next_turn() - ↓ -Orchestrator due? → speak (or PASS → skip) - ↓ -_forced_next_agent set? → use it, clear it - ↓ -session_type = DYNAMIC? - → @mention in last message? → forced agent - → Score all agents by expertise → pick best - → fallback: round robin - ↓ -agent.generate_response() - ↓ -response == "PASS"? → skip, return {skipped: True} - ↓ -append to history with timestamp, return turn_data -``` - -## Custom Model Integrations (Bring Your Own Code) - -If you do not want to use LiteLLM at all, the framework allows you to inject arbitrary Python scripts as the "brain" for an agent. - -1. Create a python file (e.g. `my_model.py`). -2. Write a function that accepts a `List[Dict[str, str]]` (the conversation history) and returns a `str` (the agent's reply). -3. In the CLI wizard, select `custom_function` as the Model Type. -4. Provide the path to `my_model.py` and the exact name of the function you wrote. - -The framework will dynamically import your file at runtime and use it exclusively for that agent's turns. - -## CI/CD and Robustness -To ensure the framework remains stable as it grows, we maintain a comprehensive CI/CD pipeline using **GitHub Actions**. Every contribution is automatically tested against Python 3.13 for: -- **Linting**: High-standard code hygiene via `flake8`. -- **Logic Robustness**: Detailed edge-case testing including word-boundary expertise matching and non-repeating HITL triggers. -- **Regression Testing**: Ensuring core orchestration types (Round Robin, Dynamic, Argumentative) remain deterministic. diff --git a/.history/docs/ARCHITECTURE_20260314001429.md b/.history/docs/ARCHITECTURE_20260314001429.md deleted file mode 100644 index 085254c..0000000 --- a/.history/docs/ARCHITECTURE_20260314001429.md +++ /dev/null @@ -1,96 +0,0 @@ -# Multi-Agent Rooms Architecture - -## How LiteLLM Works -**LiteLLM is a universal routing library, not an AI model or an API endpoint itself.** - -The Multi-Agent Rooms framework uses LiteLLM as an adapter. Different AI providers (OpenAI, Anthropic, local applications like Ollama) expect code formatted in entirely different ways. LiteLLM acts as a universal translator. The Rooms framework sends standard "messages" to LiteLLM, and LiteLLM translates them and routes them to the correct backend. - -### Is it Free? Does it require API Keys? -LiteLLM is a completely free, open-source python package. Because it's just a translator, it does not charge money or process your data externally. - -When you configure an agent's `model` to a string like `ollama/llama3`, LiteLLM recognizes the `ollama/` prefix. It explicitly **does not** send the request over the internet to a commercial provider. Instead, it routes the HTTP request to `http://localhost:11434`, which is the default port Ollama uses on your local machine. - -Because the request never leaves your computer, **there are no API keys required, no rate limits, and zero costs.** - -When the Agents reply to you in the terminal, it means your computer's local CPU/GPU is quietly processing the inference via the Ollama application running in your background! - -## Session Memory & Timestamps -All conversation history is held in RAM for the duration of the session. Each entry — agent turn, user message, and system introduction — is tagged with a `timestamp` (format: `YYYY-MM-DD HH:MM:SS`). This makes transcripts auditable without any external database. - -The history is accessible internally to all agents via `Session.get_agent_context()`, which formats it into the LLM-standard `role/content` format before passing it to each model. - -## Saving Transcripts -At the end of a session, the user is prompted to optionally save. Two formats are available: - -- **Markdown (`.md`)**: Human-readable transcript with speaker headings and timestamps. -- **CSV (`.csv`)**: Machine-parseable table with columns: `Timestamp`, `Speaker`, `Message`. - -The filename is auto-suggested from a short slug of the session topic (e.g. `bioethics_of_designer_babies.md`). System bootstrap messages are excluded from saved output. - -## User Profile & Participant Identity -The wizard captures a **user name and background** before the session starts. This is injected into the global session introduction, so all agents are explicitly informed of who the human participant is and can treat them as an equal voice in the room. - -## Smart Agent Selection - -The framework goes beyond simple rotation with an intelligent selection system for `dynamic` sessions. - -### Expertise-Weighted Scoring -Each `AgentConfig` has an `expertise` list (e.g., `["law", "contracts", "compliance"]`). In `dynamic` mode, before each turn the session scores all available agents against the last 5 messages in the history. The agent with the most matching keywords in the live conversation context is selected to speak next. - -This means a room with a lawyer and an engineer will naturally shift voice depending on whether the topic drifts toward legal frameworks or technical implementation. - -### User-Directed Addressing (`@AgentName`) -At any human input prompt, the user may type `@AgentName` to lock the next response to that specific agent. The `add_user_message()` method parses the message for `@mentions` as well as natural patterns (e.g., *"What does Elena think?"*) and stores the forced agent. It is consumed on the next `generate_next_turn()` call and then cleared. - -### PASS Mechanic -Agents can respond with just `PASS` if they genuinely have nothing meaningful to add. The session detects this, increments the turn count, marks the turn as `skipped: True`, and does not add it to the visible history. The CLI renders these silently. This prevents verbose filler text that degrades the quality of the transcript. - -### Early Human-In-The-Loop Trigger -Beyond the configured turn interval, the `needs_human_input()` method performs an additional check: if the **last agent message explicitly addresses the user by name**, the HITL prompt fires immediately. This ensures the conversation never inadvertently "speaks for" the human participant. - -**Decision Flow:** -```mermaid -flowchart TD - -A[generate_next_turn()] --> B{Orchestrator due?} -B -->|Yes| C[Speak or PASS → skip] -B -->|No| D{_forced_next_agent set?} - -D -->|Yes| E[Use forced agent and clear flag] -D -->|No| F{session_type = DYNAMIC?} - -F -->|Yes| G{@mention in last message?} -G -->|Yes| H[Use mentioned agent] -G -->|No| I[Score agents by expertise] - -I --> J[Pick best agent] -J --> K[Fallback: round robin] - -F -->|No| K - -K --> L[agent.generate_response()] - -L --> M{response == PASS?} - -M -->|Yes| N[Skip turn return skipped true] -M -->|No| O[Append to history with timestamp] - -O --> P[Return turn_data] -``` - -## Custom Model Integrations (Bring Your Own Code) - -If you do not want to use LiteLLM at all, the framework allows you to inject arbitrary Python scripts as the "brain" for an agent. - -1. Create a python file (e.g. `my_model.py`). -2. Write a function that accepts a `List[Dict[str, str]]` (the conversation history) and returns a `str` (the agent's reply). -3. In the CLI wizard, select `custom_function` as the Model Type. -4. Provide the path to `my_model.py` and the exact name of the function you wrote. - -The framework will dynamically import your file at runtime and use it exclusively for that agent's turns. - -## CI/CD and Robustness -To ensure the framework remains stable as it grows, we maintain a comprehensive CI/CD pipeline using **GitHub Actions**. Every contribution is automatically tested against Python 3.13 for: -- **Linting**: High-standard code hygiene via `flake8`. -- **Logic Robustness**: Detailed edge-case testing including word-boundary expertise matching and non-repeating HITL triggers. -- **Regression Testing**: Ensuring core orchestration types (Round Robin, Dynamic, Argumentative) remain deterministic. diff --git a/.history/rooms/session_20260314000717.py b/.history/rooms/session_20260314000717.py deleted file mode 100644 index 8c2b7ce..0000000 --- a/.history/rooms/session_20260314000717.py +++ /dev/null @@ -1,213 +0,0 @@ -import re -from datetime import datetime -from typing import List, Dict, Any, Optional - -from .config import SessionConfig, SessionType -from .agent import Agent - -def _now() -> str: - return datetime.now().strftime("%Y-%m-%d %H:%M:%S") - -def _score_agent_expertise(agent: Agent, context_text: str) -> int: - """Score an agent's expertise keywords against the context text. - Returns the number of matching keywords found.""" - if not agent.config.expertise: - return 0 - context_lower = context_text.lower() - score = 0 - for kw in agent.config.expertise: - if re.search(rf"\b{re.escape(kw.lower())}\b", context_lower): - score += 1 - return score - -def _find_forced_agent(message: str, agents: List[Agent]) -> Optional[Agent]: - """Detect if user explicitly addresses an agent by name via @Name or 'I want to hear X'. - Returns the matching Agent or None.""" - message_lower = message.lower() - - # Check for @AgentName shorthand - at_mentions = re.findall(r"@(\w[\w\s]*)", message, re.IGNORECASE) - for mention in at_mentions: - for agent in agents: - # Match on first name or full name (case-insensitive) - first_name = agent.name.split("(")[0].strip().lower() - if mention.strip().lower() in (agent.name.lower(), first_name): - return agent - - # Check for "I want to hear [Name]" or "What does [Name] think" patterns - for agent in agents: - first_name = agent.name.split("(")[0].strip() - if re.search(rf"\b{re.escape(first_name.lower())}\b", message_lower): - return agent - - return None - -def _user_is_addressed(message: str, user_profile: Optional[Dict[str, str]]) -> bool: - """Detect if the agent's response is directly addressing the user by name.""" - if not user_profile or not user_profile.get("name"): - return False - user_name = user_profile["name"].lower() - return bool(re.search(rf"\b{re.escape(user_name)}\b", message.lower())) - - -class Session: - def __init__(self, config: SessionConfig, agents: List[Agent], user_profile: Optional[Dict[str, str]] = None): - self.config = config - self.agents = agents - self.user_profile = user_profile # {"name": "...", "background": "..."} - self.history: List[Dict[str, str]] = [] - self.turn_count = 0 - self._last_orchestrator_turn = -1 - self._forced_next_agent: Optional[Agent] = None # Locked next agent from @mention or user direction - self._hitl_triggered = False # Track if early HITL was already triggered for current turn - - # Build global introduction including user if provided - agent_names = ', '.join([a.name for a in self.agents]) - user_intro = "" - if self.user_profile: - user_intro = f"\nUser Participant: {self.user_profile['name']} - {self.user_profile.get('background', '')}" - - self.global_intro = ( - f"The room is active. Topic: {self.config.topic}\n" - f"Participating Agents: {agent_names}{user_intro}\n" - f"Note: Agents may respond with 'PASS' if they have nothing meaningful to add." - ) - self.history.append({"role": "system", "content": self.global_intro, "timestamp": _now()}) - self._agent_index = 0 - - def add_user_message(self, username: str, message: str): - """Inject a human message into the session history, and detect @mention agent forcing.""" - self.history.append({"role": username, "content": message, "timestamp": _now()}) - - # Check if the user addressed a specific agent - forced = _find_forced_agent(message, self.agents) - if forced: - self._forced_next_agent = forced - - # Reset HITL trigger on user input - self._hitl_triggered = False - - def get_agent_context(self, current_agent: Agent) -> List[Dict[str, str]]: - """Format history into an LLM context including system prompt.""" - context = [] - for msg in self.history: - role = "user" - if msg["role"] == current_agent.name: - role = "assistant" - elif msg["role"] == "system": - role = "system" - - content = msg["content"] - if msg["role"] not in (current_agent.name, "system"): - content = f"[{msg['role']} said]: {content}" - - context.append({"role": role, "content": content}) - return context - - def generate_next_turn(self) -> Optional[Dict[str, str]]: - """Determine next agent, get response, and log it.""" - if self.turn_count >= self.config.max_turns: - return None - - # Optional Orchestrator Intervention (every 3 turns, once per interval) - if ( - self.config.orchestrator - and self.turn_count > 0 - and self.turn_count % 3 == 0 - and self._last_orchestrator_turn != self.turn_count - ): - self._last_orchestrator_turn = self.turn_count - orch_agent = Agent(self.config.orchestrator) - context = self.get_agent_context(orch_agent) - msg = orch_agent.generate_response(context) - self.turn_count += 1 - if "PASS" not in msg: - turn_data = { - "role": f"Orchestrator ({orch_agent.name})", - "content": msg, - "color": orch_agent.config.color, - "timestamp": _now() - } - self.history.append(turn_data) - return turn_data - # Orchestrator passed — fall through to normal turn - - # Resolve agent: forced @mention > smart selection - if self._forced_next_agent: - agent = self._forced_next_agent - self._forced_next_agent = None - else: - agent = self._select_next_agent() - - context = self.get_agent_context(agent) - response_text = agent.generate_response(context) - - # Handle PASS: agent has nothing to add — silently skip turn - if response_text.strip().upper() == "PASS": - self.turn_count += 1 - return {"role": agent.name, "content": "PASS", "color": agent.config.color, "timestamp": _now(), "skipped": True} - - turn_data = { - "role": agent.name, - "content": response_text, - "color": agent.config.color, - "timestamp": _now() - } - self.history.append(turn_data) - self.turn_count += 1 - return turn_data - - def _select_next_agent(self) -> Agent: - """Select who talks next based on SessionType and expertise scoring.""" - if self.config.session_type == SessionType.ROUND_ROBIN: - agent = self.agents[self._agent_index] - self._agent_index = (self._agent_index + 1) % len(self.agents) - return agent - - elif self.config.session_type == SessionType.ARGUMENTATIVE: - if len(self.agents) >= 2: - return self.agents[self.turn_count % 2] - return self.agents[0] - - elif self.config.session_type == SessionType.DYNAMIC: - # Build context text from recent history for scoring - recent = " ".join(m["content"] for m in self.history[-5:]) - - # 1. Check for @mention or name reference in last user/agent message - if self.history: - last_content = self.history[-1].get("content", "") - forced = _find_forced_agent(last_content, self.agents) - if forced: - return forced - - # 2. Score all agents by expertise relevance to recent context - scored = [(agent, _score_agent_expertise(agent, recent)) for agent in self.agents] - scored.sort(key=lambda x: x[1], reverse=True) - - # If top scorer has meaningful score AND is NOT the agent that just spoke, pick them - if scored and scored[0][1] > 0: - last_speaker = self.history[-1].get("role", "") if self.history else "" - for candidate, score in scored: - if candidate.name != last_speaker: - return candidate - - # 3. Fallback to round-robin - agent = self.agents[self._agent_index] - self._agent_index = (self._agent_index + 1) % len(self.agents) - return agent - - return self.agents[0] - - def needs_human_input(self) -> bool: - """Check if it's time for human input, including early trigger if user is directly addressed.""" - if self.config.human_in_the_loop_turns <= 0: - return False - - # Early HITL: if the last agent's message addressed the user by name - if not self._hitl_triggered and len(self.history) > 1: - last = self.history[-1] - if last.get("role") not in ("system",) and _user_is_addressed(last.get("content", ""), self.user_profile): - self._hitl_triggered = True - return True - - return self.turn_count > 0 and self.turn_count % self.config.human_in_the_loop_turns == 0 diff --git a/.history/rooms/session_20260321140457.py b/.history/rooms/session_20260321140457.py deleted file mode 100644 index 0cb0ffc..0000000 --- a/.history/rooms/session_20260321140457.py +++ /dev/null @@ -1,213 +0,0 @@ -import re -from datetime import datetime -from typing import List, Dict, Any, Optional - -from .config import SessionConfig, SessionType -from .agent import Agent - -def _now() -> str: - return datetime.now().strftime("%Y-%m-%d %H:%M:%S") - -def _score_agent_expertise(agent: Agent, context_text: str) -> int: - """Score an agent's expertise keywords against the context text. - Returns the number of matching keywords found.""" - if not agent.config.expertise: - return 0 - context_lower = context_text.lower() - score = 0 - for kw in agent.config.expertise: - if re.search(rf"\b{re.escape(kw)}\b", context_lower, re.IGNORECASE): - score += 1 - return score - -def _find_forced_agent(message: str, agents: List[Agent]) -> Optional[Agent]: - """Detect if user explicitly addresses an agent by name via @Name or 'I want to hear X'. - Returns the matching Agent or None.""" - message_lower = message.lower() - - # Check for @AgentName shorthand - at_mentions = re.findall(r"@(\w[\w\s]*)", message, re.IGNORECASE) - for mention in at_mentions: - for agent in agents: - # Match on first name or full name (case-insensitive) - first_name = agent.name.split("(")[0].strip().lower() - if mention.strip().lower() in (agent.name.lower(), first_name): - return agent - - # Check for "I want to hear [Name]" or "What does [Name] think" patterns - for agent in agents: - first_name = agent.name.split("(")[0].strip() - if re.search(rf"\b{re.escape(first_name.lower())}\b", message_lower): - return agent - - return None - -def _user_is_addressed(message: str, user_profile: Optional[Dict[str, str]]) -> bool: - """Detect if the agent's response is directly addressing the user by name.""" - if not user_profile or not user_profile.get("name"): - return False - user_name = user_profile["name"].lower() - return bool(re.search(rf"\b{re.escape(user_name)}\b", message.lower())) - - -class Session: - def __init__(self, config: SessionConfig, agents: List[Agent], user_profile: Optional[Dict[str, str]] = None): - self.config = config - self.agents = agents - self.user_profile = user_profile # {"name": "...", "background": "..."} - self.history: List[Dict[str, str]] = [] - self.turn_count = 0 - self._last_orchestrator_turn = -1 - self._forced_next_agent: Optional[Agent] = None # Locked next agent from @mention or user direction - self._hitl_triggered = False # Track if early HITL was already triggered for current turn - - # Build global introduction including user if provided - agent_names = ', '.join([a.name for a in self.agents]) - user_intro = "" - if self.user_profile: - user_intro = f"\nUser Participant: {self.user_profile['name']} - {self.user_profile.get('background', '')}" - - self.global_intro = ( - f"The room is active. Topic: {self.config.topic}\n" - f"Participating Agents: {agent_names}{user_intro}\n" - f"Note: Agents may respond with 'PASS' if they have nothing meaningful to add." - ) - self.history.append({"role": "system", "content": self.global_intro, "timestamp": _now()}) - self._agent_index = 0 - - def add_user_message(self, username: str, message: str): - """Inject a human message into the session history, and detect @mention agent forcing.""" - self.history.append({"role": username, "content": message, "timestamp": _now()}) - - # Check if the user addressed a specific agent - forced = _find_forced_agent(message, self.agents) - if forced: - self._forced_next_agent = forced - - # Reset HITL trigger on user input - self._hitl_triggered = False - - def get_agent_context(self, current_agent: Agent) -> List[Dict[str, str]]: - """Format history into an LLM context including system prompt.""" - context = [] - for msg in self.history: - role = "user" - if msg["role"] == current_agent.name: - role = "assistant" - elif msg["role"] == "system": - role = "system" - - content = msg["content"] - if msg["role"] not in (current_agent.name, "system"): - content = f"[{msg['role']} said]: {content}" - - context.append({"role": role, "content": content}) - return context - - def generate_next_turn(self) -> Optional[Dict[str, str]]: - """Determine next agent, get response, and log it.""" - if self.turn_count >= self.config.max_turns: - return None - - # Optional Orchestrator Intervention (every 3 turns, once per interval) - if ( - self.config.orchestrator - and self.turn_count > 0 - and self.turn_count % 3 == 0 - and self._last_orchestrator_turn != self.turn_count - ): - self._last_orchestrator_turn = self.turn_count - orch_agent = Agent(self.config.orchestrator) - context = self.get_agent_context(orch_agent) - msg = orch_agent.generate_response(context) - self.turn_count += 1 - if "PASS" not in msg: - turn_data = { - "role": f"Orchestrator ({orch_agent.name})", - "content": msg, - "color": orch_agent.config.color, - "timestamp": _now() - } - self.history.append(turn_data) - return turn_data - # Orchestrator passed — fall through to normal turn - - # Resolve agent: forced @mention > smart selection - if self._forced_next_agent: - agent = self._forced_next_agent - self._forced_next_agent = None - else: - agent = self._select_next_agent() - - context = self.get_agent_context(agent) - response_text = agent.generate_response(context) - - # Handle PASS: agent has nothing to add — silently skip turn - if response_text.strip().upper() == "PASS": - self.turn_count += 1 - return {"role": agent.name, "content": "PASS", "color": agent.config.color, "timestamp": _now(), "skipped": True} - - turn_data = { - "role": agent.name, - "content": response_text, - "color": agent.config.color, - "timestamp": _now() - } - self.history.append(turn_data) - self.turn_count += 1 - return turn_data - - def _select_next_agent(self) -> Agent: - """Select who talks next based on SessionType and expertise scoring.""" - if self.config.session_type == SessionType.ROUND_ROBIN: - agent = self.agents[self._agent_index] - self._agent_index = (self._agent_index + 1) % len(self.agents) - return agent - - elif self.config.session_type == SessionType.ARGUMENTATIVE: - if len(self.agents) >= 2: - return self.agents[self.turn_count % 2] - return self.agents[0] - - elif self.config.session_type == SessionType.DYNAMIC: - # Build context text from recent history for scoring - recent = " ".join(m["content"] for m in self.history[-5:]) - - # 1. Check for @mention or name reference in last user/agent message - if self.history: - last_content = self.history[-1].get("content", "") - forced = _find_forced_agent(last_content, self.agents) - if forced: - return forced - - # 2. Score all agents by expertise relevance to recent context - scored = [(agent, _score_agent_expertise(agent, recent)) for agent in self.agents] - scored.sort(key=lambda x: x[1], reverse=True) - - # If top scorer has meaningful score AND is NOT the agent that just spoke, pick them - if scored and scored[0][1] > 0: - last_speaker = self.history[-1].get("role", "") if self.history else "" - for candidate, score in scored: - if candidate.name != last_speaker: - return candidate - - # 3. Fallback to round-robin - agent = self.agents[self._agent_index] - self._agent_index = (self._agent_index + 1) % len(self.agents) - return agent - - return self.agents[0] - - def needs_human_input(self) -> bool: - """Check if it's time for human input, including early trigger if user is directly addressed.""" - if self.config.human_in_the_loop_turns <= 0: - return False - - # Early HITL: if the last agent's message addressed the user by name - if not self._hitl_triggered and len(self.history) > 1: - last = self.history[-1] - if last.get("role") not in ("system",) and _user_is_addressed(last.get("content", ""), self.user_profile): - self._hitl_triggered = True - return True - - return self.turn_count > 0 and self.turn_count % self.config.human_in_the_loop_turns == 0