feat: add DiscussionEngine with event-driven architecture
- engine.py: DiscussionEngine class with run() and inject_message() - All print() calls replaced with EventEmitter.emit() - Event mapping: DISCUSSION_START, ROUND_START, AGENT_TURN_START, AGENT_MESSAGE, TOOL_CALL, TOOL_RESULT, AGENT_ERROR, ROUND_END, DISCUSSION_END, BOSS_TURN - Accepts typed dependencies (APIClient, ToolRegistry, EventEmitter) - 27 engine tests, all 168 total tests passing Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
337
meeting_room/engine.py
Normal file
337
meeting_room/engine.py
Normal file
@@ -0,0 +1,337 @@
|
||||
"""DiscussionEngine — event-driven round-robin discussion runner.
|
||||
|
||||
Replaces the monolithic backends/custom/run.py with a clean, testable
|
||||
architecture that emits typed events instead of printing to stdout.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from typing import Union
|
||||
|
||||
from meeting_room.api_client import APIClient
|
||||
from meeting_room.events import (
|
||||
AGENT_ERROR,
|
||||
AGENT_MESSAGE,
|
||||
AGENT_TURN_START,
|
||||
BOSS_TURN,
|
||||
DISCUSSION_END,
|
||||
DISCUSSION_START,
|
||||
ROUND_END,
|
||||
ROUND_START,
|
||||
TOOL_CALL,
|
||||
TOOL_RESULT,
|
||||
EventEmitter,
|
||||
)
|
||||
from meeting_room.models import (
|
||||
DiscussionConfig,
|
||||
Message,
|
||||
RoleConfig,
|
||||
ScenarioConfig,
|
||||
ToolCallRecord,
|
||||
)
|
||||
from meeting_room.tools import ToolRegistry
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
MAX_TOOL_ITERATIONS = 5
|
||||
|
||||
|
||||
class DiscussionEngine:
|
||||
"""Orchestrates a multi-agent discussion using events for output.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
config : DiscussionConfig
|
||||
Top-level configuration (providers, roles, defaults).
|
||||
api_client : APIClient
|
||||
Client for making LLM API calls.
|
||||
tool_registry : ToolRegistry
|
||||
Registry for tool schema lookup and execution.
|
||||
events : EventEmitter
|
||||
Event bus for emitting lifecycle events.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
config: DiscussionConfig,
|
||||
api_client: APIClient,
|
||||
tool_registry: ToolRegistry,
|
||||
events: EventEmitter,
|
||||
) -> None:
|
||||
self.config = config
|
||||
self.api = api_client
|
||||
self.tools = tool_registry
|
||||
self.events = events
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public API
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def run(
|
||||
self,
|
||||
scenario: Union[ScenarioConfig, dict],
|
||||
) -> list[Message]:
|
||||
"""Run a full discussion and return the conversation as Message objects.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
scenario : ScenarioConfig or dict
|
||||
The scenario to run. A dict is accepted for backward
|
||||
compatibility and is coerced to ScenarioConfig.
|
||||
|
||||
Returns
|
||||
-------
|
||||
list[Message]
|
||||
The ordered list of messages produced during the discussion.
|
||||
"""
|
||||
if isinstance(scenario, dict):
|
||||
scenario = ScenarioConfig(**scenario)
|
||||
|
||||
participants = scenario.participants or list(self.config.roles.keys())
|
||||
max_rounds = scenario.max_rounds or self.config.defaults.max_rounds
|
||||
|
||||
# Eagerly resolve role configs so missing role_id fails fast.
|
||||
roles: dict[str, RoleConfig] = {}
|
||||
for rid in participants:
|
||||
if rid not in self.config.roles:
|
||||
raise ValueError(f"Unknown role_id '{rid}' — not in config.roles")
|
||||
roles[rid] = self.config.roles[rid]
|
||||
|
||||
conversation: list[Message] = []
|
||||
|
||||
self.events.emit(
|
||||
DISCUSSION_START,
|
||||
name=scenario.name,
|
||||
participants=[roles[rid].name for rid in participants],
|
||||
rounds=max_rounds,
|
||||
problem=scenario.problem,
|
||||
)
|
||||
|
||||
for round_num in range(1, max_rounds + 1):
|
||||
self.events.emit(ROUND_START, round_num=round_num, total_rounds=max_rounds)
|
||||
|
||||
for role_id in participants:
|
||||
role = roles[role_id]
|
||||
is_first = len(conversation) == 0
|
||||
|
||||
self.events.emit(
|
||||
AGENT_TURN_START,
|
||||
role_id=role_id,
|
||||
name=role.name,
|
||||
)
|
||||
|
||||
try:
|
||||
content = self._run_agent_turn(
|
||||
role=role,
|
||||
role_id=role_id,
|
||||
problem=scenario.problem,
|
||||
conversation=conversation,
|
||||
is_first_message=is_first,
|
||||
)
|
||||
except Exception as exc:
|
||||
content = f"[Error: {exc}]"
|
||||
self.events.emit(
|
||||
AGENT_ERROR,
|
||||
role_id=role_id,
|
||||
name=role.name,
|
||||
error=str(exc),
|
||||
)
|
||||
|
||||
conversation.append(
|
||||
Message(role_id=role_id, name=role.name, content=content)
|
||||
)
|
||||
self.events.emit(
|
||||
AGENT_MESSAGE,
|
||||
role_id=role_id,
|
||||
name=role.name,
|
||||
content=content,
|
||||
)
|
||||
|
||||
self.events.emit(ROUND_END, round_num=round_num, total_rounds=max_rounds)
|
||||
|
||||
self.events.emit(
|
||||
DISCUSSION_END,
|
||||
total_messages=len(conversation),
|
||||
)
|
||||
|
||||
return conversation
|
||||
|
||||
def inject_message(self, role_id: str, content: str) -> Message:
|
||||
"""Inject an arbitrary message from a Boss/CSO role.
|
||||
|
||||
This does **not** make an LLM call — it simply appends a message
|
||||
and emits a BOSS_TURN event so subscribers can log or relay it.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
role_id : str
|
||||
Identifier of the injecting role (e.g. "boss", "cso").
|
||||
content : str
|
||||
The message content to inject.
|
||||
|
||||
Returns
|
||||
-------
|
||||
Message
|
||||
The created Message object.
|
||||
|
||||
Raises
|
||||
------
|
||||
ValueError
|
||||
If *role_id* is not in config.roles.
|
||||
"""
|
||||
if role_id not in self.config.roles:
|
||||
raise ValueError(f"Unknown role_id '{role_id}' — not in config.roles")
|
||||
role = self.config.roles[role_id]
|
||||
|
||||
msg = Message(role_id=role_id, name=role.name, content=content)
|
||||
|
||||
self.events.emit(
|
||||
BOSS_TURN,
|
||||
role_id=role_id,
|
||||
name=role.name,
|
||||
content=content,
|
||||
)
|
||||
return msg
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Internal helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@staticmethod
|
||||
def _format_conversation(conversation: list[Message]) -> str:
|
||||
"""Format a list of Message objects into a prompt-friendly string."""
|
||||
return "\n\n".join(
|
||||
f"[{msg.name}]: {msg.content}" for msg in conversation
|
||||
)
|
||||
|
||||
def _build_messages(
|
||||
self,
|
||||
role: RoleConfig,
|
||||
problem: str,
|
||||
conversation: list[Message],
|
||||
is_first_message: bool,
|
||||
) -> list[dict]:
|
||||
"""Build the API message list for one agent turn."""
|
||||
system_msg = {"role": "system", "content": role.system_prompt.strip()}
|
||||
|
||||
if is_first_message:
|
||||
context_msg = {
|
||||
"role": "user",
|
||||
"content": (
|
||||
f"Problem for discussion:\n{problem}\n\n"
|
||||
"Discuss it with other participants. Express your position. "
|
||||
"Use tools if necessary."
|
||||
),
|
||||
}
|
||||
else:
|
||||
context_msg = {
|
||||
"role": "user",
|
||||
"content": (
|
||||
f"Here is what has been discussed so far:\n\n"
|
||||
f"{self._format_conversation(conversation)}\n\n"
|
||||
"Continue the discussion. Respond to arguments, offer your own. "
|
||||
"Use tools if necessary."
|
||||
),
|
||||
}
|
||||
|
||||
return [system_msg, context_msg]
|
||||
|
||||
def _run_agent_turn(
|
||||
self,
|
||||
role: RoleConfig,
|
||||
role_id: str,
|
||||
problem: str,
|
||||
conversation: list[Message],
|
||||
is_first_message: bool,
|
||||
) -> str:
|
||||
"""Run one agent turn with tool support.
|
||||
|
||||
Returns the final text content produced by the agent.
|
||||
"""
|
||||
messages = self._build_messages(role, problem, conversation, is_first_message)
|
||||
tool_schemas = self.tools.get_tool_schemas(role.tools)
|
||||
has_tools = len(tool_schemas) > 0
|
||||
|
||||
for _ in range(MAX_TOOL_ITERATIONS):
|
||||
if has_tools:
|
||||
result = self.api.chat(
|
||||
provider=role.provider,
|
||||
model=role.model,
|
||||
messages=messages,
|
||||
temperature=role.temperature,
|
||||
tools=tool_schemas,
|
||||
)
|
||||
else:
|
||||
content = self.api.chat_stream(
|
||||
provider=role.provider,
|
||||
model=role.model,
|
||||
messages=messages,
|
||||
temperature=role.temperature,
|
||||
)
|
||||
return content
|
||||
|
||||
if result["tool_calls"]:
|
||||
# Append assistant message with tool calls
|
||||
assistant_msg: dict = {
|
||||
"role": "assistant",
|
||||
"content": result["content"] or None,
|
||||
"tool_calls": [
|
||||
{
|
||||
"id": tc["id"],
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": tc["name"],
|
||||
"arguments": tc["arguments"],
|
||||
},
|
||||
}
|
||||
for tc in result["tool_calls"]
|
||||
],
|
||||
}
|
||||
messages.append(assistant_msg)
|
||||
|
||||
# Execute each tool call and append results
|
||||
for tc in result["tool_calls"]:
|
||||
self.events.emit(
|
||||
TOOL_CALL,
|
||||
role_id=role_id,
|
||||
tool_name=tc["name"],
|
||||
arguments=tc["arguments"],
|
||||
)
|
||||
|
||||
try:
|
||||
args = (
|
||||
json.loads(tc["arguments"])
|
||||
if isinstance(tc["arguments"], str)
|
||||
else tc["arguments"]
|
||||
)
|
||||
except json.JSONDecodeError:
|
||||
args = {}
|
||||
|
||||
tool_result = self.tools.execute_tool(tc["name"], args)
|
||||
|
||||
preview = tool_result[:200].replace("\n", " ")
|
||||
self.events.emit(
|
||||
TOOL_RESULT,
|
||||
role_id=role_id,
|
||||
tool_name=tc["name"],
|
||||
result_preview=preview,
|
||||
full_result=tool_result,
|
||||
)
|
||||
|
||||
messages.append(
|
||||
{
|
||||
"role": "tool",
|
||||
"tool_call_id": tc["id"],
|
||||
"content": tool_result,
|
||||
}
|
||||
)
|
||||
continue
|
||||
|
||||
# No tool calls — return the content
|
||||
if result["content"]:
|
||||
return result["content"]
|
||||
|
||||
# Exhausted tool iterations — return whatever we have
|
||||
return result.get("content", "")
|
||||
Reference in New Issue
Block a user