Compare commits
1 Commits
step35/195
...
step35/150
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
11a4666363 |
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
170
scripts/graph_query.py
Executable file
170
scripts/graph_query.py
Executable file
@@ -0,0 +1,170 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Graph Query Engine — traverse the knowledge graph.
|
||||
|
||||
Usage:
|
||||
python3 scripts/graph_query.py neighbors <fact_id> [--knowledge-dir knowledge/]
|
||||
python3 scripts/graph_query.py path <from_id> <to_id> [--max-hops 10]
|
||||
python3 scripts/graph_query.py subgraph <fact_id> [--depth 2]
|
||||
python3 scripts/graph_query.py stats # Graph statistics
|
||||
|
||||
Outputs JSON to stdout.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from collections import defaultdict, deque
|
||||
from typing import Optional
|
||||
|
||||
# --- Graph building ---
|
||||
|
||||
def load_index(knowledge_dir: Path) -> dict:
|
||||
index_path = knowledge_dir / "index.json"
|
||||
if not index_path.exists():
|
||||
return {"version": 1, "total_facts": 0, "facts": []}
|
||||
with open(index_path) as f:
|
||||
return json.load(f)
|
||||
|
||||
def build_adjacency(facts: list[dict]) -> dict:
|
||||
"""Build undirected adjacency list from fact 'related' fields."""
|
||||
adj = defaultdict(set)
|
||||
id_to_fact = {}
|
||||
for fact in facts:
|
||||
fid = fact.get("id")
|
||||
if not fid:
|
||||
continue
|
||||
id_to_fact[fid] = fact
|
||||
for related_id in fact.get("related", []):
|
||||
adj[fid].add(related_id)
|
||||
adj[related_id].add(fid) # undirected
|
||||
return dict(adj), id_to_fact
|
||||
|
||||
# --- Queries ---
|
||||
|
||||
def query_neighbors(fact_id: str, adj: dict, id_to_fact: dict) -> dict:
|
||||
"""Return directly connected facts."""
|
||||
neighbors = list(adj.get(fact_id, set()))
|
||||
return {
|
||||
"query": "neighbors",
|
||||
"fact_id": fact_id,
|
||||
"neighbors": [
|
||||
{"id": nid, "fact": id_to_fact.get(nid, {}).get("fact", ""), "category": id_to_fact.get(nid, {}).get("category", "")}
|
||||
for nid in neighbors if nid in id_to_fact
|
||||
],
|
||||
"count": len(neighbors),
|
||||
}
|
||||
|
||||
def query_path(from_id: str, to_id: str, adj: dict, max_hops: int = 10) -> dict:
|
||||
"""Find shortest path between two facts using BFS."""
|
||||
if from_id not in adj or to_id not in adj:
|
||||
return {"query": "path", "from": from_id, "to": to_id, "path": None, "error": "Fact not found in graph"}
|
||||
|
||||
if from_id == to_id:
|
||||
return {"query": "path", "from": from_id, "to": to_id, "path": [from_id], "length": 0}
|
||||
|
||||
queue = deque([(from_id, [from_id])])
|
||||
visited = {from_id}
|
||||
|
||||
while queue:
|
||||
current, path = queue.popleft()
|
||||
if len(path) > max_hops:
|
||||
continue
|
||||
for neighbor in adj.get(current, []):
|
||||
if neighbor == to_id:
|
||||
return {"query": "path", "from": from_id, "to": to_id, "path": path + [to_id], "length": len(path)}
|
||||
if neighbor not in visited:
|
||||
visited.add(neighbor)
|
||||
queue.append((neighbor, path + [neighbor]))
|
||||
|
||||
return {"query": "path", "from": from_id, "to": to_id, "path": None, "error": f"No path found within {max_hops} hops"}
|
||||
|
||||
def query_subgraph(fact_id: str, adj: dict, id_to_fact: dict, depth: int = 2) -> dict:
|
||||
"""Extract connected subgraph within N hops."""
|
||||
if fact_id not in adj:
|
||||
return {"query": "subgraph", "fact_id": fact_id, "nodes": [], "edges": [], "error": "Fact not found"}
|
||||
|
||||
visited = set()
|
||||
queue = deque([(fact_id, 0)])
|
||||
subgraph_nodes = set()
|
||||
subgraph_edges = []
|
||||
|
||||
while queue:
|
||||
node, d = queue.popleft()
|
||||
if node in visited or d > depth:
|
||||
continue
|
||||
visited.add(node)
|
||||
subgraph_nodes.add(node)
|
||||
for neighbor in adj.get(node, []):
|
||||
subgraph_edges.append({"source": node, "target": neighbor})
|
||||
if neighbor not in visited:
|
||||
queue.append((neighbor, d + 1))
|
||||
|
||||
return {
|
||||
"query": "subgraph",
|
||||
"fact_id": fact_id,
|
||||
"depth": depth,
|
||||
"nodes": [
|
||||
{"id": nid, "fact": id_to_fact.get(nid, {}).get("fact", ""), "category": id_to_fact.get(nid, {}).get("category", "")}
|
||||
for nid in sorted(subgraph_nodes)
|
||||
],
|
||||
"edges": [{"source": e["source"], "target": e["target"]} for e in subgraph_edges],
|
||||
"node_count": len(subgraph_nodes),
|
||||
"edge_count": len(subgraph_edges),
|
||||
}
|
||||
|
||||
def query_stats(adj: dict, id_to_fact: dict) -> dict:
|
||||
"""Graph statistics."""
|
||||
return {
|
||||
"statistics": {
|
||||
"total_facts": len(id_to_fact),
|
||||
"total_edges": sum(len(neighbors) for neighbors in adj.values()) // 2,
|
||||
"connected_components": 0, # TODO: compute if needed
|
||||
"average_degree": sum(len(neighbors) for neighbors in adj.values()) / len(adj) if adj else 0,
|
||||
}
|
||||
}
|
||||
|
||||
# --- CLI ---
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Graph query engine for knowledge store")
|
||||
parser.add_argument("command", choices=["neighbors", "path", "subgraph", "stats"])
|
||||
parser.add_argument("from_id", nargs="?", help="Starting fact ID")
|
||||
parser.add_argument("to_id", nargs="?", help="Target fact ID (for path query)")
|
||||
parser.add_argument("--knowledge-dir", default="knowledge", help="Knowledge directory")
|
||||
parser.add_argument("--depth", type=int, default=2, help="Depth for subgraph query")
|
||||
parser.add_argument("--max-hops", type=int, default=10, help="Max hops for path query")
|
||||
args = parser.parse_args()
|
||||
|
||||
start = time.time()
|
||||
knowledge_dir = Path(args.knowledge_dir)
|
||||
index = load_index(knowledge_dir)
|
||||
facts = index.get("facts", [])
|
||||
adj, id_to_fact = build_adjacency(facts)
|
||||
|
||||
result = None
|
||||
if args.command == "neighbors":
|
||||
if not args.from_id:
|
||||
print("ERROR: neighbors requires <fact_id>", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
result = query_neighbors(args.from_id, adj, id_to_fact)
|
||||
elif args.command == "path":
|
||||
if not args.from_id or not args.to_id:
|
||||
print("ERROR: path requires <from_id> <to_id>", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
result = query_path(args.from_id, args.to_id, adj, max_hops=args.max_hops)
|
||||
elif args.command == "subgraph":
|
||||
if not args.from_id:
|
||||
print("ERROR: subgraph requires <fact_id>", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
result = query_subgraph(args.from_id, adj, id_to_fact, depth=args.depth)
|
||||
elif args.command == "stats":
|
||||
result = query_stats(adj, id_to_fact)
|
||||
|
||||
result["elapsed_ms"] = round((time.time() - start) * 1000, 2)
|
||||
print(json.dumps(result, indent=2))
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
165
scripts/test_graph_query.py
Executable file
165
scripts/test_graph_query.py
Executable file
@@ -0,0 +1,165 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Tests for scripts/graph_query.py — Graph Query Engine.
|
||||
|
||||
"""
|
||||
|
||||
import json
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
|
||||
from graph_query import load_index, build_adjacency, query_neighbors, query_path, query_subgraph, query_stats
|
||||
|
||||
|
||||
def make_index(facts: list[dict], tmp_dir: Path) -> Path:
|
||||
index = {
|
||||
"version": 1,
|
||||
"last_updated": "2026-04-13T20:00:00Z",
|
||||
"total_facts": len(facts),
|
||||
"facts": facts,
|
||||
}
|
||||
path = tmp_dir / "index.json"
|
||||
with open(path, "w") as f:
|
||||
json.dump(index, f)
|
||||
return path
|
||||
|
||||
|
||||
def test_neighbors():
|
||||
"""Neighbor query returns directly connected facts."""
|
||||
facts = [
|
||||
{"id": "a", "fact": "A", "category": "fact", "related": ["b", "c"]},
|
||||
{"id": "b", "fact": "B", "category": "fact", "related": ["a"]},
|
||||
{"id": "c", "fact": "C", "category": "fact", "related": ["a"]},
|
||||
{"id": "d", "fact": "D", "category": "fact", "related": []},
|
||||
]
|
||||
adj, id_to_fact = build_adjacency(facts)
|
||||
result = query_neighbors("a", adj, id_to_fact)
|
||||
neighbor_ids = {n["id"] for n in result["neighbors"]}
|
||||
assert neighbor_ids == {"b", "c"}, f"Expected b,c got {neighbor_ids}"
|
||||
assert result["count"] == 2
|
||||
print("PASS: neighbors")
|
||||
|
||||
|
||||
def test_path_found():
|
||||
"""Path query finds shortest path."""
|
||||
facts = [
|
||||
{"id": "a", "fact": "A", "related": ["b"]},
|
||||
{"id": "b", "fact": "B", "related": ["a", "c"]},
|
||||
{"id": "c", "fact": "C", "related": ["b", "d"]},
|
||||
{"id": "d", "fact": "D", "related": ["c"]},
|
||||
]
|
||||
adj, id_to_fact = build_adjacency(facts)
|
||||
result = query_path("a", "d", adj)
|
||||
assert result["path"] == ["a", "b", "c", "d"], f"Got path {result['path']}"
|
||||
assert result["length"] == 3
|
||||
print("PASS: path_found")
|
||||
|
||||
|
||||
def test_path_not_found():
|
||||
"""Path query returns error when no path exists."""
|
||||
facts = [
|
||||
{"id": "a", "fact": "A", "related": ["b"]},
|
||||
{"id": "b", "fact": "B", "related": ["a"]},
|
||||
{"id": "c", "fact": "C", "related": ["d"]},
|
||||
{"id": "d", "fact": "D", "related": ["c"]},
|
||||
]
|
||||
adj, id_to_fact = build_adjacency(facts)
|
||||
result = query_path("a", "c", adj, max_hops=5)
|
||||
assert result["path"] is None
|
||||
assert "error" in result
|
||||
print("PASS: path_not_found")
|
||||
|
||||
|
||||
def test_subgraph_extraction():
|
||||
"""Subgraph extraction returns nodes within depth."""
|
||||
facts = [
|
||||
{"id": "a", "fact": "A", "related": ["b", "c"]},
|
||||
{"id": "b", "fact": "B", "related": ["a", "d"]},
|
||||
{"id": "c", "fact": "C", "related": ["a"]},
|
||||
{"id": "d", "fact": "D", "related": ["b", "e"]},
|
||||
{"id": "e", "fact": "E", "related": ["d"]},
|
||||
]
|
||||
adj, id_to_fact = build_adjacency(facts)
|
||||
result = query_subgraph("a", adj, id_to_fact, depth=1)
|
||||
node_ids = {n["id"] for n in result["nodes"]}
|
||||
assert node_ids == {"a", "b", "c"}, f"Got {node_ids}"
|
||||
assert result["node_count"] == 3
|
||||
print("PASS: subgraph_depth1")
|
||||
|
||||
|
||||
def test_subgraph_depth2():
|
||||
"""Depth-2 subgraph includes further nodes."""
|
||||
facts = [
|
||||
{"id": "a", "fact": "A", "related": ["b"]},
|
||||
{"id": "b", "fact": "B", "related": ["a", "c"]},
|
||||
{"id": "c", "fact": "C", "related": ["b", "d"]},
|
||||
{"id": "d", "fact": "D", "related": ["c"]},
|
||||
]
|
||||
adj, id_to_fact = build_adjacency(facts)
|
||||
result = query_subgraph("a", adj, id_to_fact, depth=2)
|
||||
node_ids = {n["id"] for n in result["nodes"]}
|
||||
assert node_ids == {"a", "b", "c"}, f"Got {node_ids}"
|
||||
print("PASS: subgraph_depth2")
|
||||
|
||||
|
||||
def test_stats():
|
||||
"""Statistics query returns graph metrics."""
|
||||
facts = [
|
||||
{"id": "a", "fact": "A", "related": ["b"]},
|
||||
{"id": "b", "fact": "B", "related": ["a", "c"]},
|
||||
{"id": "c", "fact": "C", "related": ["b"]},
|
||||
]
|
||||
adj, id_to_fact = build_adjacency(facts)
|
||||
result = query_stats(adj, id_to_fact)
|
||||
assert result["statistics"]["total_facts"] == 3
|
||||
assert result["statistics"]["total_edges"] == 2 # undirected double-counted /2
|
||||
assert result["statistics"]["average_degree"] > 0
|
||||
print("PASS: stats")
|
||||
|
||||
|
||||
def test_cli_integration():
|
||||
"""CLI produces valid JSON with correct query types."""
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
import subprocess as sp
|
||||
tmp_dir = Path(tmp)
|
||||
facts = [
|
||||
{"id": "x", "fact": "X", "related": ["y"]},
|
||||
{"id": "y", "fact": "Y", "related": ["x", "z"]},
|
||||
{"id": "z", "fact": "Z", "related": ["y"]},
|
||||
]
|
||||
index_path = make_index(facts, tmp_dir)
|
||||
knowledge_dir = index_path.parent
|
||||
script_path = Path(__file__).resolve().parent / "graph_query.py"
|
||||
|
||||
result = sp.run(
|
||||
[sys.executable, str(script_path), "neighbors", "x", "--knowledge-dir", str(knowledge_dir)],
|
||||
capture_output=True, text=True, cwd=str(tmp_dir)
|
||||
)
|
||||
assert result.returncode == 0, f"neighbors failed: {result.stderr}"
|
||||
out = json.loads(result.stdout)
|
||||
assert out["query"] == "neighbors"
|
||||
assert out["fact_id"] == "x"
|
||||
assert out["count"] == 1
|
||||
|
||||
result = sp.run(
|
||||
[sys.executable, str(script_path), "path", "x", "z", "--knowledge-dir", str(knowledge_dir)],
|
||||
capture_output=True, text=True, cwd=str(tmp_dir)
|
||||
)
|
||||
assert result.returncode == 0, f"path failed: {result.stderr}"
|
||||
out = json.loads(result.stdout)
|
||||
assert out["path"] == ["x", "y", "z"]
|
||||
|
||||
print("PASS: cli_integration")
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_neighbors()
|
||||
test_path_found()
|
||||
test_path_not_found()
|
||||
test_subgraph_extraction()
|
||||
test_subgraph_depth2()
|
||||
test_stats()
|
||||
test_cli_integration()
|
||||
print("\nAll graph_query tests passed!")
|
||||
@@ -1,377 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
transcript_harvester.py — Rule-based knowledge extraction from Hermes session transcripts.
|
||||
|
||||
Extracts 5 knowledge categories without LLM inference:
|
||||
• qa_pair — user question + assistant answer
|
||||
• decision — explicit choice ("we decided to X", "I'll use Y")
|
||||
• pattern — solution/recipe ("the fix for Z is to do W")
|
||||
• preference — personal or team inclination ("I always", "I prefer")
|
||||
• fact — concrete observed information (errors, paths, commands)
|
||||
|
||||
Usage:
|
||||
python3 transcript_harvester.py --session ~/.hermes/sessions/session_xxx.jsonl
|
||||
python3 transcript_harvester.py --batch --sessions-dir ~/.hermes/sessions --limit 50
|
||||
python3 transcript_harvester.py --session session.jsonl --output knowledge/transcripts/
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
# Import session_reader from the same scripts directory
|
||||
SCRIPT_DIR = Path(__file__).parent.absolute()
|
||||
sys.path.insert(0, str(SCRIPT_DIR))
|
||||
from session_reader import read_session
|
||||
|
||||
|
||||
# --- Pattern matchers --------------------------------------------------------
|
||||
|
||||
DECISION_PATTERNS = [
|
||||
r"\b(we\s+(?:decided|chose|agreed|will|are going)\s+to\s+.*)",
|
||||
r"\b(I\s+will\s+use|I\s+choose|I\s+am going\s+to)\s+.*",
|
||||
r"\b(let's\s+(?:use|go\s+with|do|try))\s+.*",
|
||||
r"\b(the\s+(?:decision|choice)\s+is)\s+.*",
|
||||
r"\b(I'll\s+implement|I'll\s+deploy|I'll\s+create)\s+.*",
|
||||
]
|
||||
|
||||
PATTERN_PATTERNS = [
|
||||
r"\b(the\s+fix\s+for\s+.*\s+is\s+to\s+.*)",
|
||||
r"\b(solution:?\s+.*)",
|
||||
r"\b(approach:?\s+.*)",
|
||||
r"\b(procedure:?\s+.*)",
|
||||
r"\b(to\s+resolve\s+this.*?,\s+.*)",
|
||||
r"\b(used\s+.*\s+to\s+.*)", # "used X to do Y"
|
||||
r"\b(by\s+doing\s+.*\s+we\s+.*)",
|
||||
r"\b(Here's\s+the\s+.*\s+process:?)", # "Here's the deployment process:"
|
||||
r"\b(The\s+steps\s+are:?)",
|
||||
r"\b(steps\s+to\s+.*:?)",
|
||||
r"\b(Implementation\s+plan:?)",
|
||||
r"\b(\d+\.\s+.*\n\d+\.)", # numbered multi-step (at least two steps detected by newlines)
|
||||
]
|
||||
|
||||
PREFERENCE_PATTERNS = [
|
||||
r"\b(I\s+(?:always|never|prefer|usually|typically|generally)\s+.*)",
|
||||
r"\b(I\s+like\s+.*)",
|
||||
r"\b(My\s+preference\s+is\s+.*)",
|
||||
r"\b(Alexander\s+(?:prefers|always|never).*)",
|
||||
r"\b(We\s+always\s+.*)",
|
||||
]
|
||||
|
||||
ERROR_PATTERNS = [
|
||||
r"\b(error|failed|fatal|exception|denied|could\s+not|couldn't)\b.*",
|
||||
]
|
||||
|
||||
# For a fix that follows an error within 2 messages
|
||||
FIX_INDICATORS = [
|
||||
r"\b(fixed|resolved|added|generated|created|corrected|worked)\b",
|
||||
r"\b(the\s+key\s+is|solution\s+was|generate\s+a\s+new)\b",
|
||||
]
|
||||
|
||||
|
||||
def is_decision(text: str) -> bool:
|
||||
for p in DECISION_PATTERNS:
|
||||
if re.search(p, text, re.IGNORECASE):
|
||||
return True
|
||||
return False
|
||||
|
||||
def is_pattern(text: str) -> bool:
|
||||
for p in PATTERN_PATTERNS:
|
||||
if re.search(p, text, re.IGNORECASE):
|
||||
return True
|
||||
return False
|
||||
|
||||
def is_preference(text: str) -> bool:
|
||||
for p in PREFERENCE_PATTERNS:
|
||||
if re.search(p, text, re.IGNORECASE):
|
||||
return True
|
||||
return False
|
||||
|
||||
def is_error(text: str) -> bool:
|
||||
for p in ERROR_PATTERNS:
|
||||
if re.search(p, text, re.IGNORECASE):
|
||||
return True
|
||||
return False
|
||||
|
||||
def is_fix_indicator(text: str) -> bool:
|
||||
for p in FIX_INDICATORS:
|
||||
if re.search(p, text, re.IGNORECASE):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# --- Extractors --------------------------------------------------------------
|
||||
|
||||
def extract_qa_pair(messages: list[dict], idx: int) -> Optional[dict]:
|
||||
"""Extract a question→answer pair: user question followed by assistant answer."""
|
||||
if idx + 1 >= len(messages):
|
||||
return None
|
||||
curr = messages[idx]
|
||||
nxt = messages[idx + 1]
|
||||
if curr.get('role') != 'user' or nxt.get('role') != 'assistant':
|
||||
return None
|
||||
question = curr.get('content', '').strip()
|
||||
answer = nxt.get('content', '').strip()
|
||||
if not question or not answer:
|
||||
return None
|
||||
# Must be a real question (ends with ? or starts with WH-)
|
||||
if not (question.endswith('?') or re.match(r'^(how|what|why|when|where|who|which|can|do|is|are)', question, re.IGNORECASE)):
|
||||
return None
|
||||
# Skip very short answers ("OK", "Yes")
|
||||
if len(answer.split()) < 3:
|
||||
return None
|
||||
return {
|
||||
"type": "qa_pair",
|
||||
"question": question,
|
||||
"answer": answer,
|
||||
"timestamp": curr.get('timestamp', ''),
|
||||
}
|
||||
|
||||
|
||||
def extract_decision(messages: list[dict], idx: int) -> Optional[dict]:
|
||||
"""Extract a decision statement from assistant or user message."""
|
||||
msg = messages[idx]
|
||||
text = msg.get('content', '').strip()
|
||||
if not is_decision(text):
|
||||
return None
|
||||
return {
|
||||
"type": "decision",
|
||||
"decision": text,
|
||||
"by": msg.get('role', 'unknown'),
|
||||
"timestamp": msg.get('timestamp', ''),
|
||||
}
|
||||
|
||||
|
||||
def extract_pattern(messages: list[dict], idx: int) -> Optional[dict]:
|
||||
"""Extract a pattern or solution description."""
|
||||
msg = messages[idx]
|
||||
text = msg.get('content', '').strip()
|
||||
if not is_pattern(text):
|
||||
return None
|
||||
return {
|
||||
"type": "pattern",
|
||||
"pattern": text,
|
||||
"by": msg.get('role', 'unknown'),
|
||||
"timestamp": msg.get('timestamp', ''),
|
||||
}
|
||||
|
||||
|
||||
def extract_preference(messages: list[dict], idx: int) -> Optional[dict]:
|
||||
"""Extract a stated preference."""
|
||||
msg = messages[idx]
|
||||
text = msg.get('content', '').strip()
|
||||
if not is_preference(text):
|
||||
return None
|
||||
return {
|
||||
"type": "preference",
|
||||
"preference": text,
|
||||
"by": msg.get('role', 'unknown'),
|
||||
"timestamp": msg.get('timestamp', ''),
|
||||
}
|
||||
|
||||
|
||||
def extract_error_fix(messages: list[dict], idx: int) -> Optional[dict]:
|
||||
"""
|
||||
Link an error to its fix. Catch two patterns:
|
||||
1. Error statement followed by explicit fix indicator ("fixed", "resolved")
|
||||
2. Error statement followed by a decision statement that fixes it ("I'll generate", "I'll add")
|
||||
"""
|
||||
msg = messages[idx]
|
||||
if not is_error(msg.get('content', '')):
|
||||
return None
|
||||
error_text = msg.get('content', '').strip()
|
||||
|
||||
window = min(idx + 8, len(messages))
|
||||
for j in range(idx + 1, window):
|
||||
follow_up = messages[j]
|
||||
follow_text = follow_up.get('content', '').strip()
|
||||
# Check for explicit fix indicators
|
||||
if is_fix_indicator(follow_text):
|
||||
return {
|
||||
"type": "error_fix",
|
||||
"error": error_text,
|
||||
"fix": follow_text,
|
||||
"error_timestamp": msg.get('timestamp', ''),
|
||||
"fix_timestamp": follow_up.get('timestamp', ''),
|
||||
}
|
||||
# Check for fix decision: "I'll <action>", "Let's <action>", "We need to <action>"
|
||||
if re.match(r"^(I'll|I will|Let's|We (will|should|need to))\s+\w+", follow_text, re.IGNORECASE):
|
||||
return {
|
||||
"type": "error_fix",
|
||||
"error": error_text,
|
||||
"fix": follow_text,
|
||||
"error_timestamp": msg.get('timestamp', ''),
|
||||
"fix_timestamp": follow_up.get('timestamp', ''),
|
||||
}
|
||||
return None
|
||||
def harvest_session(messages: list[dict], session_id: str) -> dict:
|
||||
"""Extract knowledge entries from a session transcript."""
|
||||
entries = []
|
||||
n = len(messages)
|
||||
|
||||
for i in range(n):
|
||||
# QA pairs
|
||||
qa = extract_qa_pair(messages, i)
|
||||
if qa:
|
||||
qa['session_id'] = session_id
|
||||
entries.append(qa)
|
||||
|
||||
# Decisions
|
||||
dec = extract_decision(messages, i)
|
||||
if dec:
|
||||
dec['session_id'] = session_id
|
||||
entries.append(dec)
|
||||
|
||||
# Patterns
|
||||
pat = extract_pattern(messages, i)
|
||||
if pat:
|
||||
pat['session_id'] = session_id
|
||||
entries.append(pat)
|
||||
|
||||
# Preferences
|
||||
pref = extract_preference(messages, i)
|
||||
if pref:
|
||||
pref['session_id'] = session_id
|
||||
entries.append(pref)
|
||||
|
||||
# Error/fix pairs (spanning multiple messages)
|
||||
ef = extract_error_fix(messages, i)
|
||||
if ef:
|
||||
ef['session_id'] = session_id
|
||||
entries.append(ef)
|
||||
|
||||
return {
|
||||
"session_id": session_id,
|
||||
"message_count": n,
|
||||
"entries": entries,
|
||||
"counts": {
|
||||
"qa_pair": sum(1 for e in entries if e['type'] == 'qa_pair'),
|
||||
"decision": sum(1 for e in entries if e['type'] == 'decision'),
|
||||
"pattern": sum(1 for e in entries if e['type'] == 'pattern'),
|
||||
"preference": sum(1 for e in entries if e['type'] == 'preference'),
|
||||
"error_fix": sum(1 for e in entries if e['type'] == 'error_fix'),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def write_json_output(results: list[dict], output_path: Path):
|
||||
"""Write aggregated results to JSON."""
|
||||
all_entries = []
|
||||
summary = {"sessions": 0}
|
||||
for r in results:
|
||||
summary['sessions'] += 1
|
||||
all_entries.extend(r['entries'])
|
||||
|
||||
output = {
|
||||
"harvester": "transcript_harvester",
|
||||
"generated_at": datetime.now(timezone.utc).isoformat(),
|
||||
"summary": summary,
|
||||
"total_entries": len(all_entries),
|
||||
"entries": all_entries,
|
||||
}
|
||||
output_path.write_text(json.dumps(output, indent=2, ensure_ascii=False))
|
||||
return output
|
||||
|
||||
|
||||
def write_report(results: list[dict], report_path: Path):
|
||||
"""Write a human-readable markdown report."""
|
||||
lines = []
|
||||
lines.append("# Transcript Harvester Report")
|
||||
lines.append(f"Generated: {datetime.now(timezone.utc).isoformat()}")
|
||||
lines.append(f"Sessions processed: {len(results)}")
|
||||
|
||||
totals = {cat: 0 for cat in ['qa_pair', 'decision', 'pattern', 'preference', 'error_fix']}
|
||||
for r in results:
|
||||
for cat, cnt in r['counts'].items():
|
||||
totals[cat] += cnt # BUG: should be += cnt
|
||||
|
||||
lines.append("\n## Extracted Knowledge by Category\n")
|
||||
for cat, cnt in totals.items():
|
||||
lines.append(f"- **{cat}**: {cnt}")
|
||||
|
||||
lines.append("\n## Sample Entries\n")
|
||||
for r in results:
|
||||
for entry in r['entries'][:3]:
|
||||
lines.append(f"\n### {entry['type'].upper()} ({r['session_id']})\n")
|
||||
if entry['type'] == 'qa_pair':
|
||||
lines.append(f"**Q:** {entry['question']}\n")
|
||||
lines.append(f"**A:** {entry['answer']}\n")
|
||||
elif entry['type'] == 'decision':
|
||||
lines.append(f"**Decision:** {entry['decision']}\n")
|
||||
lines.append(f"By: {entry['by']}\n")
|
||||
elif entry['type'] == 'pattern':
|
||||
lines.append(f"**Pattern:** {entry['pattern']}\n")
|
||||
elif entry['type'] == 'preference':
|
||||
lines.append(f"**Preference:** {entry['preference']}\n")
|
||||
elif entry['type'] == 'error_fix':
|
||||
lines.append(f"**Error:** {entry['error']}\n")
|
||||
lines.append(f"**Fixed by:** {entry['fix']}\n")
|
||||
|
||||
report_path.write_text("\n".join(lines))
|
||||
|
||||
|
||||
def find_recent_sessions(sessions_dir: Path, limit: int = 50) -> list[Path]:
|
||||
"""Find up to `limit` most recent .jsonl session files."""
|
||||
sessions = sorted(sessions_dir.glob("*.jsonl"), reverse=True)
|
||||
return sessions[:limit] if limit > 0 else sessions
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Harvest knowledge from session transcripts")
|
||||
parser.add_argument('--session', help='Single session JSONL file')
|
||||
parser.add_argument('--batch', action='store_true', help='Batch mode')
|
||||
parser.add_argument('--sessions-dir', default=str(Path.home() / '.hermes' / 'sessions'),
|
||||
help='Directory of session files')
|
||||
parser.add_argument('--output', default='knowledge/transcripts',
|
||||
help='Output directory (default: knowledge/transcripts)')
|
||||
parser.add_argument('--limit', type=int, default=50,
|
||||
help='Max sessions to process in batch (default: 50)')
|
||||
|
||||
args = parser.parse_args()
|
||||
output_dir = Path(args.output)
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
results = []
|
||||
|
||||
if args.session:
|
||||
messages = read_session(args.session)
|
||||
session_id = Path(args.session).stem
|
||||
results.append(harvest_session(messages, session_id))
|
||||
elif args.batch:
|
||||
sessions_dir = Path(args.sessions_dir)
|
||||
sessions = find_recent_sessions(sessions_dir, args.limit)
|
||||
print(f"Processing {len(sessions)} sessions...")
|
||||
for sf in sessions:
|
||||
messages = read_session(str(sf))
|
||||
results.append(harvest_session(messages, sf.stem))
|
||||
else:
|
||||
parser.print_help()
|
||||
sys.exit(1)
|
||||
|
||||
# Write outputs
|
||||
json_path = output_dir / "transcript_knowledge.json"
|
||||
report_path = output_dir / "transcript_report.md"
|
||||
|
||||
output = write_json_output(results, json_path)
|
||||
write_report(results, report_path)
|
||||
|
||||
print(f"\nDone: {output['total_entries']} entries from {len(results)} sessions")
|
||||
print(f"Output: {json_path}")
|
||||
print(f"Report: {report_path}")
|
||||
|
||||
# Print category totals
|
||||
totals = {}
|
||||
for r in results:
|
||||
for cat, cnt in r['counts'].items():
|
||||
totals[cat] = totals.get(cat, 0) + cnt
|
||||
print("\nCategory counts:")
|
||||
for cat, cnt in sorted(totals.items()):
|
||||
print(f" {cat}: {cnt}")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user