mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
brain: port graph-links + recipes to hanzo-memory (Python parity with bot/extensions)
Adds the zero-LLM typed-link extractor + YAML recipe loader to the
existing hanzo-memory Python SDK so a single brain.db file can be
written by the TS bot and read by the Python SDK and vice versa.
graph_links.py — Edge dataclass, extract_edges(), reconcile(),
slugify(). Same six edge types
(mentions/attended/works_at/invested_in/founded/
advises), same regex + role inference, same code-
fence stripping. Pure — no I/O, no LLM.
recipes.py — list_recipes() + load_recipe(name). Reads built-in
recipes/ dir + optional HANZO_BRAIN_RECIPES path.
recipes/email.yaml — flagship swipe-to-reply recipe, byte-identical
to the TS pack so cross-runtime brains agree.
tests/test_graph_links.py — mirror of TS suite. 12 assertions cover
slugify, every edge type, code-fence
stripping, dedup, bare slugs, reconcile,
and recipe load.
hanzo-memory already had pluggable backends (SQLite + sqlite-vec via
.db.factory) and an MCP server; this fills in the gbrain-shape graph +
recipe surfaces so the Python SDK reaches feature parity with the bot
extensions in TS.
This commit is contained in:
@@ -0,0 +1,147 @@
|
||||
"""Hanzo Brain — typed-link extractor (Python port of @hanzo/bot-graph-links).
|
||||
|
||||
Zero-LLM. Pure regex + role inference. Mirrors the TS extractor 1:1 so a
|
||||
brain.db written by the bot is consumed identically by the Python SDK
|
||||
and vice versa. Suitable for >10K pages/sec.
|
||||
|
||||
Edge types: mentions / attended / works_at / invested_in / founded / advises.
|
||||
|
||||
Schema target — `edges` table:
|
||||
source TEXT, target TEXT, type TEXT, evidence TEXT,
|
||||
PRIMARY KEY (source, target, type)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import unicodedata
|
||||
from dataclasses import dataclass
|
||||
from typing import Iterable, Literal
|
||||
|
||||
EdgeType = Literal["mentions", "attended", "works_at", "invested_in", "founded", "advises"]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Edge:
|
||||
source: str
|
||||
target: str
|
||||
type: EdgeType
|
||||
evidence: str | None = None
|
||||
|
||||
|
||||
# ── Patterns ────────────────────────────────────────────────────────
|
||||
|
||||
_MD_LINK = re.compile(r"\[([^\]]+)\]\(([^)#\s]+)\)")
|
||||
_BARE_SLUG = re.compile(
|
||||
r"(?<![/\w])@?((?:people|companies|deals|projects|investors|firms)/[a-z0-9][a-z0-9-]*)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_CODE_FENCE = re.compile(r"```[\s\S]*?```")
|
||||
_INLINE_CODE = re.compile(r"`[^`\n]+`")
|
||||
|
||||
# Order matters — first match wins per (source, target).
|
||||
_ROLE_PATTERNS: list[tuple[re.Pattern[str], EdgeType]] = [
|
||||
# FOUNDED
|
||||
(re.compile(r"\b(?:co-?)?founded\s+([^.\n]+?)(?:[.\n]|$)", re.IGNORECASE), "founded"),
|
||||
(re.compile(r"\bfounder\s+(?:and\s+\w+\s+)?of\s+([^.\n]+?)(?:[.\n]|$)", re.IGNORECASE), "founded"),
|
||||
# INVESTED_IN
|
||||
(re.compile(r"\binvested\s+in\s+([^.\n]+?)(?:[.\n]|$)", re.IGNORECASE), "invested_in"),
|
||||
(re.compile(r"\bled\s+([^.\n]+?)['’]s\s+(?:seed|series|round)", re.IGNORECASE), "invested_in"),
|
||||
(re.compile(r"\bwrote\s+(?:a\s+)?check\s+into\s+([^.\n]+?)(?:[.\n]|$)", re.IGNORECASE), "invested_in"),
|
||||
# ADVISES
|
||||
(re.compile(r"\badvises\s+([^.\n]+?)(?:[.\n]|$)", re.IGNORECASE), "advises"),
|
||||
(re.compile(r"\badvisor\s+(?:to|at|for)\s+([^.\n]+?)(?:[.\n]|$)", re.IGNORECASE), "advises"),
|
||||
# WORKS_AT
|
||||
(
|
||||
re.compile(
|
||||
r"\b(?:CEO|CTO|COO|CFO|VP|head\s+of\s+\w+|director)\s+of\s+([^.\n]+?)(?:[.\n]|$)",
|
||||
re.IGNORECASE,
|
||||
),
|
||||
"works_at",
|
||||
),
|
||||
(
|
||||
re.compile(
|
||||
r"\bjoined\s+([A-Z][^\s.]*(?:\s+[A-Z][^\s.]*)*)(?=\s+(?:as|in)\b|[\s.,;:!?\n]|$)"
|
||||
),
|
||||
"works_at",
|
||||
),
|
||||
(re.compile(r"\bworks\s+at\s+([^.\n]+?)(?:[.\n]|$)", re.IGNORECASE), "works_at"),
|
||||
]
|
||||
|
||||
|
||||
def slugify(s: str) -> str:
|
||||
"""Match gbrain's slug convention — lowercase ascii dashes."""
|
||||
s = unicodedata.normalize("NFKD", s)
|
||||
s = "".join(c for c in s if not unicodedata.combining(c))
|
||||
s = s.lower().replace("&", " and ")
|
||||
s = re.sub(r"[^a-z0-9]+", "-", s)
|
||||
s = s.strip("-")
|
||||
return s[:80]
|
||||
|
||||
|
||||
def _strip_code(md: str) -> str:
|
||||
md = _CODE_FENCE.sub("", md)
|
||||
md = _INLINE_CODE.sub("", md)
|
||||
return md
|
||||
|
||||
|
||||
def _infer_category(edge_type: EdgeType) -> str:
|
||||
if edge_type in ("founded", "invested_in", "works_at"):
|
||||
return "companies"
|
||||
if edge_type == "advises":
|
||||
return "people"
|
||||
return "entities"
|
||||
|
||||
|
||||
def extract_edges(
|
||||
slug: str,
|
||||
content: str,
|
||||
page_type: str | None = None,
|
||||
) -> list[Edge]:
|
||||
"""Extract typed edges from one page. Pure — no I/O, no LLM."""
|
||||
cleaned = _strip_code(content)
|
||||
seen: dict[tuple[str, EdgeType], Edge] = {}
|
||||
|
||||
def add(e: Edge) -> None:
|
||||
key = (e.target, e.type)
|
||||
seen.setdefault(key, e)
|
||||
|
||||
# 1. Markdown links — `mentions`, or `attended` on meeting pages.
|
||||
for m in _MD_LINK.finditer(cleaned):
|
||||
target = m.group(2).strip()
|
||||
if target.startswith("http") or target.startswith("/") or "/" not in target:
|
||||
continue
|
||||
et: EdgeType = "attended" if page_type == "meeting" else "mentions"
|
||||
add(Edge(source=slug, target=target, type=et, evidence=m.group(0)))
|
||||
|
||||
# 2. Bare slug refs (`people/alice`).
|
||||
for m in _BARE_SLUG.finditer(cleaned):
|
||||
et = "attended" if page_type == "meeting" else "mentions"
|
||||
add(Edge(source=slug, target=m.group(1).lower(), type=et, evidence=m.group(0)))
|
||||
|
||||
# 3. Role inference.
|
||||
for pat, etype in _ROLE_PATTERNS:
|
||||
m = pat.search(cleaned)
|
||||
if not m:
|
||||
continue
|
||||
raw = m.group(1).strip().rstrip(".,;:!?")
|
||||
target_slug = f"{_infer_category(etype)}/{slugify(raw)}"
|
||||
if target_slug.endswith("/"):
|
||||
continue
|
||||
add(Edge(source=slug, target=target_slug, type=etype, evidence=m.group(0)))
|
||||
|
||||
return list(seen.values())
|
||||
|
||||
|
||||
def reconcile(prior: Iterable[Edge], next_: Iterable[Edge]) -> tuple[list[Edge], list[Edge]]:
|
||||
"""Return (add, remove) deltas between two edge sets.
|
||||
|
||||
Used by the persistence layer to stale-delete dropped refs when a
|
||||
page is edited — same contract as the TS extractor's `reconcile`.
|
||||
"""
|
||||
prior_set = {(e.source, e.target, e.type) for e in prior}
|
||||
next_list = list(next_)
|
||||
next_set = {(e.source, e.target, e.type) for e in next_list}
|
||||
prior_list = [e for e in prior if (e.source, e.target, e.type) not in next_set]
|
||||
add = [e for e in next_list if (e.source, e.target, e.type) not in prior_set]
|
||||
return add, prior_list
|
||||
@@ -0,0 +1,58 @@
|
||||
"""Hanzo Brain — recipe loader (Python port of @hanzo/bot-recipes-brain).
|
||||
|
||||
YAML recipes for daily-life automation:
|
||||
auth + cron + ingest + classify + draft + enqueue + on_swipe.
|
||||
|
||||
Loads recipes from `<this_package>/recipes/*.yaml` plus any
|
||||
user-defined dir in `HANZO_BRAIN_RECIPES`. Same shape as the TS pack
|
||||
so a single brain.db file works for either runtime.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
|
||||
_HERE = Path(__file__).parent
|
||||
_BUILTIN = _HERE / "recipes"
|
||||
|
||||
|
||||
def _recipe_dirs() -> list[Path]:
|
||||
dirs: list[Path] = []
|
||||
if _BUILTIN.is_dir():
|
||||
dirs.append(_BUILTIN)
|
||||
env_dir = os.environ.get("HANZO_BRAIN_RECIPES")
|
||||
if env_dir:
|
||||
p = Path(env_dir).expanduser()
|
||||
if p.is_dir():
|
||||
dirs.append(p)
|
||||
return dirs
|
||||
|
||||
|
||||
def list_recipes() -> list[str]:
|
||||
"""Return the names (without `.yaml`) of every available recipe."""
|
||||
seen: dict[str, None] = {}
|
||||
for d in _recipe_dirs():
|
||||
for f in d.glob("*.yaml"):
|
||||
seen.setdefault(f.stem, None)
|
||||
return list(seen.keys())
|
||||
|
||||
|
||||
def load_recipe(name: str) -> dict[str, Any]:
|
||||
"""Load and parse one recipe by name."""
|
||||
for d in _recipe_dirs():
|
||||
path = d / f"{name}.yaml"
|
||||
if path.is_file():
|
||||
with path.open("r", encoding="utf-8") as f:
|
||||
data = yaml.safe_load(f) or {}
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError(f"recipe `{name}` did not parse to a mapping")
|
||||
return data
|
||||
available = ", ".join(list_recipes()) or "(none)"
|
||||
raise FileNotFoundError(
|
||||
f"recipe `{name}` not found. Available: {available}. "
|
||||
f"Drop a yaml into {_BUILTIN} or set HANZO_BRAIN_RECIPES to your own dir."
|
||||
)
|
||||
@@ -0,0 +1,92 @@
|
||||
# Hanzo Brain — Email Recipe
|
||||
#
|
||||
# Daily Gmail pull → classify → draft response → enqueue for swipe-review
|
||||
# in the Hanzo app (~/work/hanzo/bot/apps/{ios,android,macos}).
|
||||
#
|
||||
# Run:
|
||||
# hanzo-bot brain run-recipe email
|
||||
# hanzo-bot brain drafts list
|
||||
# hanzo-bot brain drafts approve <id> # sends via gmail
|
||||
# hanzo-bot brain drafts reject <id> # logs as decline-fact for training
|
||||
#
|
||||
# Trains your reply tone over time. Every swipe is a signal.
|
||||
|
||||
recipe: email
|
||||
version: 1
|
||||
backend: gmail
|
||||
|
||||
auth:
|
||||
provider: hanzo.id # OIDC via hanzo IAM — single sign-on
|
||||
scopes:
|
||||
- https://www.googleapis.com/auth/gmail.readonly
|
||||
- https://www.googleapis.com/auth/gmail.compose
|
||||
|
||||
cron: "*/30 * * * *" # every 30 min — adjust per volume
|
||||
|
||||
ingest:
|
||||
source: inbox
|
||||
since: 24h
|
||||
max: 100
|
||||
filter:
|
||||
has_thread: any
|
||||
skip_labels: [SPAM, TRASH, PROMOTIONS]
|
||||
|
||||
classify:
|
||||
model: zen-2-haiku # cheap classifier, fast
|
||||
schema:
|
||||
needs_reply: bool
|
||||
priority: one_of [P0, P1, P2, P3]
|
||||
intent: one_of [question, fyi, scheduling, ask, update, spam]
|
||||
entities: [person, company, deal]
|
||||
sentiment: one_of [positive, neutral, negative]
|
||||
action_items: list
|
||||
|
||||
draft:
|
||||
when: needs_reply == true
|
||||
model: zen-2 # main model — bigger context, better tone
|
||||
context:
|
||||
- brain.recall(from_email) # who is this person (facts)
|
||||
- brain.search(thread_id, limit=10) # past replies in thread (RAG)
|
||||
- brain.facts({entity:from_email, since:30d, limit:30})
|
||||
- brain.style(sender:me, recent:200) # 200-shot style examples from outbox
|
||||
prompt: |
|
||||
You are drafting a reply for me. Match my tone exactly — terse,
|
||||
direct, no exclamation marks, no emoji. Answer the question in the
|
||||
fewest words that are still useful. Sign off only if I usually do.
|
||||
Never invent commitments. If you're unsure, suggest "let me check
|
||||
and get back to you" and add an action_item.
|
||||
output_schema:
|
||||
subject: string
|
||||
body: string
|
||||
action_items: list
|
||||
confidence: number # 0..1, used for triage ranking
|
||||
|
||||
enqueue:
|
||||
queue: drafts.email
|
||||
ttl: 7d
|
||||
rank_by: [priority, confidence]
|
||||
fact_on_enqueue:
|
||||
subject: from_email
|
||||
predicate: thread_active
|
||||
object: true
|
||||
source: thread_id
|
||||
|
||||
notify:
|
||||
channel: hanzo-app
|
||||
badge: count_unswiped
|
||||
push: drafts.email.new
|
||||
|
||||
# After the user swipes in the app, the result feeds back to the brain:
|
||||
on_swipe_send:
|
||||
- send via gmail
|
||||
- upsertFact: { subject: from_email, predicate: replied, object: thread_id, ts: now }
|
||||
- learn: append (received_msg, sent_draft) to style examples
|
||||
|
||||
on_swipe_reject:
|
||||
- delete draft
|
||||
- upsertFact: { subject: from_email, predicate: declined_reply, object: thread_id, ts: now }
|
||||
- learn: down-weight similar drafts in future ranking
|
||||
|
||||
on_swipe_edit:
|
||||
- open editor with the draft pre-filled
|
||||
- on save: send + learn (received_msg, edited_draft) as positive example
|
||||
@@ -0,0 +1,116 @@
|
||||
"""Tests for hanzo_memory.graph_links — mirror of @hanzo/bot-graph-links TS suite."""
|
||||
|
||||
from hanzo_memory.graph_links import Edge, extract_edges, reconcile, slugify
|
||||
|
||||
|
||||
# ── slugify ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_slugify_lowercases_dashes_ascii() -> None:
|
||||
assert slugify("Acme AI Inc.") == "acme-ai-inc"
|
||||
assert slugify("José's Pizza") == "jose-s-pizza"
|
||||
assert slugify("Slack & Discord") == "slack-and-discord"
|
||||
|
||||
|
||||
# ── extract_edges ───────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _has(edges: list[Edge], **fields: object) -> bool:
|
||||
return any(all(getattr(e, k) == v for k, v in fields.items()) for e in edges)
|
||||
|
||||
|
||||
def test_mentions_from_md_links() -> None:
|
||||
edges = extract_edges("originals/idea-1", "Inspired by [Alice](people/alice) at Acme.")
|
||||
assert _has(edges, target="people/alice", type="mentions")
|
||||
|
||||
|
||||
def test_meeting_emits_attended_not_mentions() -> None:
|
||||
edges = extract_edges(
|
||||
"meetings/2026-05-10",
|
||||
"Met with [Bob](people/bob) and [Carol](people/carol).",
|
||||
page_type="meeting",
|
||||
)
|
||||
types = {e.type for e in edges}
|
||||
assert "attended" in types
|
||||
assert "mentions" not in types
|
||||
|
||||
|
||||
def test_founded_inference() -> None:
|
||||
edges = extract_edges("people/alice", "Alice co-founded Acme AI. She also runs Beta Co.")
|
||||
assert _has(edges, type="founded", target="companies/acme-ai")
|
||||
|
||||
|
||||
def test_invested_in_inference() -> None:
|
||||
e1 = extract_edges("people/dan", "Dan invested in Foobar.")
|
||||
assert _has(e1, type="invested_in", target="companies/foobar")
|
||||
|
||||
e2 = extract_edges("people/erin", "Erin led Quux's seed round.")
|
||||
assert _has(e2, type="invested_in", target="companies/quux")
|
||||
|
||||
|
||||
def test_advises_inference() -> None:
|
||||
edges = extract_edges("people/frank", "Frank is an advisor to Globex.")
|
||||
assert _has(edges, type="advises", target="people/globex")
|
||||
|
||||
|
||||
def test_works_at_inference() -> None:
|
||||
e1 = extract_edges("people/grace", "Grace is the CEO of Acme.")
|
||||
assert _has(e1, type="works_at", target="companies/acme")
|
||||
|
||||
e2 = extract_edges("people/henry", "Henry joined Initech in 2024.")
|
||||
assert _has(e2, type="works_at", target="companies/initech")
|
||||
|
||||
|
||||
def test_strip_code_fences() -> None:
|
||||
edges = extract_edges(
|
||||
"concepts/snippet",
|
||||
"Normal: [link](people/real). Code:\n```\nfake = [should](people/fake)\n```\n",
|
||||
)
|
||||
targets = {e.target for e in edges}
|
||||
assert "people/real" in targets
|
||||
assert "people/fake" not in targets
|
||||
|
||||
|
||||
def test_dedup_same_target_same_type() -> None:
|
||||
edges = extract_edges("originals/x", "[Alice](people/alice) and again [Alice](people/alice).")
|
||||
alice = [e for e in edges if e.target == "people/alice" and e.type == "mentions"]
|
||||
assert len(alice) == 1
|
||||
|
||||
|
||||
def test_bare_slug_refs() -> None:
|
||||
edges = extract_edges("concepts/note", "See people/alice and companies/acme-ai.")
|
||||
targets = {e.target for e in edges}
|
||||
assert "people/alice" in targets
|
||||
assert "companies/acme-ai" in targets
|
||||
|
||||
|
||||
# ── reconcile ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_reconcile_add_remove() -> None:
|
||||
prior = [Edge("a", "x", "mentions"), Edge("a", "y", "mentions")]
|
||||
next_ = [Edge("a", "y", "mentions"), Edge("a", "z", "mentions")]
|
||||
add, remove = reconcile(prior, next_)
|
||||
assert [e.target for e in add] == ["z"]
|
||||
assert [e.target for e in remove] == ["x"]
|
||||
|
||||
|
||||
def test_reconcile_no_change() -> None:
|
||||
same = [Edge("a", "x", "mentions")]
|
||||
add, remove = reconcile(same, same)
|
||||
assert add == [] and remove == []
|
||||
|
||||
|
||||
# ── recipes ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_recipes_list_and_load() -> None:
|
||||
from hanzo_memory.recipes import list_recipes, load_recipe
|
||||
|
||||
names = list_recipes()
|
||||
assert "email" in names
|
||||
|
||||
email = load_recipe("email")
|
||||
assert email["recipe"] == "email"
|
||||
assert email["version"] == 1
|
||||
assert email["cron"].startswith("*/30")
|
||||
Reference in New Issue
Block a user