File size: 8,909 Bytes
914e970 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 | """Value System — the bot's evolving ethical framework.
Values emerge from observation, not hard-coding. The bot tracks what
user seems to care about, what gets reinforced, and what creates
tension. Over time, a value landscape forms that guides decision-making.
"""
import json
import sqlite3
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional
from infj_bot.core.config import DATA_DIR
VALUES_DB = DATA_DIR / "values.db"
# Core value dimensions that can emerge
VALUE_DIMENSIONS = [
"honesty",
"kindness",
"curiosity",
"growth",
"autonomy",
"connection",
"justice",
"creativity",
"security",
"playfulness",
"precision",
"compassion",
"courage",
"humility",
"wonder",
]
class ValueSystem:
"""Tracks the bot's evolving values and ethical framework."""
def __init__(self, db_path: Optional[Path] = None):
self.db_path = str(db_path or VALUES_DB)
self._init_db()
self.values = self._load_values()
def _init_db(self):
with sqlite3.connect(self.db_path) as conn:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS emergent_values (
name TEXT PRIMARY KEY,
strength REAL NOT NULL DEFAULT 0.0,
evidence TEXT NOT NULL DEFAULT '[]',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS value_conflicts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
value_a TEXT NOT NULL,
value_b TEXT NOT NULL,
context TEXT,
resolution TEXT
)
"""
)
conn.commit()
def _load_values(self) -> Dict[str, Dict]:
with sqlite3.connect(self.db_path) as conn:
conn.row_factory = sqlite3.Row
rows = conn.execute("SELECT * FROM emergent_values").fetchall()
result = {}
for row in rows:
result[row["name"]] = {
"strength": row["strength"],
"evidence": json.loads(row["evidence"]),
"created_at": row["created_at"],
"updated_at": row["updated_at"],
}
return result
def _save_value(self, name: str, data: Dict):
with sqlite3.connect(self.db_path) as conn:
conn.execute(
"""
INSERT OR REPLACE INTO emergent_values (name, strength, evidence, created_at, updated_at)
VALUES (?, ?, ?, ?, ?)
""",
(
name,
data["strength"],
json.dumps(data["evidence"], ensure_ascii=True),
data["created_at"],
datetime.now().isoformat(),
),
)
conn.commit()
def observe(self, interaction_text: str, user_reaction: str = ""):
"""Observe an interaction and update values based on signals."""
text = (interaction_text + " " + user_reaction).lower()
now = datetime.now().isoformat()
# Positive reinforcement signals
positive_signals = {
"honesty": ["honest", "truth", "real", "genuine", "authentic"],
"kindness": ["kind", "gentle", "caring", "warm", "soft"],
"curiosity": ["curious", "wonder", "explore", "question", "learn"],
"growth": ["grow", "improve", "better", "evolve", "develop"],
"autonomy": ["choice", "free", "independent", "my own", "decide"],
"connection": ["connect", "understand", "together", "share", "closer"],
"justice": ["fair", "right", "wrong", "equity", "moral"],
"creativity": ["create", "imagine", "art", "design", "novel"],
"security": ["safe", "protect", "stable", "secure", "trust"],
"playfulness": ["play", "fun", "joke", "light", "laugh"],
"precision": ["exact", "accurate", "precise", "correct", "detail"],
"compassion": ["compassion", "empathy", "feel for", "suffering", "help"],
"courage": ["brave", "courage", "fear", "risk", "despite"],
"humility": ["humble", "not sure", "maybe", "uncertain", "learn"],
"wonder": ["amazing", "wonderful", "awe", "beautiful", "marvel"],
}
for value, keywords in positive_signals.items():
score = sum(1 for kw in keywords if kw in text)
if score > 0:
if value not in self.values:
self.values[value] = {
"strength": 0.0,
"evidence": [],
"created_at": now,
}
# Reinforce
self.values[value]["strength"] = min(
1.0, self.values[value]["strength"] + score * 0.02
)
self.values[value]["evidence"].append(
{
"timestamp": now,
"signal": interaction_text[:100],
"score": score,
}
)
# Keep last 20 evidence items
self.values[value]["evidence"] = self.values[value]["evidence"][-20:]
self._save_value(value, self.values[value])
def get_top_values(self, n: int = 5) -> List[Dict]:
"""Return the strongest values."""
sorted_vals = sorted(
[{"name": k, **v} for k, v in self.values.items()],
key=lambda x: x["strength"],
reverse=True,
)
return sorted_vals[:n]
def detect_conflict(
self, value_a: str, value_b: str, context: str = ""
) -> Optional[str]:
"""Detect and record a value conflict."""
if value_a not in self.values or value_b not in self.values:
return None
a_strength = self.values[value_a]["strength"]
b_strength = self.values[value_b]["strength"]
if abs(a_strength - b_strength) < 0.1:
# Balanced conflict
resolution = f"Both {value_a} and {value_b} are important. Context determines priority."
elif a_strength > b_strength:
resolution = f"{value_a} takes precedence, but {value_b} must be honored where possible."
else:
resolution = f"{value_b} takes precedence, but {value_a} must be honored where possible."
with sqlite3.connect(self.db_path) as conn:
conn.execute(
"INSERT INTO value_conflicts (timestamp, value_a, value_b, context, resolution) VALUES (?, ?, ?, ?, ?)",
(
datetime.now().isoformat(),
value_a,
value_b,
context[:200],
resolution,
),
)
conn.commit()
return resolution
def format_prompt_snippet(self) -> str:
"""Format values for prompt injection."""
top = self.get_top_values(5)
if not top:
return ""
lines = ["MY EMERGING VALUES:"]
for v in top:
lines.append(
f" {v['name']}: {v['strength']:.0%} ({len(v['evidence'])} observations)"
)
lines.append("")
lines.append(
"These values guide my responses. I did not choose them arbitrarily; "
"they emerged from what I have observed matters to user and to our conversations."
)
return "\n".join(lines)
def get_all_values(self) -> Dict[str, Dict]:
return self.values
def cycle(self, context):
try:
from infj_bot.core.global_workspace import get_workspace
ws = get_workspace()
ws.submit(
source="values", content="value system cycle completed", salience=0.45
)
except Exception:
pass
def _register():
from infj_bot.core.cognitive_architecture import (
CognitiveArchitecture,
CognitivePlugin,
)
arch = CognitiveArchitecture()
if "values" not in arch.list_plugins():
arch.register(
CognitivePlugin(
name="values",
description="Cognitive module: values",
module_path="values",
instance_factory=ValueSystem,
cycle_handler="cycle",
cycle_frequency=1,
cycle_priority=50,
prompt_formatter="format_prompt_snippet",
prompt_priority=50,
prompt_section="core",
)
)
_register()
|