Spaces:
Sleeping
Sleeping
File size: 3,263 Bytes
e661e91 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | """Pluggable conversation structures and decision-making methods.
The orchestrator drives one ConversationStructure (how the discussion
is organized) followed by one DecisionMethod (how a verdict is
reached). The two are decoupled by the `DecisionInput` contract in
`types.py`: every structure produces one, every method consumes one,
so in principle any combination is valid.
Adding a new structure or decision method means:
1. Subclass `ConversationStructure` or `DecisionMethod` in
`structures/` or `decisions/`.
2. Register the class in the appropriate registry below.
Built-in choices:
Structures:
- "collaborative" → CollaborativeDiscussion (default)
- "roberts_rules" → RobertsRulesDiscussion
Decision methods:
- "consensus" → ConsensusDecision (default)
- "majority" → MajorityRulesDecision
- "ranked_choice" → RankedChoiceDecision
- "roberts_rules_vote" → RobertsRulesVote
"""
from __future__ import annotations
from typing import Type
from app.services.conversation.decisions.base import DecisionMethod
from app.services.conversation.decisions.consensus import ConsensusDecision
from app.services.conversation.decisions.majority import MajorityRulesDecision
from app.services.conversation.decisions.ranked_choice import RankedChoiceDecision
from app.services.conversation.decisions.roberts_rules_vote import RobertsRulesVote
from app.services.conversation.structures.base import ConversationStructure
from app.services.conversation.structures.collaborative import CollaborativeDiscussion
from app.services.conversation.structures.roberts_rules import RobertsRulesDiscussion
from app.services.conversation.types import DecisionInput # noqa: F401 re-export
DEFAULT_STRUCTURE_ID = "collaborative"
DEFAULT_DECISION_ID = "consensus"
STRUCTURE_REGISTRY: dict[str, Type[ConversationStructure]] = {
"collaborative": CollaborativeDiscussion,
"roberts_rules": RobertsRulesDiscussion,
}
DECISION_REGISTRY: dict[str, Type[DecisionMethod]] = {
"consensus": ConsensusDecision,
"majority": MajorityRulesDecision,
"ranked_choice": RankedChoiceDecision,
"roberts_rules_vote": RobertsRulesVote,
}
def get_structure(structure_id: str | None) -> Type[ConversationStructure]:
"""Resolve a structure id to its class, defaulting on miss."""
if not structure_id or structure_id not in STRUCTURE_REGISTRY:
return STRUCTURE_REGISTRY[DEFAULT_STRUCTURE_ID]
return STRUCTURE_REGISTRY[structure_id]
def get_decision(decision_id: str | None) -> Type[DecisionMethod]:
"""Resolve a decision-method id to its class, defaulting on miss."""
if not decision_id or decision_id not in DECISION_REGISTRY:
return DECISION_REGISTRY[DEFAULT_DECISION_ID]
return DECISION_REGISTRY[decision_id]
def list_structures() -> list[dict[str, str]]:
"""Catalog data for the frontend picker."""
return [
{"id": sid, "name": cls.NAME, "description": cls.DESCRIPTION}
for sid, cls in STRUCTURE_REGISTRY.items()
]
def list_decisions() -> list[dict[str, str]]:
"""Catalog data for the frontend picker."""
return [
{"id": did, "name": cls.NAME, "description": cls.DESCRIPTION}
for did, cls in DECISION_REGISTRY.items()
]
|