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

Updated tool-calling

Browse Source
This commit is contained in:
vh
2026-04-03 22:11:36 -07:00
parent 4e411f8060
commit 7472654587
150 changed files with 13014 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
orchestration/orchestrator.py
+152 -47
View File
@@ -17,6 +17,8 @@ import re
from dataclasses import dataclass, field
from datetime import datetime
_THINK_TAG_RE = re.compile(r"<think>.*?</think>", re.DOTALL | re.IGNORECASE)
import config
import task_store
import tools as _tools
@@ -80,7 +82,8 @@ class OpenAIClient(ProviderClient):
messages=full_messages,
extra_body=self.extra_body or None,
)
return response.choices[0].message.content or ""
content = response.choices[0].message.content or ""
return _THINK_TAG_RE.sub("", content).strip()
def call(self, system, messages, model, temperature, max_tokens):
return asyncio.run(
@@ -100,8 +103,9 @@ class OpenAIClient(ProviderClient):
extra_body=self.extra_body or None,
)
msg = response.choices[0].message
content = _THINK_TAG_RE.sub("", msg.content or "").strip()
return _ToolMessage(
content=msg.content or "",
content=content,
tool_calls=msg.tool_calls or None,
)
@@ -432,14 +436,24 @@ async def call_agent_async(
dry_run: bool = False,
stream: bool = True,
stream_mode: str = "deliverable",
use_tools: bool = True,
) -> str:
"""
Call an agent with a user message and return the response.
use_tools controls whether the tool-calling loop is engaged.
Pass use_tools=False to force a plain streaming/non-streaming call
even when the agent has tools assigned — used for synthesis and
standing-brief updates where tool dispatch must not occur.
"""
if dry_run:
return f"[DRY RUN] Would call {agent.name} ({config.agent_provider_name(agent.name)}) with: {user_message[:100]}..."
agent.add_user(user_message)
# If the agent has tools, use the tool-calling loop instead of streaming directly
if agent.tools:
# Route through the tool loop only when the agent has tools AND the caller
# has not explicitly suppressed tool use for this call.
if agent.tools and use_tools:
try:
return await _run_tool_loop(agent, stream_mode=stream_mode, dry_run=dry_run)
except NotImplementedError as e:
@@ -478,34 +492,43 @@ async def call_agent_async(
# ---------------------------------------------------------------------------
# Task Brief Parser
# Lead Brief Formatter
# Converts a dispatch_task spec dict into the structured brief sent to a lead.
# ---------------------------------------------------------------------------
def parse_task_briefs(orchestrator_response: str, task_id_prefix: str) -> list[Task]:
tasks = []
blocks = re.findall(
r"TASK BRIEF\s*\n(.+?)(?=TASK BRIEF|\Z)",
orchestrator_response,
re.DOTALL | re.IGNORECASE,
)
for i, block in enumerate(blocks):
to_match = re.search(r"To:\s*(.+)", block)
if not to_match:
continue
to_text = to_match.group(1).lower()
lead = next(
(name for name in config.LEAD_AGENT_NAMES if name in to_text), None
)
if not lead:
continue
task_id = f"{task_id_prefix}-{chr(65 + i)}"
tasks.append(Task(
task_id=task_id,
directive=to_text,
assigned_to=lead,
brief=f"TASK BRIEF\n{block.strip()}",
))
return tasks
def _format_lead_brief(task_id: str, spec: dict) -> str:
"""Build the task brief message sent to a lead from a dispatch_task spec."""
lead_name = spec["to"]
lead_title = config.agent_title(lead_name)
lines = [
f"TASK {task_id}",
f"From: Miranda, Chief of Staff",
f"To: {lead_title}",
"",
f"DIRECTIVE: {spec['directive']}",
"",
f"SCOPE: {spec['scope']}",
]
if spec.get("constraints") and spec["constraints"].upper() != "NONE":
lines += ["", f"CONSTRAINTS: {spec['constraints']}"]
if spec.get("dependencies") and spec["dependencies"].upper() != "NONE":
lines += ["", f"DEPENDENCIES: {spec['dependencies']}"]
if spec.get("return_format"):
lines += ["", f"RETURN FORMAT: {spec['return_format']}"]
lines += [
"",
"---",
"Return your response using this exact structure:",
"",
"STATUS: Complete | Partial | Blocked",
"SUMMARY: [2–5 sentence executive summary of your findings]",
"",
"FINDINGS:",
"[Your full output here]",
"",
"OPEN ITEMS: [Blockers, unresolved questions, or follow-up work — or NONE]",
]
return "\n".join(lines)
# ---------------------------------------------------------------------------
@@ -615,7 +638,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, stream_mode="background")
return await call_agent_async(orchestrator, update_request, dry_run=dry_run, stream_mode="background", use_tools=False)
# ---------------------------------------------------------------------------
@@ -641,6 +664,7 @@ class Session:
ui.print_system("Loading agents…")
self.orchestrator = build_agent(config.ORCHESTRATOR_AGENT)
self.auditor = build_agent(config.AUDITOR_AGENT)
self.summarizer = build_agent(config.SUMMARIZER_AGENT) if config.SUMMARIZER_AGENT else None
self.leads = {
name: build_agent(name) for name in config.LEAD_AGENT_NAMES
}
@@ -662,26 +686,37 @@ class Session:
ui.print_task_received(task_id, directive)
# ── Step 1: Orchestrator receives the directive ─────────────────────
# Safety: clear any leftover queued tasks from a previous aborted directive
_tools.get_pending_tasks()
# ── Step 1: Miranda analyses the directive, dispatches via tool calls ─
cos_prompt = (
f"The Principal has issued the following directive. "
f"Analyse it, determine what work is needed, and either:\n"
f" (a) Respond directly if no Lead work is required, or\n"
f" (b) Issue TASK BRIEF blocks for the appropriate Leads.\n\n"
f"Analyse it, then either respond directly or use the dispatch_task tool "
f"to assign work to the appropriate Leads.\n\n"
f"Directive: {directive}"
)
cos_response = await call_agent_async(
self.orchestrator, cos_prompt, dry_run=self.dry_run, stream=True
)
# ── Step 2: Parse any task briefs the orchestrator issued ───────────
tasks = parse_task_briefs(cos_response, task_id)
# ── Step 2: Drain task specs queued during Miranda's tool loop ───────
task_specs = _tools.get_pending_tasks()
if not tasks:
if not task_specs:
ui.print_direct_response_notice()
deliverable = cos_response
else:
# ── Step 3: Persist task records then dispatch ──────────────────
# ── Step 3: Build Task objects, persist, and dispatch ───────────
tasks = [
Task(
task_id=f"{task_id}-{chr(65 + i)}",
directive=directive,
assigned_to=spec["to"],
brief=_format_lead_brief(f"{task_id}-{chr(65 + i)}", spec),
)
for i, spec in enumerate(task_specs)
]
ui.print_task_dispatch_plan(tasks)
task_store.record_tasks(tasks, self.session_id, directive)
self.all_tasks.extend(tasks)
@@ -689,20 +724,21 @@ class Session:
self.leads, tasks, dry_run=self.dry_run, session_id=self.session_id,
)
# ── Step 4: Orchestrator synthesizes ────────────────────────────
# ── Step 4: Miranda synthesizes lead outputs ────────────────────
lead_outputs_text = "\n\n".join(
f"--- Output from {t.assigned_to.upper()} [{t.task_id}] ---\n{t.output}"
f"--- {t.assigned_to.upper()} [{t.task_id}] ---\n{t.output}"
if t.status == "complete"
else f"--- {t.assigned_to.upper()} [{t.task_id}] FAILED: {t.error} ---"
for t in completed_tasks
)
synthesis_prompt = (
f"All Lead outputs have returned for {task_id}. "
f"Please synthesize these into a final deliverable for the Principal.\n\n"
f"Synthesize these into your final deliverable for the Principal.\n\n"
f"{lead_outputs_text}"
)
deliverable = await call_agent_async(
self.orchestrator, synthesis_prompt, dry_run=self.dry_run, stream=True
self.orchestrator, synthesis_prompt, dry_run=self.dry_run,
stream=True, use_tools=False,
)
dispatched_leads = sorted({t.assigned_to for t in completed_tasks})
@@ -713,19 +749,85 @@ class Session:
self.last_deliverable = deliverable
# ── Step 5: Auditor review ──────────────────────────────────────────
# ── Step 5: Auditor + Miranda summary in parallel ───────────────────
# Vera streams (Rich Live — UI only); Tillie calls the API silently
# (stream=False, no Rich output), so they do not conflict visually.
# We use FIRST_COMPLETED so Tillie's panel appears as soon as she
# resolves, even if Vera is still streaming.
audit_memo = ""
if self.auto_audit:
vera_task = asyncio.create_task(
run_auditor(self.auditor, deliverable, task_id, dry_run=self.dry_run)
)
tillie_task = asyncio.create_task(
self._call_summarizer(deliverable)
)
done, pending = await asyncio.wait(
{vera_task, tillie_task},
return_when=asyncio.FIRST_COMPLETED,
)
# If Tillie resolved first, display immediately (Vera may still be streaming)
miranda_summary = ""
if tillie_task in done:
try:
miranda_summary = tillie_task.result()
if miranda_summary:
ui.print_summary(miranda_summary, source_agent=config.ORCHESTRATOR_AGENT)
except Exception:
pass
# Ensure Vera finishes
try:
audit_memo = await run_auditor(
self.auditor, deliverable, task_id, dry_run=self.dry_run
)
audit_memo = await vera_task
self.session_log.append(f"Audit complete for {task_id}.")
except RuntimeError as e:
ui.print_warning(f"Audit skipped — {e}")
# If Vera resolved first, wait for Tillie now (with spinner) then display
if tillie_task in pending:
with ui.summary_progress():
try:
miranda_summary = await tillie_task
except Exception:
pass
if miranda_summary:
ui.print_summary(miranda_summary, source_agent=config.ORCHESTRATOR_AGENT)
else:
await self._summarize(deliverable, source_agent=config.ORCHESTRATOR_AGENT)
# ── Tillie: summarize Vera's audit memo (sequential — depends on Vera) ─
if audit_memo:
await self._summarize(audit_memo, source_agent=config.AUDITOR_AGENT)
return deliverable, audit_memo
async def _call_summarizer(self, content: str) -> str:
"""Call Tillie silently and return the summary text. No UI output.
Used when running in parallel with another agent that owns the display."""
if not self.summarizer or not content:
return ""
self.summarizer.history = []
return await call_agent_async(
self.summarizer,
f"Summarize the following:\n\n{content}",
dry_run=self.dry_run,
stream=False,
) or ""
async def _summarize(self, content: str, source_agent: str = "") -> None:
"""Call Tillie with a progress spinner and display the summary panel."""
if not self.summarizer or not content:
return
try:
with ui.summary_progress():
summary = await self._call_summarizer(content)
if summary:
ui.print_summary(summary, source_agent=source_agent)
except Exception as e:
ui.print_warning(f"Summary skipped — {e}")
async def close(self):
summary = "\n".join(self.session_log)
updated_brief = await update_standing_brief(
@@ -818,6 +920,7 @@ async def cmd_reload(session: Session, args: str) -> bool:
# Rebuild all agents with fresh config
session.orchestrator = build_agent(config.ORCHESTRATOR_AGENT)
session.auditor = build_agent(config.AUDITOR_AGENT)
session.summarizer = build_agent(config.SUMMARIZER_AGENT) if config.SUMMARIZER_AGENT else None
session.leads = {
name: build_agent(name) for name in config.LEAD_AGENT_NAMES
}
@@ -849,6 +952,8 @@ async def cmd_audit(session: Session, args: str) -> bool:
session.last_task_id, dry_run=session.dry_run,
)
ui.print_audit_memo(memo, task_id=session.last_task_id)
if memo:
await session._summarize(memo, source_agent=config.AUDITOR_AGENT)
return True