NeonClary
perf(orchestrator): parallel phases, streaming, early table summaries
83add3c
Raw
History Blame Contribute Delete
18.4 kB
"""Voting helpers shared by Majority, Ranked-Choice, and Robert's
Rules decision methods.
Three pieces live here:
* `extract_candidate_options(...)`: takes a list of finalized
position texts, asks the orchestrator LLM to cluster them into a
small set of distinct option labels, returns them as a list of
short strings.
* `cast_vote(...)`: asks one participant to vote on a list of
options (or to rank them, depending on `mode`), returning a
structured dict.
* `run_irv(...)`: instant-runoff tally for ranked-choice ballots.
All three are pure helpers (no Session mutation) so the decision
method classes can compose them however they want.
"""
from __future__ import annotations
import asyncio
import json
import logging
import time
from typing import Any, Awaitable, Callable
from app.clients.llm_router import chat_completion
from app.services.json_calls import (
orchestrator_call,
parse_json_response,
)
from app.services.models import Participant, Session
from app.services.prompts import NO_REASONING_DIRECTIVE
from app.utils.sanitize import strip_thinking
LOG = logging.getLogger(__name__)
# How many distinct options the candidate-extraction LLM call may
# return. Cap is intentional: ranked-choice with > 6 options gets
# unwieldy for both LLM voters and human readers of the report.
MAX_CANDIDATES = 6
CANDIDATE_EXTRACTION_PROMPT = """The following are participants' final \
positions on this question:
Question: {question}
Participant positions:
{positions_block}
Cluster these into between 2 and {max_candidates} distinct option \
labels that capture the main answers being supported. Each label \
should be short (one sentence) and self-contained — a voter who \
hadn't read the transcript should still understand what they're \
voting for.
Return ONLY this JSON shape (no prose, no fences):
{{
"options": ["short option 1", "short option 2", ...]
}}
"""
VOTE_SINGLE_PROMPT = """You are {participant_name}.
The group has been discussing this question:
{question}
After deliberation, the choices on the table are:
{options_block}
Cast your vote for the ONE option you support. Return ONLY this JSON \
(no prose, no fences):
{{
"choice": <integer 1..N of your top choice>,
"reason": "one sentence on why"
}}
"""
VOTE_RANK_PROMPT = """You are {participant_name}.
The group has been discussing this question:
{question}
After deliberation, the choices on the table are:
{options_block}
Rank ALL of the options from most preferred (rank 1) to least \
preferred. You must include every option exactly once. Return ONLY \
this JSON (no prose, no fences):
{{
"ranking": [<rank-1 option number>, <rank-2 option number>, ...],
"reason": "one sentence on your top choice"
}}
"""
VOTE_YESNO_PROMPT = """You are {participant_name}.
The chair has put the following motion before the assembly:
Motion: {motion}
Cast your vote. Return ONLY this JSON (no prose, no fences):
{{
"vote": "aye" | "nay" | "abstain",
"reason": "one sentence on why"
}}
"""
def _format_positions_block(positions: dict[str, str], participants: list[Participant]) -> str:
by_id = {p.participant_id: p for p in participants}
lines: list[str] = []
for pid, text in positions.items():
name = by_id[pid].name if pid in by_id else pid
lines.append(f"- {name}: {text}".strip())
return "\n".join(lines) if lines else "(no positions recorded)"
def _format_options_block(options: list[str]) -> str:
return "\n".join(f" {i + 1}. {opt}" for i, opt in enumerate(options))
async def extract_candidate_options(
*,
session: Session,
question: str,
positions: dict[str, str],
participants: list[Participant],
max_candidates: int = MAX_CANDIDATES,
) -> list[str]:
"""Cluster finalized positions into a short list of distinct
option labels for a vote.
Falls back to using each participant's position verbatim (up to
`max_candidates`) if the LLM call fails or returns nothing
parseable.
"""
if not positions:
return []
positions_block = _format_positions_block(positions, participants)
prompt = CANDIDATE_EXTRACTION_PROMPT.format(
question=question,
positions_block=positions_block,
max_candidates=max_candidates,
)
from app.services.orchestrator import _orchestrator_model_id, _bump_orchestrator_count
raw, parsed = await orchestrator_call(
orchestrator_model_id=_orchestrator_model_id(session),
user_prompt=prompt,
label="vote_candidate_extraction",
api_log=session.api_log,
expect_json=True,
max_tokens=500,
temperature=0.2,
)
_bump_orchestrator_count(session)
options: list[str] = []
if isinstance(parsed, dict):
raw_opts = parsed.get("options") or []
if isinstance(raw_opts, list):
options = [str(o).strip() for o in raw_opts if str(o).strip()]
# Truncate to cap.
options = options[:max_candidates]
if not options:
# Fallback: take each unique position verbatim, truncated.
seen: set[str] = set()
for txt in positions.values():
cleaned = (txt or "").strip()
if not cleaned:
continue
key = cleaned[:80].lower()
if key in seen:
continue
seen.add(key)
# Single-line short version
single = " ".join(cleaned.split())
if len(single) > 160:
single = single[:157] + "..."
options.append(single)
if len(options) >= max_candidates:
break
return options
async def gather_votes_parallel(
voters: list[Participant],
cast_fn: Callable[..., Awaitable[dict[str, Any]]],
*,
session: Session,
default_mode: str = "single",
**cast_kwargs: Any,
) -> list[tuple[Participant, dict[str, Any]]]:
"""Run ballot calls concurrently; roster order is preserved."""
if not voters:
return []
async def _one(p: Participant) -> tuple[Participant, dict[str, Any]]:
result = await cast_fn(session=session, participant=p, **cast_kwargs)
return p, result
gathered = await asyncio.gather(
*[_one(p) for p in voters],
return_exceptions=True,
)
out: list[tuple[Participant, dict[str, Any]]] = []
for p, item in zip(voters, gathered):
if isinstance(item, BaseException):
LOG.exception("Parallel vote failed for %s: %s", p.participant_id, item)
out.append((p, _vote_default(default_mode)))
else:
out.append(item)
return out
async def cast_vote_single(
*,
session: Session,
participant: Participant,
question: str,
options: list[str],
) -> dict[str, Any]:
"""Ask one participant to pick exactly one option.
Returns {"choice": int (1-based, 0 if invalid), "reason": str,
"ok": bool}. Always non-fatal — a malformed reply just yields
{"choice": 0, ...} so the tally can skip it.
"""
return await _cast_vote(
session=session,
participant=participant,
prompt=VOTE_SINGLE_PROMPT.format(
participant_name=participant.name,
question=question,
options_block=_format_options_block(options),
),
mode="single",
n_options=len(options),
)
async def cast_vote_ranking(
*,
session: Session,
participant: Participant,
question: str,
options: list[str],
) -> dict[str, Any]:
"""Ask one participant to fully rank all options.
Returns {"ranking": [int, ...] (1-based, may be partial if the
model misbehaved), "reason": str, "ok": bool}.
"""
return await _cast_vote(
session=session,
participant=participant,
prompt=VOTE_RANK_PROMPT.format(
participant_name=participant.name,
question=question,
options_block=_format_options_block(options),
),
mode="rank",
n_options=len(options),
)
async def cast_vote_yesno(
*,
session: Session,
participant: Participant,
motion: str,
) -> dict[str, Any]:
"""Ask one participant to vote aye/nay/abstain on a motion.
Returns {"vote": "aye"|"nay"|"abstain"|"", "reason": str,
"ok": bool}.
"""
return await _cast_vote(
session=session,
participant=participant,
prompt=VOTE_YESNO_PROMPT.format(
participant_name=participant.name,
motion=motion,
),
mode="yesno",
n_options=0,
)
async def _cast_vote(
*,
session: Session,
participant: Participant,
prompt: str,
mode: str,
n_options: int,
) -> dict[str, Any]:
if participant.kind == "human":
# For now, treat human participants as abstaining in the
# automated vote path. Future work: pause the orchestrator
# and route through human_io so the user can cast a real
# ballot. The decision method can detect this and surface a
# note in the report.
if mode == "yesno":
return {"vote": "abstain", "reason": "(human participant)", "ok": False}
if mode == "rank":
return {"ranking": [], "reason": "(human participant)", "ok": False}
return {"choice": 0, "reason": "(human participant)", "ok": False}
system_text = (
f"{participant.role_prompt}\n\n{NO_REASONING_DIRECTIVE}\n\n"
"When asked to cast a vote, reply with ONLY the requested JSON "
"object and no other text."
)
messages = [
{"role": "system", "content": system_text},
{"role": "user", "content": prompt},
]
resolved = {
"model_id": participant.model_id,
"base_url": participant.base_url,
"api_key": participant.api_key,
"is_neon": participant.is_neon,
"hana_model_id": participant.hana_model_id,
"persona_name": participant.persona_name,
"neon_direct_vllm": participant.neon_direct_vllm,
"vllm_base_url": participant.vllm_base_url,
"vllm_api_key": participant.vllm_api_key,
}
log_entry: dict[str, Any] = {
"timestamp": time.time(),
"label": f"vote:{mode}:{participant.participant_id}",
"model": participant.model_id,
"request": {"messages": messages, "max_tokens": 300},
}
try:
result = await chat_completion(
resolved=resolved,
messages=messages,
max_tokens=300,
temperature=0.2,
timeout=45.0,
)
except Exception as exc: # noqa: BLE001
LOG.warning("vote %s for %s failed: %s", mode, participant.participant_id, exc)
log_entry["response"] = {"error": str(exc)}
session.api_log.append(log_entry)
return _vote_default(mode)
log_entry["response"] = result
session.api_log.append(log_entry)
if result.get("error"):
return _vote_default(mode)
raw = strip_thinking(result.get("response", ""))
return _parse_vote(raw, mode=mode, n_options=n_options)
def _vote_default(mode: str) -> dict[str, Any]:
if mode == "yesno":
return {"vote": "", "reason": "", "ok": False}
if mode == "rank":
return {"ranking": [], "reason": "", "ok": False}
return {"choice": 0, "reason": "", "ok": False}
def _parse_vote(raw: str, *, mode: str, n_options: int) -> dict[str, Any]:
parsed = parse_json_response(raw)
if not isinstance(parsed, dict):
return _vote_default(mode)
reason = str(parsed.get("reason") or "").strip()
if mode == "yesno":
vote = str(parsed.get("vote") or "").strip().lower()
if vote not in ("aye", "nay", "abstain"):
return {"vote": "", "reason": reason, "ok": False}
return {"vote": vote, "reason": reason, "ok": True}
if mode == "rank":
raw_rank = parsed.get("ranking") or parsed.get("rank") or []
if not isinstance(raw_rank, list):
return {"ranking": [], "reason": reason, "ok": False}
ranking: list[int] = []
seen: set[int] = set()
for item in raw_rank:
try:
idx = int(item)
except (TypeError, ValueError):
continue
if 1 <= idx <= n_options and idx not in seen:
seen.add(idx)
ranking.append(idx)
ok = len(ranking) == n_options
return {"ranking": ranking, "reason": reason, "ok": ok}
# single-choice
try:
choice = int(parsed.get("choice") or 0)
except (TypeError, ValueError):
choice = 0
if not (1 <= choice <= n_options):
choice = 0
return {"choice": choice, "reason": reason, "ok": choice > 0}
# ---------------------------------------------------------------------------
# Tallying
# ---------------------------------------------------------------------------
def tally_single_votes(
ballots: list[dict[str, Any]],
n_options: int,
) -> dict[str, Any]:
"""Tally one-shot plurality votes.
`ballots` items shape: {"choice": int 1..N or 0, ...}.
Returns:
{
"counts": [vote_count_for_option_1, ..._for_option_N],
"winner": int (1-based; 0 if no votes),
"tied_for_first": [int, ...],
"total_cast": int,
"abstentions": int,
}
"""
counts = [0] * n_options
cast = 0
abst = 0
for b in ballots:
choice = b.get("choice", 0)
if isinstance(choice, int) and 1 <= choice <= n_options:
counts[choice - 1] += 1
cast += 1
else:
abst += 1
if cast == 0:
return {
"counts": counts, "winner": 0, "tied_for_first": [],
"total_cast": 0, "abstentions": abst,
}
top = max(counts)
leaders = [i + 1 for i, c in enumerate(counts) if c == top]
return {
"counts": counts,
"winner": leaders[0] if len(leaders) == 1 else 0,
"tied_for_first": leaders if len(leaders) > 1 else [],
"total_cast": cast,
"abstentions": abst,
}
def tally_yesno_votes(ballots: list[dict[str, Any]]) -> dict[str, Any]:
"""Tally aye/nay/abstain motion votes.
Returns: {"aye": int, "nay": int, "abstain": int,
"passes": bool, "majority": "aye"|"nay"|"tie",
"ratio_aye": float}.
"""
aye = sum(1 for b in ballots if b.get("vote") == "aye")
nay = sum(1 for b in ballots if b.get("vote") == "nay")
abst = sum(
1 for b in ballots
if b.get("vote") == "abstain" or not b.get("vote")
)
cast = aye + nay
if cast == 0:
return {
"aye": aye, "nay": nay, "abstain": abst,
"passes": False, "majority": "tie", "ratio_aye": 0.0,
}
if aye > nay:
majority = "aye"
elif nay > aye:
majority = "nay"
else:
majority = "tie"
return {
"aye": aye, "nay": nay, "abstain": abst,
"passes": aye > nay,
"majority": majority,
"ratio_aye": aye / cast,
}
def run_irv(
ballots: list[list[int]],
n_options: int,
) -> dict[str, Any]:
"""Instant-runoff tally on 1-based ranking ballots.
A ballot is a list of 1-based option indices in order of
preference; partial ballots are tolerated. Eliminate the
lowest-first-choice option each round, redistribute its top-rank
votes to the next still-eligible choice on each ballot, until one
option has >50% or all but one are eliminated.
Returns:
{
"rounds": [ {round: int, counts: {option_idx: count},
eliminated: int or None,
winner: int or None}, ... ],
"winner": int (1-based; 0 if no ballots),
"tied": bool,
}
"""
if not ballots or n_options <= 0:
return {"rounds": [], "winner": 0, "tied": False}
eligible: set[int] = set(range(1, n_options + 1))
# Per-ballot cursor (skips eliminated options on the fly)
rounds: list[dict[str, Any]] = []
round_n = 0
while True:
round_n += 1
counts: dict[int, int] = {opt: 0 for opt in eligible}
for ballot in ballots:
top: int | None = None
for opt in ballot:
if opt in eligible:
top = opt
break
if top is not None:
counts[top] = counts.get(top, 0) + 1
total = sum(counts.values())
if total == 0:
rounds.append({"round": round_n, "counts": counts,
"eliminated": None, "winner": None})
return {"rounds": rounds, "winner": 0, "tied": False}
# Check for majority winner
for opt, c in counts.items():
if c * 2 > total:
rounds.append({"round": round_n, "counts": counts,
"eliminated": None, "winner": opt})
return {"rounds": rounds, "winner": opt, "tied": False}
# Only one option left → it wins by default
if len(eligible) == 1:
sole = next(iter(eligible))
rounds.append({"round": round_n, "counts": counts,
"eliminated": None, "winner": sole})
return {"rounds": rounds, "winner": sole, "tied": False}
# Eliminate the option with the fewest first-rank votes; on a
# tie at the bottom, eliminate the one with the lowest 1-based
# index (deterministic tiebreak).
lowest = min(counts.values())
candidates_to_drop = sorted(
opt for opt, c in counts.items() if c == lowest
)
# If ALL remaining options tie at the bottom, we have a
# degenerate tie — no winner.
if len(candidates_to_drop) == len(eligible):
rounds.append({"round": round_n, "counts": counts,
"eliminated": None, "winner": None})
return {"rounds": rounds, "winner": 0, "tied": True}
drop = candidates_to_drop[0]
eligible.discard(drop)
rounds.append({"round": round_n, "counts": counts,
"eliminated": drop, "winner": None})
# Safety: cap iterations at n_options + 1.
if round_n > n_options + 1: # pragma: no cover
return {"rounds": rounds, "winner": 0, "tied": True}