grant/ten31-database
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Architect agent: Claude-powered thesis generation (backend scaffolding)

Browse Source 
- backend/mcp/architect_agent.py: generate_options + revise on Claude (prompt-
  cached thesis context, claude-opus-4-8, Ten31 voice rules). Writes N variant
  drafts to a node's variant group; nothing canonical without human approval.
  Fails gracefully if the API key / SDK is absent.
- server.py endpoints: GET /api/architect/status, GET /api/thesis/{key}/tree,
  GET /api/thesis/nodes/{id}/variants, POST .../generate, POST .../feedback,
  POST /api/thesis/lines, POST /api/thesis/lines/{key}/nodes. architect_tools
  gains get_node_variants.
- Dockerfile installs `anthropic`; docker_entrypoint loads ANTHROPIC_API_KEY from
  /data/secrets/anthropic-api-key (self-disabling until the key is dropped in).

Full HTTP surface verified end-to-end (graceful 502 without a key).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Keysat
2026-06-05 13:25:47 -05:00
parent 91361042e7
commit dd25bbc08d
5 changed files with 281 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/mcp/architect_agent.py
+137
View File
@@ -0,0 +1,137 @@
"""The Architect agent — generates and revises thesis options using Claude.
This is the one step that runs on the frontier model (Claude) rather than the
local Qwen, because sharpening narrative is exactly Claude's strength. Thesis
content is non-sensitive house messaging, so it goes to Claude directly with no
redaction. The agent reads the current thesis and writes new VARIANT drafts back
through architect_tools; nothing it produces is canonical until a human approves.
Flexibility by design (per Grant): a node can hold any number of competing
options at once. generate_options adds N drafts to a node's variant group;
partners keep, refine, branch, or retire them freely.
Requires ANTHROPIC_API_KEY in the environment (wired as a StartOS config secret
on the box). Fails gracefully with a clear message if the key or SDK is absent.
"""
import json
import os
import re
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import architect_tools as at # noqa: E402
MODEL = os.environ.get("CRM_ARCHITECT_MODEL", "claude-opus-4-8")
VOICE = ("Direct, concrete, conviction-driven. Never use 'bet'/'betting'/'gamble' "
"language (we invest with conviction). No em dashes. No 'X, not Y' "
"constructions. No kitchen-sink lists. Plain sentences a serious LP can "
"verify in their head, with real specifics where possible.")
def _client():
try:
from anthropic import Anthropic
except Exception:
raise RuntimeError("anthropic SDK not installed in this environment")
key = os.environ.get("ANTHROPIC_API_KEY")
if not key:
raise RuntimeError("ANTHROPIC_API_KEY not configured")
return Anthropic(api_key=key)
def _render_thesis(thesis):
"""Flatten the current thesis tree into text for the (cacheable) system prompt."""
line = thesis.get("line", {})
lines = [f"Thesis line: {line.get('name')} ({line.get('line_key')})"]
def walk(nodes, depth=0):
for n in nodes:
body = (n.get("body") or "").strip()
tag = n.get("node_type")
var = f" [variant_group={n['variant_group']}]" if n.get("variant_group") else ""
lines.append(f"{' ' * depth}- {tag}{var}: {n.get('title') or ''} {body}".rstrip())
walk(n.get("children", []), depth + 1)
walk(thesis.get("tree", []))
return "\n".join(lines)
def _system(thesis):
text = ("You are the Architect, the in-house copilot that sharpens Ten31's investment "
"thesis with the partners. Ten31 invests in critical infrastructure across bitcoin, "
"AI, energy, and freedom technologies, with scarcity as the connecting idea. "
f"VOICE RULES (follow exactly): {VOICE}\n\n"
"Here is the current working thesis:\n" + _render_thesis(thesis))
# Cache the thesis context so iterating across requests is cheap.
return [{"type": "text", "text": text, "cache_control": {"type": "ephemeral"}}]
def _parse_options(text):
text = re.sub(r"^```(json)?|```$", "", text.strip(), flags=re.MULTILINE).strip()
m = re.search(r"\[.*\]", text, re.DOTALL)
if not m:
return []
try:
arr = json.loads(m.group(0))
return [o for o in arr if isinstance(o, dict) and o.get("text")]
except json.JSONDecodeError:
return []
def _call(thesis, user_text, max_tokens=1800):
resp = _client().messages.create(model=MODEL, max_tokens=max_tokens,
system=_system(thesis),
messages=[{"role": "user", "content": user_text}])
return "".join(b.text for b in resp.content if getattr(b, "type", None) == "text")
def generate_options(line_key, node_id, n=3, guidance="", db=None):
"""Generate N fresh alternative versions of one node and add them as variant
drafts in that node's variant group."""
thesis = at.get_thesis(line_key, db=db)
if thesis.get("error"):
return thesis
node = at.get_node(node_id, db=db)
if node.get("error"):
return node
user = (f'Generate {n} distinct alternative versions of this {node["node_type"]} '
f'(current text: "{(node.get("body") or "").strip()}").\n'
f'Guidance from the partner: {guidance.strip() or "sharpen it and make it more concrete"}.\n'
f'Each option should be genuinely different in angle, not minor rewordings. '
f'Return ONLY a JSON array of {n} objects, each {{"text": "...", "rationale": "one short line"}}.')
raw = _call(thesis, user)
options = _parse_options(raw)[:n]
if not options:
return {"error": "no_options_parsed", "raw": raw[:300]}
group = node.get("variant_group") or node_id
# Make sure the original node shares the group, so the UI shows them together.
if not node.get("variant_group"):
at.upsert_thesis_node(node["line_id"], node["node_type"], node.get("body") or "",
title=node.get("title"), node_id=node_id, variant_group=group,
change_reason="grouped for variants", db=db)
created = []
for opt in options:
r = at.upsert_thesis_node(node["line_id"], node["node_type"], opt["text"],
title=node.get("title"), parent_id=node.get("parent_id"),
variant_group=group,
meta={"architect_generated": True, "rationale": opt.get("rationale"),
"guidance": guidance}, db=db)
created.append({"id": r.get("id"), "text": opt["text"], "rationale": opt.get("rationale")})
return {"node_id": node_id, "variant_group": group, "generated": len(created), "options": created}
def revise(line_key, node_id, feedback, n=2, db=None):
"""Revise one node in light of partner feedback, producing N options that
address it (kept as variant drafts)."""
return generate_options(line_key, node_id, n=n,
guidance=f"Address this partner feedback directly: {feedback}", db=db)
def status():
"""Whether the Architect can run (key + SDK present)."""
try:
_client()
return {"ready": True, "model": MODEL}
except Exception as exc:
return {"ready": False, "reason": str(exc)}
backend/mcp/architect_tools.py
+17
View File
@@ -95,6 +95,23 @@ def get_node(node_id, db=None):
return dict(r) if r else {"error": "not_found", "node_id": node_id}
def get_node_variants(node_id, db=None):
"""All competing options for a node (its variant group). The number is fluid:
a node may have one option or many at any moment."""
c = _conn(db)
node = c.execute("SELECT line_id, variant_group, node_type, title FROM thesis_nodes WHERE id=?", (node_id,)).fetchone()
if not node:
c.close()
return {"error": "not_found", "node_id": node_id}
group = node["variant_group"] or node_id
rows = [dict(r) for r in c.execute(
"SELECT id, body, title, status, variant_group, meta FROM thesis_nodes "
"WHERE (variant_group=? OR id=?) AND deleted_at IS NULL ORDER BY created_at", (group, node_id))]
c.close()
return {"node_id": node_id, "variant_group": group, "node_type": node["node_type"],
"title": node["title"], "variants": rows}
def get_node_history(node_id, db=None):
c = _conn(db)
rows = [dict(r) for r in c.execute(