vh/AIPA
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added streaming

Browse Source
This commit is contained in:
Vuong Hoang
2026-04-02 22:47:30 -07:00
parent f8fb8aebdb
commit a282d853a6
6 changed files with 400 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
orchestration/__pycache__/config.cpython-311.pyc
BIN
View File
Binary file not shown.
orchestration/__pycache__/orchestrator.cpython-311.pyc
BIN
View File
Binary file not shown.
orchestration/__pycache__/task_store.cpython-311.pyc
BIN
View File
Binary file not shown.
orchestration/__pycache__/ui.cpython-311.pyc
BIN
View File
Binary file not shown.
orchestration/orchestrator.py
+89 -30
View File
@@ -44,6 +44,13 @@ class ProviderClient:
temperature: float, max_tokens: int) -> str:
return self.call(system, messages, model, temperature, max_tokens)
async def call_async_streaming(self, system: str, messages: list[dict], model: str,
temperature: float, max_tokens: int):
"""Async generator yielding ('thinking'|'content', text) chunks.
Default: emits the full response as a single 'content' chunk."""
response = await self.call_async(system, messages, model, temperature, max_tokens)
yield ('content', response)
# --- OpenAI and any OpenAI-compatible endpoint ---
# Covers AIPA_PROVIDER / {AGENT}_PROVIDER = openai or openai_compatible.
@@ -71,6 +78,29 @@ class OpenAIClient(ProviderClient):
self.call_async(system, messages, model, temperature, max_tokens)
)
async def call_async_streaming(self, system, messages, model, temperature, max_tokens):
"""Stream via the OpenAI-compatible API, yielding thinking and content chunks.
Thinking tokens arrive in delta.reasoning_content (Qwen3, DeepSeek R1 style);
regular tokens arrive in delta.content."""
full_messages = [{"role": "system", "content": system}] + messages
stream = await self.client.chat.completions.create(
model=model,
temperature=temperature,
max_tokens=max_tokens,
messages=full_messages,
extra_body=self.extra_body or None,
stream=True,
)
async for chunk in stream:
if not chunk.choices:
continue
delta = chunk.choices[0].delta
thinking = getattr(delta, 'reasoning_content', None)
if thinking:
yield ('thinking', thinking)
if delta.content:
yield ('content', delta.content)
# --- Anthropic ---
# Uncomment when ready to use. Install: pip install anthropic
@@ -283,6 +313,8 @@ async def call_agent_async(
agent: AgentState,
user_message: str,
dry_run: bool = False,
stream: bool = True,
stream_mode: str = "deliverable",
) -> str:
if dry_run:
return f"[DRY RUN] Would call {agent.name} ({config.agent_provider_name(agent.name)}) with: {user_message[:100]}..."
@@ -290,19 +322,32 @@ async def call_agent_async(
agent.add_user(user_message)
try:
response = await agent.client.call_async(
system=agent.system_prompt,
messages=agent.messages_for_call(),
model=agent.model,
temperature=agent.temperature,
max_tokens=agent.max_tokens,
)
if stream:
response = await ui.stream_agent_output(
agent.name,
agent.client.call_async_streaming(
system=agent.system_prompt,
messages=agent.messages_for_call(),
model=agent.model,
temperature=agent.temperature,
max_tokens=agent.max_tokens,
),
mode=stream_mode,
)
else:
response = await agent.client.call_async(
system=agent.system_prompt,
messages=agent.messages_for_call(),
model=agent.model,
temperature=agent.temperature,
max_tokens=agent.max_tokens,
)
except NotImplementedError as e:
raise RuntimeError(str(e)) from e
agent.add_assistant(response)
if config.DEBUG_PRINT_AGENT_OUTPUTS:
if config.DEBUG_PRINT_AGENT_OUTPUTS and not stream:
ui.print_agent_panel(agent.name, response)
return response
@@ -357,7 +402,7 @@ async def dispatch_to_lead(
if session_id:
task_store.update_task(task, session_id)
try:
output = await call_agent_async(lead, task.brief, dry_run=dry_run)
output = await call_agent_async(lead, task.brief, dry_run=dry_run, stream=False)
task.output = output
task.status = "complete"
except Exception as e:
@@ -424,7 +469,7 @@ async def run_auditor(
f"{deliverable}"
)
auditor.history = [] # Auditor is stateless — fresh call each time
return await call_agent_async(auditor, audit_request, dry_run=dry_run)
return await call_agent_async(auditor, audit_request, dry_run=dry_run, stream_mode="audit")
# ---------------------------------------------------------------------------
@@ -446,7 +491,7 @@ async def update_standing_brief(
f"log entry at the top of Section 9. Archive anything older than "
f"{config.SESSION_LOG_RETENTION} sessions."
)
return await call_agent_async(orchestrator, update_request, dry_run=dry_run)
return await call_agent_async(orchestrator, update_request, dry_run=dry_run, stream_mode="background")
# ---------------------------------------------------------------------------
@@ -502,7 +547,7 @@ class Session:
f"Directive: {directive}"
)
cos_response = await call_agent_async(
self.orchestrator, cos_prompt, dry_run=self.dry_run
self.orchestrator, cos_prompt, dry_run=self.dry_run, stream=True
)
# ── Step 2: Parse any task briefs the orchestrator issued ───────────
@@ -532,10 +577,9 @@ class Session:
f"Please synthesize these into a final deliverable for the Principal.\n\n"
f"{lead_outputs_text}"
)
with ui.synthesis_progress():
deliverable = await call_agent_async(
self.orchestrator, synthesis_prompt, dry_run=self.dry_run
)
deliverable = await call_agent_async(
self.orchestrator, synthesis_prompt, dry_run=self.dry_run, stream=True
)
dispatched_leads = sorted({t.assigned_to for t in completed_tasks})
self.session_log.append(
@@ -549,10 +593,9 @@ class Session:
audit_memo = ""
if self.auto_audit:
try:
with ui.audit_progress():
audit_memo = await run_auditor(
self.auditor, deliverable, task_id, dry_run=self.dry_run
)
audit_memo = await run_auditor(
self.auditor, deliverable, task_id, dry_run=self.dry_run
)
self.session_log.append(f"Audit complete for {task_id}.")
except RuntimeError as e:
ui.print_warning(f"Audit skipped — {e}")
@@ -561,10 +604,9 @@ class Session:
async def close(self):
summary = "\n".join(self.session_log)
with ui.brief_update_progress():
updated_brief = await update_standing_brief(
self.orchestrator, summary, dry_run=self.dry_run
)
updated_brief = await update_standing_brief(
self.orchestrator, summary, dry_run=self.dry_run
)
if not self.dry_run:
save_standing_brief(updated_brief)
ui.print_session_footer(self.directive_count)
@@ -645,17 +687,26 @@ async def cmd_history(session: Session, args: str) -> bool:
return True
@command(["/thinking"], "Show the thinking output from the last response")
async def cmd_thinking(session: Session, args: str) -> bool:
last = ui.get_last_thinking()
if not last:
ui.print_system("No thinking output recorded — thinking mode may not be active.")
else:
ui.print_thinking_expansion(last)
return True
@command(["/audit"], "Re-run Vera's audit on the last deliverable")
async def cmd_audit(session: Session, args: str) -> bool:
if not session.last_deliverable:
ui.print_warning("No deliverable yet — issue a directive first.")
return True
ui.print_system("Running audit on last deliverable…")
with ui.audit_progress():
memo = await run_auditor(
session.auditor, session.last_deliverable,
session.last_task_id, dry_run=session.dry_run,
)
memo = await run_auditor(
session.auditor, session.last_deliverable,
session.last_task_id, dry_run=session.dry_run,
)
ui.print_audit_memo(memo, task_id=session.last_task_id)
return True
@@ -731,7 +782,15 @@ async def direct_agent_session(agent_name: str, dry_run: bool = False):
if not user_input or user_input.lower() in ("/quit", "/exit", "/q"):
break
response = await call_agent_async(agent, user_input, dry_run=dry_run)
if user_input.lower() == "/thinking":
last = ui.get_last_thinking()
if not last:
ui.print_system("No thinking output recorded — thinking mode may not be active.")
else:
ui.print_thinking_expansion(last)
continue
response = await call_agent_async(agent, user_input, dry_run=dry_run, stream=True)
ui.print_agent_panel(agent_name, response)
ui.print_session_footer(0)
orchestration/ui.py
+311 -3
View File
@@ -23,8 +23,9 @@ from datetime import datetime
import config as _config
from rich.console import Console
from rich.console import Console, Group
from rich.columns import Columns
from rich.live import Live
from rich.markdown import Markdown
from rich.panel import Panel
from rich.progress import (
@@ -75,6 +76,10 @@ def _build_theme() -> Theme:
console = Console(theme=_build_theme(), highlight=False)
_last_thinking: str = "" # Stored from most recent streamed response
_last_stream_rendered: bool = False # True when stream_agent_output rendered the final panel
# ---------------------------------------------------------------------------
# Agent visual config
# ---------------------------------------------------------------------------
@@ -268,7 +273,13 @@ def print_agent_panel(agent_name: str, content: str, task_id: str = ""):
"""
Render an agent's output in a styled panel.
Content is treated as Markdown.
Skips printing if stream_agent_output already rendered this response.
"""
global _last_stream_rendered
if _last_stream_rendered:
_last_stream_rendered = False
return
color = _agent_color(agent_name)
title_text = _agent_display_title(agent_name)
subtitle = f"[dim]{task_id}[/dim]" if task_id else ""
@@ -286,15 +297,51 @@ def print_agent_panel(agent_name: str, content: str, task_id: str = ""):
def print_deliverable(content: str, task_id: str = ""):
"""Orchestrator's synthesized deliverable."""
print_agent_panel(_config.ORCHESTRATOR_AGENT, content, task_id)
"""
Orchestrator's synthesized deliverable.
Skips printing if stream_agent_output already rendered this response.
"""
global _last_stream_rendered
if _last_stream_rendered:
_last_stream_rendered = False
return
_print_deliverable_panel(content, task_id)
def _print_deliverable_panel(content: str, task_id: str = ""):
"""Internal: always render the deliverable panel unconditionally."""
agent_name = _config.ORCHESTRATOR_AGENT
color = _agent_color(agent_name)
title_text = _agent_display_title(agent_name)
subtitle = f"[dim]{task_id}[/dim]" if task_id else ""
panel = Panel(
Markdown(content),
title=f"[{color}] {title_text} [/{color}]",
subtitle=subtitle,
border_style=color,
padding=(1, 2),
box=box.ROUNDED,
)
console.print()
console.print(panel)
def print_audit_memo(content: str, task_id: str = ""):
"""
Auditor's memo. Verdict line drives the border colour.
PASS = green, PASS WITH NOTES = yellow, FLAG = yellow, REJECT = red.
Skips printing if stream_agent_output already rendered this response.
"""
global _last_stream_rendered
if _last_stream_rendered:
_last_stream_rendered = False
return
_print_audit_memo_panel(content, task_id)
def _print_audit_memo_panel(content: str, task_id: str = ""):
"""Internal: always render the audit memo panel with verdict-based border."""
aud_name = _config.AUDITOR_AGENT
aud_title = _config.agent_title(aud_name)
aud_color = _agent_color(aud_name)
@@ -390,6 +437,267 @@ def brief_update_progress():
progress.add_task("", total=None)
yield
# ---------------------------------------------------------------------------
# Streaming output with thinking support
# ---------------------------------------------------------------------------
def _thinking_panel_title(agent_name: str, mode: str) -> str:
"""Return the thinking panel title, tinted by agent identity."""
if mode == "audit":
color = _agent_color(agent_name)
return f"[{color}]Vera thinking…[/{color}]"
color = _agent_color(agent_name)
return f"[{color} dim]Thinking…[/{color} dim]"
def _build_thinking_renderable(
thinking_text: str,
active: bool,
agent_name: str,
mode: str,
window: int = 300,
) -> object:
"""
Build the thinking panel (active) or collapsed summary line (done).
Active: dim panel with a rolling window of the last `window` chars,
plus an overflow indicator when earlier content has scrolled off.
Collapsed: single dim line — ' ▸ Thinking (N chars, L lines)'.
"""
if not thinking_text:
return None
if active:
overflow = len(thinking_text) - window
if overflow > 0:
preview = thinking_text[-window:]
header = Text(f" +{overflow} chars earlier\n", style="dim italic")
body = Text(preview, style="dim")
content = Group(header, body)
else:
content = Text(thinking_text, style="dim")
return Panel(
content,
title=_thinking_panel_title(agent_name, mode),
border_style="dim",
padding=(0, 1),
box=box.SIMPLE,
)
else:
lines = thinking_text.count("\n") + 1
return Text(
f" ▸ Thinking [{len(thinking_text):,} chars · {lines} lines]",
style="dim",
)
async def stream_agent_output(
agent_name: str,
stream_gen,
mode: str = "deliverable",
) -> str:
"""
Consume an async generator of ('thinking'|'content', text) chunks and
render them live in the terminal.
mode values
-----------
"deliverable" Normal agent response — thinking panel + full styled panel.
The Live display is NOT transient: the final Markdown panel
is rendered inside the Live block, so there is no erase/
reprint flash. Sets _last_stream_rendered = True so that
the caller's print_deliverable / print_agent_panel becomes
a no-op.
"audit" Like "deliverable" but uses Vera's yellow identity cues
during the thinking phase. Sets _last_stream_rendered so
print_audit_memo is a no-op (the verdict-border panel is
rendered here instead).
"background" Housekeeping call (e.g. standing-brief update). A compact
transient progress line is shown; no full panel is printed.
_last_stream_rendered is set so the caller's print_* is a
no-op.
In all modes the full content string is returned and _last_thinking is
updated so /thinking works.
Thinking tokens stream inside a dim panel. When the first content token
arrives the panel collapses to a single dim summary line and content begins
streaming in the agent's normal styled panel.
The full thinking text is stored in _last_thinking and can be retrieved
with get_last_thinking() (exposed to the /thinking command).
Returns the full content string.
"""
global _last_thinking, _last_stream_rendered
color = _agent_color(agent_name)
title_text = _agent_display_title(agent_name)
# ── Background mode: compact transient progress line ────────────────────
if mode == "background":
thinking_parts: list[str] = []
content_parts: list[str] = []
with Progress(
SpinnerColumn(style="dim"),
TextColumn("[dim]Updating standing brief…[/dim]"),
TimeElapsedColumn(),
console=console,
transient=True,
) as prog:
prog.add_task("", total=None)
async for chunk_type, text in stream_gen:
if chunk_type == "thinking":
thinking_parts.append(text)
elif chunk_type == "content":
content_parts.append(text)
_last_thinking = "".join(thinking_parts)
_last_stream_rendered = True
return "".join(content_parts)
# ── Deliverable / audit modes ────────────────────────────────────────────
thinking_parts: list[str] = []
content_parts: list[str] = []
phase = "start" # "start" | "thinking" | "content"
def _build_final_panel(content_text: str) -> Panel:
"""Render the agent's finished output as a polished Markdown panel."""
if mode == "audit":
# Re-derive verdict border from completed content
aud_name = _config.AUDITOR_AGENT
aud_title = _config.agent_title(aud_name)
aud_color = _agent_color(aud_name)
verdict_style = "dim"
for line in content_text.splitlines():
upper = line.upper()
if "VERDICT:" in upper:
if "REJECT" in upper:
verdict_style = "reject"
elif "FLAG" in upper:
verdict_style = "flag"
elif "PASS WITH NOTES" in upper:
verdict_style = "warning"
elif "PASS" in upper:
verdict_style = "pass"
break
return Panel(
Markdown(content_text),
title=f"[{aud_color}] {aud_name.capitalize()} — {aud_title} [/{aud_color}]",
border_style=verdict_style if verdict_style != "dim" else aud_color,
padding=(1, 2),
box=box.ROUNDED,
)
else:
return Panel(
Markdown(content_text),
title=f"[{color}] {title_text} [/{color}]",
border_style=color,
padding=(1, 2),
box=box.ROUNDED,
)
def _renderable(final: bool = False):
"""
Build the live renderable for the current stream state.
final=True: replace plain Text content with Markdown and apply the
verdict-aware border so the Live display ends in its final polished
form. This avoids a transient erase + reprint flash.
"""
thinking_text = "".join(thinking_parts)
content_text = "".join(content_parts)
items: list = []
# ── Agent identity line (shown while only thinking, not yet content) ──
if phase in ("start", "thinking") and not content_text:
items.append(Text(
f" {title_text}",
style=f"dim {color}",
))
# ── Thinking panel / collapsed summary ──────────────────────────────
thinking_active = (phase == "thinking")
think_renderable = _build_thinking_renderable(
thinking_text, thinking_active, agent_name, mode
)
if think_renderable is not None:
items.append(think_renderable)
# ── Content panel ────────────────────────────────────────────────────
if content_text or phase == "content":
if final and content_text:
items.append(_build_final_panel(content_text))
else:
items.append(Panel(
Text(content_text),
title=f"[{color}] {title_text} [/{color}]",
border_style=color,
padding=(1, 2),
box=box.ROUNDED,
))
if not items:
# Nothing yet — show a faint connecting indicator
return Text(
f" [{_agent_color(agent_name)} dim]{title_text} …[/{_agent_color(agent_name)} dim]",
style="dim",
)
return Group(*items) if len(items) > 1 else items[0]
console.print()
# transient=False: the Live display persists — we render the final panel
# inside it before stopping, so there is no erase/reprint cycle.
with Live(
_renderable(),
console=console,
refresh_per_second=15,
transient=False,
vertical_overflow="visible",
) as live:
async for chunk_type, text in stream_gen:
if chunk_type == "thinking":
if phase == "start":
phase = "thinking"
thinking_parts.append(text)
elif chunk_type == "content":
if phase in ("start", "thinking"):
phase = "content"
content_parts.append(text)
live.update(_renderable())
# Final update: replace plain Text with formatted Markdown panel
# (and apply verdict border for audit mode) before the Live context
# exits and the display freezes in place.
live.update(_renderable(final=True))
_last_thinking = "".join(thinking_parts)
_last_stream_rendered = True
return "".join(content_parts)
def get_last_thinking() -> str:
"""Return the thinking text captured from the most recent streamed response."""
return _last_thinking
def print_thinking_expansion(thinking_text: str):
"""Render the stored thinking output in a dim panel."""
panel = Panel(
Text(thinking_text, style="dim"),
title="[dim] Thinking — last response [/dim]",
border_style="dim",
padding=(1, 2),
box=box.ROUNDED,
)
console.print()
console.print(panel)
# ---------------------------------------------------------------------------
# Dry-run callout
# ---------------------------------------------------------------------------