siaochuan
f24ca87d16
Add a personalization layer that learns the user's local environment from conversation history and injects discovered rules into the system prompt. Problem: Every OH user has unique infrastructure (server IPs, data paths, conda envs, API endpoints, cron schedules) that must be manually configured in CLAUDE.md or system_prompt. This is tedious and easy to forget. Solution: Automatically extract environment-specific facts from each session and persist them as local rules that are injected into future sessions. How it works: 1. SESSION END: extractor scans conversation for patterns (SSH hosts, data paths, conda envs, API endpoints, env vars, Ray config, etc.) 2. PERSISTENCE: facts are merged into ~/.openharness/local_rules/facts.json with deduplication and confidence scoring 3. SESSION START: local rules are loaded as markdown and injected into the system prompt alongside CLAUDE.md and memory Files: - personalization/extractor.py: regex-based fact extraction (10 pattern types) - personalization/rules.py: facts persistence and rules markdown generation - personalization/session_hook.py: session-end integration - prompts/context.py: inject local rules into system prompt - ui/runtime.py: call extraction at session close (best-effort, non-blocking) - 12 tests covering extraction, merging, and markdown generation The extraction is pattern-based (no LLM calls needed), zero-cost, and runs in <10ms at session end. Facts accumulate over sessions, building a progressively richer environment profile.
88 lines
3.7 KiB
Python
88 lines
3.7 KiB
Python
"""Tests for personalization fact extraction."""
|
|
|
|
from openharness.personalization.extractor import extract_facts_from_text, facts_to_rules_markdown
|
|
from openharness.personalization.rules import merge_facts
|
|
|
|
|
|
class TestExtractFacts:
|
|
def test_extracts_ssh_host(self):
|
|
text = "ssh konghm@192.168.91.212 'tail -20 /var/log/syslog'"
|
|
facts = extract_facts_from_text(text)
|
|
ssh_facts = [f for f in facts if f["type"] == "ssh_host"]
|
|
assert len(ssh_facts) == 1
|
|
assert "konghm@192.168.91.212" in ssh_facts[0]["value"]
|
|
|
|
def test_extracts_data_path(self):
|
|
text = "ls /ext/data_auto_stage/landing/CS_sp/1d/"
|
|
facts = extract_facts_from_text(text)
|
|
path_facts = [f for f in facts if f["type"] == "data_path"]
|
|
assert any("/ext/data_auto_stage" in f["value"] for f in path_facts)
|
|
|
|
def test_extracts_conda_env(self):
|
|
text = "conda activate dev312"
|
|
facts = extract_facts_from_text(text)
|
|
conda_facts = [f for f in facts if f["type"] == "conda_env"]
|
|
assert len(conda_facts) == 1
|
|
assert conda_facts[0]["value"] == "dev312"
|
|
|
|
def test_extracts_env_var(self):
|
|
text = 'export OPENAI_BASE_URL="https://relay.nf.video/v1"'
|
|
facts = extract_facts_from_text(text)
|
|
env_facts = [f for f in facts if f["type"] == "env_var"]
|
|
assert any("OPENAI_BASE_URL" in f["value"] for f in env_facts)
|
|
|
|
def test_extracts_api_endpoint(self):
|
|
text = "curl https://api.minimax.chat/v1/chat/completions"
|
|
facts = extract_facts_from_text(text)
|
|
api_facts = [f for f in facts if f["type"] == "api_endpoint"]
|
|
assert any("minimax" in f["value"] for f in api_facts)
|
|
|
|
def test_skips_localhost(self):
|
|
text = "ping 127.0.0.1"
|
|
facts = extract_facts_from_text(text)
|
|
ip_facts = [f for f in facts if f["type"] == "ip_address"]
|
|
assert len(ip_facts) == 0
|
|
|
|
def test_deduplicates(self):
|
|
text = "ssh user@10.0.0.1\nssh user@10.0.0.1\nssh user@10.0.0.1"
|
|
facts = extract_facts_from_text(text)
|
|
ssh_facts = [f for f in facts if f["type"] == "ssh_host"]
|
|
assert len(ssh_facts) == 1
|
|
|
|
|
|
class TestMergeFacts:
|
|
def test_merge_new_facts(self):
|
|
existing = {"facts": [{"key": "ssh_host:a@1.1.1.1", "value": "a@1.1.1.1", "confidence": 0.7}]}
|
|
new = [{"key": "conda_env:dev312", "value": "dev312", "confidence": 0.7}]
|
|
merged = merge_facts(existing, new)
|
|
assert len(merged["facts"]) == 2
|
|
|
|
def test_merge_updates_higher_confidence(self):
|
|
existing = {"facts": [{"key": "ssh_host:a@1.1.1.1", "value": "a@1.1.1.1", "confidence": 0.5}]}
|
|
new = [{"key": "ssh_host:a@1.1.1.1", "value": "a@1.1.1.1", "confidence": 0.9}]
|
|
merged = merge_facts(existing, new)
|
|
assert len(merged["facts"]) == 1
|
|
assert merged["facts"][0]["confidence"] == 0.9
|
|
|
|
def test_merge_keeps_existing_if_higher(self):
|
|
existing = {"facts": [{"key": "ssh_host:a@1.1.1.1", "value": "a@1.1.1.1", "confidence": 0.9}]}
|
|
new = [{"key": "ssh_host:a@1.1.1.1", "value": "a@1.1.1.1", "confidence": 0.5}]
|
|
merged = merge_facts(existing, new)
|
|
assert merged["facts"][0]["confidence"] == 0.9
|
|
|
|
|
|
class TestFactsToMarkdown:
|
|
def test_empty_facts(self):
|
|
assert facts_to_rules_markdown([]) == ""
|
|
|
|
def test_generates_sections(self):
|
|
facts = [
|
|
{"key": "ssh_host:a@1.1", "type": "ssh_host", "value": "a@1.1", "confidence": 0.7},
|
|
{"key": "conda_env:dev312", "type": "conda_env", "value": "dev312", "confidence": 0.7},
|
|
]
|
|
md = facts_to_rules_markdown(facts)
|
|
assert "## SSH Hosts" in md
|
|
assert "## Python Environments" in md
|
|
assert "`a@1.1`" in md
|
|
assert "`dev312`" in md
|