#!/usr/bin/env python3 """人物档案管理器 - Profile Manager 功能： 1. 基础档案：姓名、头像、简介、偏好 2. 事实抽取：从对话中提取事实三元组 3. 关系图谱：维护人与人之间的关系 4. 事件时间线：重要事件的时序记录 """ import json import time import sqlite3 import hashlib import re from datetime import datetime, timedelta from pathlib import Path from typing import Optional, List, Dict, Any import logging logger = logging.getLogger(__name__) class ProfileManager: """ 人物档案管理器 数据结构： - profiles: 基础档案（姓名、头像、简介） - facts: 事实三元组（主体-谓词-客体） - relationships: 关系图谱 - events: 事件时间线 """ def __init__(self, db_path: str = None): self.db_path = db_path or str(Path.home() / ".hermes" / "workspace" / "livekit-agents" / "profiles.db") self._conn = None def initialize(self): """初始化数据库""" self._conn = sqlite3.connect(self.db_path) c = self._conn.cursor() # 人物基础档案 c.execute(''' CREATE TABLE IF NOT EXISTS profiles ( identity TEXT PRIMARY KEY, name TEXT, avatar_hash TEXT, bio TEXT, preferences TEXT, created_at REAL NOT NULL, updated_at REAL NOT NULL ) ''') # 事实三元组表 c.execute(''' CREATE TABLE IF NOT EXISTS facts ( id INTEGER PRIMARY KEY AUTOINCREMENT, subject_id TEXT NOT NULL, predicate TEXT NOT NULL, object TEXT NOT NULL, topic TEXT, time TEXT, source_dialogue_id INTEGER, confidence REAL DEFAULT 1.0, verified INTEGER DEFAULT 0, created_at REAL NOT NULL, expires_at REAL, FOREIGN KEY (subject_id) REFERENCES profiles(identity) ) ''') # 关系表 c.execute(''' CREATE TABLE IF NOT EXISTS relationships ( from_identity TEXT NOT NULL, to_identity TEXT NOT NULL, relation_type TEXT NOT NULL, strength REAL DEFAULT 0.5, note TEXT, source_dialogue_id INTEGER, created_at REAL NOT NULL, updated_at REAL NOT NULL, PRIMARY KEY (from_identity, to_identity) ) ''') # 关系变更历史（追加式） c.execute(''' CREATE TABLE IF NOT EXISTS relationship_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, from_identity TEXT NOT NULL, to_identity TEXT NOT NULL, relation_type TEXT, strength REAL, changed_at REAL NOT NULL, source_dialogue_id INTEGER, change_reason TEXT ) ''') c.execute('CREATE INDEX IF NOT EXISTS idx_rel_hist_pair ON relationship_history(from_identity, to_identity)') c.execute('CREATE INDEX IF NOT EXISTS idx_rel_hist_time ON relationship_history(changed_at)') # 通用变更日志表（六要素完整记录） c.execute(''' CREATE TABLE IF NOT EXISTS change_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, entity_type TEXT NOT NULL, -- profile/fact/relationship/event entity_id TEXT NOT NULL, field_name TEXT, -- 变更字段 -- 六要素 who TEXT, -- 谁引起的变更 what TEXT, -- 变更内容摘要 time_desc TEXT, -- 时间描述（绝对+相对） location TEXT, -- 地点 reason TEXT, -- 原因 trigger_method TEXT, -- 如何触发 old_value TEXT, -- JSON new_value TEXT, -- JSON changed_at REAL NOT NULL, source_dialogue_id INTEGER, scene_context TEXT, -- 场景描述（在做什么） trigger_dialogue TEXT, -- 触发变更的原始对话 metadata TEXT -- JSON: GPS/设备/参与者等辅助信息 ) ''') c.execute('CREATE INDEX IF NOT EXISTS idx_change_entity ON change_log(entity_type, entity_id)') c.execute('CREATE INDEX IF NOT EXISTS idx_change_time ON change_log(changed_at)') # 事件时间线 c.execute(''' CREATE TABLE IF NOT EXISTS events ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp REAL NOT NULL, participant_ids TEXT, event_type TEXT NOT NULL, description TEXT, scene_hash TEXT, importance REAL DEFAULT 0.5, metadata TEXT ) ''') # 创建索引 c.execute('CREATE INDEX IF NOT EXISTS idx_facts_subject ON facts(subject_id)') c.execute('CREATE INDEX IF NOT EXISTS idx_facts_predicate ON facts(predicate)') c.execute('CREATE INDEX IF NOT EXISTS idx_events_timestamp ON events(timestamp)') c.execute('CREATE INDEX IF NOT EXISTS idx_events_type ON events(event_type)') self._conn.commit() logger.info(f"📋 人物档案数据库: {self.db_path}") def register_profile(self, identity: str, name: str = None, bio: str = None, avatar_hash: str = None, preferences: Dict = None, scene_context: str = None, trigger_dialogue: str = None, who: str = None, what: str = None, time_desc: str = None, location: str = None, reason: str = None, trigger_method: str = None, metadata: Dict = None) -> bool: """注册新人物档案或更新现有档案（自动记录变更历史）""" if not self._conn: self.initialize() now = time.time() # 检查是否已存在 existing = self._conn.execute( 'SELECT identity, name, bio, avatar_hash, preferences FROM profiles WHERE identity = ?', (identity,) ).fetchone() # 构建新数据 new_data = {'name': name, 'bio': bio, 'avatar_hash': avatar_hash, 'preferences': json.dumps(preferences) if preferences else None} if existing: # 记录变更历史 old_data = {'name': existing[1], 'bio': existing[2], 'avatar_hash': existing[3], 'preferences': existing[4]} for field in ['name', 'bio', 'avatar_hash']: if new_data[field] is not None and str(new_data[field]) != str(old_data[field]): self._log_change( entity_type='profile', entity_id=identity, field_name=field, old_value=json.dumps(old_data[field]), new_value=json.dumps(new_data[field]), scene_context=scene_context, trigger_dialogue=trigger_dialogue, who=who, what=what, time_desc=time_desc, location=location, reason=reason, trigger_method=trigger_method, metadata=metadata ) if new_data['preferences'] is not None: old_prefs = old_data['preferences'] if str(new_data['preferences']) != str(old_prefs): self._log_change( entity_type='profile', entity_id=identity, field_name='preferences', old_value=old_prefs or 'null', new_value=new_data['preferences'], scene_context=scene_context, trigger_dialogue=trigger_dialogue, who=who, what=what, time_desc=time_desc, location=location, reason=reason, trigger_method=trigger_method, metadata=metadata ) logger.info(f"🔄 更新档案: {identity}") else: # 首次注册 - 记录创建日志 self._log_change( entity_type='profile', entity_id=identity, field_name='create', old_value=None, new_value=json.dumps({'name': name, 'bio': bio}), scene_context=scene_context, trigger_dialogue=trigger_dialogue, who=who, what=what, time_desc=time_desc, location=location, reason=reason, trigger_method=trigger_method, metadata=metadata ) # 插入 self._conn.execute(''' INSERT INTO profiles (identity, name, bio, avatar_hash, preferences, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?) ''', (identity, name, bio, avatar_hash, new_data['preferences'], now, now)) logger.info(f"👤 注册新档案: {identity} ({name})") self._conn.commit() return True def add_fact(self, subject_id: str, predicate: str, object: str, source_dialogue_id: int = None, confidence: float = 1.0, expires_at: float = None, topic: str = None, fact_time: str = None, source_type: str = None, source_id: str = None, who: str = None, what: str = None, time_desc: str = None, location: str = None, reason: str = None, trigger_method: str = None, scene_context: str = None, trigger_dialogue: str = None, metadata: Dict = None) -> int: """ 添加事实（追加模式：自动过期同predicate旧事实） Args: subject_id: 主体身份 ID predicate: 谓词（如 "favorite_color", "works_at"） object: 客体值 source_dialogue_id: 来源对话 ID confidence: 可信度 0-1 expires_at: 手动指定过期时间（可选，通常由系统自动设置） topic: 主题分类 fact_time: 时间维度 source_type: 数据来源类型（dialogue/photo/video/file） source_id: 来源唯一ID Returns: fact_id """ if not self._conn: self.initialize() now = time.time() # 追加模式：如有同(subject_id, predicate)的未过期记录，标记为过期 old_fact = None if expires_at is None: old_fact = self._conn.execute( "SELECT id, object FROM facts WHERE subject_id = ? AND predicate = ? AND expires_at IS NULL", (subject_id, predicate) ).fetchone() if old_fact: self._conn.execute( "UPDATE facts SET expires_at = ? WHERE id = ?", (now, old_fact[0]) ) cursor = self._conn.execute(''' INSERT INTO facts (subject_id, predicate, object, topic, time, source_dialogue_id, confidence, created_at, expires_at, source_type, source_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', (subject_id, predicate, object, topic, normalize_time(fact_time), source_dialogue_id, confidence, now, expires_at, source_type, source_id)) self._conn.commit() fact_id = cursor.lastrowid logger.info(f"📝 添加事实: {subject_id} -> {predicate} -> {object} (ID: {fact_id})") # 记录变更日志（首次添加或旧值被替换） self._log_change( entity_type='fact', entity_id=f'{subject_id}:{predicate}', field_name='object', old_value=json.dumps(old_fact[1]) if old_fact else None, new_value=json.dumps(object), source_dialogue_id=source_dialogue_id, scene_context=scene_context, trigger_dialogue=trigger_dialogue, who=who, what=what, time_desc=time_desc, location=location, reason=reason, trigger_method=trigger_method, metadata=metadata ) return fact_id def update_relationship(self, from_id: str, to_id: str, relation_type: str, strength: float = 0.5, note: str = None, source_dialogue_id: int = None, change_reason: str = None, who: str = None, what: str = None, time_desc: str = None, location: str = None, reason: str = None, trigger_method: str = None, scene_context: str = None, trigger_dialogue: str = None, metadata: Dict = None): """ 更新关系并记录历史 Args: from_id: 关系发起方 to_id: 关系目标方 relation_type: 关系类型 (colleague/friend/family/...) strength: 关系强度 0-1 note: 备注 source_dialogue_id: 来源对话ID change_reason: 变更原因（由LLM提取，用于历史记录） """ if not self._conn: self.initialize() now = time.time() # 1. 获取旧状态（用于历史记录） old_row = self._conn.execute( 'SELECT relation_type, strength FROM relationships WHERE from_identity=? AND to_identity=?', (from_id, to_id) ).fetchone() # 2. UPSERT 当前关系 self._conn.execute(''' INSERT INTO relationships (from_identity, to_identity, relation_type, strength, note, source_dialogue_id, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(from_identity, to_identity) DO UPDATE SET relation_type = excluded.relation_type, strength = excluded.strength, note = COALESCE(excluded.note, relationships.note), updated_at = excluded.updated_at ''', (from_id, to_id, relation_type, strength, note, source_dialogue_id, now, now)) # 3. 写入历史记录 self._conn.execute(''' INSERT INTO relationship_history (from_identity, to_identity, relation_type, strength, changed_at, source_dialogue_id, change_reason) VALUES (?, ?, ?, ?, ?, ?, ?) ''', (from_id, to_id, relation_type, strength, now, source_dialogue_id, change_reason)) # 4. 写入通用变更日志 old_info = {'relation_type': old_row[0], 'strength': old_row[1]} if old_row else None new_info = {'relation_type': relation_type, 'strength': strength} if old_info: # 关系变更 self._log_change( entity_type='relationship', entity_id=f'{from_id}→{to_id}', field_name='relationship', old_value=json.dumps(old_info), new_value=json.dumps(new_info), source_dialogue_id=source_dialogue_id, scene_context=scene_context, trigger_dialogue=trigger_dialogue, who=who, what=what, time_desc=time_desc, location=location, reason=reason, trigger_method=trigger_method, metadata=metadata ) else: # 首次建立关系 self._log_change( entity_type='relationship', entity_id=f'{from_id}→{to_id}', field_name='relationship', old_value=None, new_value=json.dumps(new_info), source_dialogue_id=source_dialogue_id, scene_context=scene_context, trigger_dialogue=trigger_dialogue, who=who, what=what, time_desc=time_desc, location=location, reason=reason, trigger_method=trigger_method, metadata=metadata ) self._conn.commit() old_info = f" ({old_row[0]}, {old_row[1]})" if old_row else "" logger.info(f"🔗 更新关系: {from_id} -[{relation_type}:{strength}]{old_info}-> {to_id}") def get_relationship_history(self, from_id: str = None, to_id: str = None, limit: int = 50) -> List[Dict]: """查询关系变更历史""" if not self._conn: self.initialize() query = 'SELECT * FROM relationship_history WHERE 1=1' params = [] if from_id: query += ' AND from_identity = ?' params.append(from_id) if to_id: query += ' AND to_identity = ?' params.append(to_id) query += ' ORDER BY changed_at DESC LIMIT ?' params.append(limit) rows = self._conn.execute(query, params).fetchall() return [{ 'id': r[0], 'from_identity': r[1], 'to_identity': r[2], 'relation_type': r[3], 'strength': r[4], 'changed_at': r[5], 'source_dialogue_id': r[6], 'change_reason': r[7] } for r in rows] def _log_change(self, entity_type: str, entity_id: str, field_name: str = None, old_value: str = None, new_value: str = None, source_dialogue_id: int = None, scene_context: str = None, trigger_dialogue: str = None, who: str = None, what: str = None, time_desc: str = None, location: str = None, reason: str = None, trigger_method: str = None, metadata: Dict = None): """记录变更日志（六要素完整上下文）""" now = time.time() self._conn.execute(''' INSERT INTO change_log (entity_type, entity_id, field_name, who, what, time_desc, location, reason, trigger_method, old_value, new_value, changed_at, source_dialogue_id, scene_context, trigger_dialogue, metadata) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', (entity_type, entity_id, field_name, who, what, time_desc, location, reason, trigger_method, old_value, new_value, now, source_dialogue_id, scene_context, trigger_dialogue, json.dumps(metadata) if metadata else None)) logger.info(f"📝 变更记录: {entity_type}:{entity_id}.{field_name}") def get_change_history(self, entity_type: str = None, entity_id: str = None, limit: int = 50) -> List[Dict]: """查询变更历史""" if not self._conn: self.initialize() query = 'SELECT * FROM change_log WHERE 1=1' params = [] if entity_type: query += ' AND entity_type = ?' params.append(entity_type) if entity_id: query += ' AND entity_id = ?' params.append(entity_id) query += ' ORDER BY changed_at DESC LIMIT ?' params.append(limit) rows = self._conn.execute(query, params).fetchall() return [{ 'id': r[0], 'entity_type': r[1], 'entity_id': r[2], 'field_name': r[3], 'who': r[4], 'what': r[5], 'time_desc': r[6], 'location': r[7], 'reason': r[8], 'trigger_method': r[9], 'old_value': r[10], 'new_value': r[11], 'changed_at': r[12], 'source_dialogue_id': r[13], 'scene_context': r[14], 'trigger_dialogue': r[15], 'metadata': r[16] } for r in rows] def add_event(self, timestamp: float, participant_ids: List[str], event_type: str, description: str = None, scene_hash: str = None, importance: float = 0.5, metadata: Dict = None, who: str = None, what: str = None, time_desc: str = None, location: str = None, reason: str = None, trigger_method: str = None, scene_context: str = None, trigger_dialogue: str = None): """添加事件到时间线""" if not self._conn: self.initialize() cursor = self._conn.execute(''' INSERT INTO events (timestamp, participant_ids, event_type, description, scene_hash, importance, metadata) VALUES (?, ?, ?, ?, ?, ?, ?) ''', (timestamp, json.dumps(participant_ids), event_type, description, scene_hash, importance, json.dumps(metadata) if metadata else None)) self._conn.commit() event_id = cursor.lastrowid logger.info(f"📅 添加事件: {event_type} at {timestamp}") # 记录变更日志 self._log_change( entity_type='event', entity_id=str(event_id), field_name='event', old_value=None, new_value=json.dumps({'event_type': event_type, 'description': description, 'participants': participant_ids}), scene_context=scene_context, trigger_dialogue=trigger_dialogue, who=who, what=what, time_desc=time_desc, location=location, reason=reason, trigger_method=trigger_method ) # ========== 查询方法 ========== def get_profile(self, identity: str) -> Optional[Dict]: """获取人物档案""" if not self._conn: self.initialize() row = self._conn.execute( 'SELECT * FROM profiles WHERE identity = ?', (identity,) ).fetchone() if row: return { 'identity': row[0], 'name': row[1], 'avatar_hash': row[2], 'bio': row[3], 'preferences': json.loads(row[4]) if row[4] else {}, 'created_at': row[5], 'updated_at': row[6] } return None def get_facts(self, subject_id: str = None, predicate: str = None, topic: str = None, limit: int = 100, include_expired: bool = False) -> List[Dict]: """查询事实列表 Args: subject_id: 过滤主体 predicate: 模糊匹配谓词 topic: 主题分类过滤 limit: 最多返回条数 include_expired: 是否包含已过期的事实（默认只返回当前有效事实） """ if not self._conn: self.initialize() query = 'SELECT subject_id, predicate, object, topic, time, confidence, created_at, expires_at FROM facts WHERE 1=1' params = [] if subject_id: query += ' AND subject_id = ?' params.append(subject_id) if predicate: query += ' AND predicate LIKE ?' params.append(f'%{predicate}%') if topic: query += ' AND topic = ?' params.append(topic) # 默认只显示当前有效事实（未过期） if not include_expired: query += ' AND (expires_at IS NULL OR expires_at > ?)' params.append(time.time()) query += ' ORDER BY created_at DESC LIMIT ?' params.append(limit) rows = self._conn.execute(query, params).fetchall() return [{'subject': r[0], 'predicate': r[1], 'object': r[2], 'topic': r[3], 'time': r[4], 'confidence': r[5], 'created_at': r[6], 'expires_at': r[7]} for r in rows] def get_relationships(self, identity: str, direction: str = 'both') -> List[Dict]: """ 查询关系 direction: 'outgoing' (我发出的) | 'incoming' (我收到的) | 'both' (双向) """ if not self._conn: self.initialize() if direction == 'outgoing': rows = self._conn.execute( 'SELECT * FROM relationships WHERE from_identity = ?', (identity,) ).fetchall() elif direction == 'incoming': rows = self._conn.execute( 'SELECT * FROM relationships WHERE to_identity = ?', (identity,) ).fetchall() else: rows = self._conn.execute( 'SELECT * FROM relationships WHERE from_identity = ? OR to_identity = ?', (identity, identity) ).fetchall() return [{ 'from': r[0], 'to': r[1], 'type': r[2], 'strength': r[3], 'note': r[4], 'created_at': r[6], 'updated_at': r[7] } for r in rows] def get_events(self, participant_id: str = None, event_type: str = None, last_hours: int = None, limit: int = 20) -> List[Dict]: """查询事件时间线""" if not self._conn: self.initialize() query = 'SELECT * FROM events WHERE 1=1' params = [] if last_hours is not None: cutoff = time.time() - (last_hours * 3600) query += ' AND timestamp > ?' params.append(cutoff) if participant_id: query += ' AND participant_ids LIKE ?' params.append(f'%"{participant_id}"%') if event_type: query += ' AND event_type = ?' params.append(event_type) query += ' ORDER BY timestamp DESC LIMIT ?' params.append(limit) rows = self._conn.execute(query, params).fetchall() return [{ 'id': r[0], 'timestamp': r[1], 'participants': json.loads(r[2]), 'type': r[3], 'description': r[4], 'importance': r[6] } for r in rows] def search_profiles(self, query: str, limit: int = 10) -> List[Dict]: """模糊搜索人物档案""" if not self._conn: self.initialize() rows = self._conn.execute(''' SELECT identity, name, bio FROM profiles WHERE name LIKE ? OR bio LIKE ? OR identity LIKE ? LIMIT ? ''', (f'%{query}%', f'%{query}%', f'%{query}%', limit)).fetchall() return [{'identity': r[0], 'name': r[1], 'bio': r[2]} for r in rows] def search_facts(self, query: str, limit: int = 20) -> List[Dict]: """搜索事实""" if not self._conn: self.initialize() rows = self._conn.execute(''' SELECT subject_id, predicate, object, confidence FROM facts WHERE object LIKE ? OR predicate LIKE ? LIMIT ? ''', (f'%{query}%', f'%{query}%', limit)).fetchall() return [{'subject': r[0], 'predicate': r[1], 'object': r[2], 'confidence': r[3]} for r in rows] def get_summary(self, identity: str, hours: int = 720) -> Dict: """获取人物完整画像""" profile = self.get_profile(identity) facts = self.get_facts(subject_id=identity) relations = self.get_relationships(identity) events = self.get_events(participant_id=identity, last_hours=hours) # 获取变更历史 history = self.get_change_history(entity_id=identity, limit=20) return { 'profile': profile, 'facts_count': len(facts), 'relations_count': len(relations), 'recent_events': events[:10], 'recent_changes': history[:10] } def close(self): """关闭数据库""" if self._conn: self._conn.close() self._conn = None logger.info(f"📋 关闭人物档案数据库") def get_context(self, identity: str, hours: int = 24) -> Dict: """获取人物最近时间窗口内的完整上下文（聚合查询）""" if not self._conn: self.initialize() # 1. 获取当前状态 profile = self.get_profile(identity) facts = self.get_facts(subject_id=identity) relations = self.get_relationships(identity) # 2. 获取近期事件 events = self.get_events(participant_id=identity, last_hours=hours) if hours else [] # 3. 获取近期变更历史 history = self.get_change_history(entity_id=identity, limit=50) # 4. 按时间排序合并事件和变更记录 timeline = [] for e in events: timeline.append({ 'type': 'event', 'time': e['timestamp'], 'content': e }) for h in history: timeline.append({ 'type': 'change', 'time': h['changed_at'], 'content': h }) # 按时间倒序 timeline.sort(key=lambda x: x['time'], reverse=True) return { 'identity': identity, 'profile': profile, 'current_facts': facts, 'current_relations': relations, 'timeline': timeline[:20], # 最近20条 'total_changes': len(history) } def detect_conflicts(self, identity: str) -> List[Dict]: """检测矛盾事实（同一 predicate 有多个不同 object）""" if not self._conn: self.initialize() # 获取所有事实（包括过期的） all_facts = self.get_facts(subject_id=identity, include_expired=True) # 按 predicate 分组 grouped = {} for f in all_facts: pred = f['predicate'] if pred not in grouped: grouped[pred] = [] grouped[pred].append(f) conflicts = [] for pred, facts_list in grouped.items(): # 收集所有不同的 object 值 objects = set() for f in facts_list: if f['object']: objects.add(f['object']) if len(objects) > 1: conflicts.append({ 'predicate': pred, 'conflicting_values': list(objects), 'count': len(facts_list), 'most_recent': max(facts_list, key=lambda x: x['created_at']), 'expired': [f for f in facts_list if f['expires_at']] }) return conflicts def extract_facts_from_dialogue(self, dialogue_text: str, speaker_id: str, llm_client=None) -> List[Dict]: """从对话文本中提取事实（对接 Hermes LLM API） Args: dialogue_text: 对话文本 speaker_id: 说话人 ID llm_client: LLM 客户端（可选） Returns: 提取的事实列表 [{'predicate': str, 'object': str, 'topic': str, 'confidence': float}] """ # 尝试使用 Agnes AI API try: import os import openai from dotenv import load_dotenv load_dotenv('/Users/leo/.hermes/workspace/livekit-agents/.env.dev') api_key = os.getenv('OPENAI_API_KEY') base_url = os.getenv('OPENAI_BASE_URL', 'https://api.agnes-ai.cn/v1') model = os.getenv('LLM_MODEL', 'agnes-2.5-flash') if not api_key or api_key == '[REDACTED]': logger.warning("LLM API Key not configured, using fallback") return self._extract_facts_fallback(dialogue_text, speaker_id) client = openai.OpenAI(api_key=api_key, base_url=base_url) prompt = f"""你是一个信息抽取助手。从以下对话中提取关于"{speaker_id}"的事实信息。 只返回 JSON 格式的数组，不要其他内容。 示例输出格式： [{{"predicate": "favorite_color", "object": "蓝色", "topic": "hobby", "confidence": 0.9}}] 对话内容：{dialogue_text} 只输出 JSON 数组，如果没有事实则返回 [] """ response = client.chat.completions.create( model=model, messages=[{"role": "user", "content": prompt}], temperature=0.1 ) result_text = (response.choices[0].message.content or '').strip() # 解析 JSON if result_text.startswith('['): facts = json.loads(result_text) logger.info(f"✅ LLM 提取 {len(facts)} 条事实: {facts}") return facts else: # 尝试提取 JSON 部分 import re json_match = re.search(r'\\[.*?\\]', result_text, re.DOTALL) if json_match: facts = json.loads(json_match.group()) logger.info(f"✅ LLM 提取 {len(facts)} 条事实: {facts}") return facts except Exception as e: logger.warning(f"LLM 提取失败: {e}，使用 fallback 方法") # Fallback: 简单规则提取 return self._extract_facts_fallback(dialogue_text, speaker_id) def _extract_facts_fallback(self, dialogue_text: str, speaker_id: str) -> List[Dict]: """基于简单规则的提取（LLM 不可用时使用）""" facts = [] # 简单的关键词匹配 patterns = [ (r'我叫(\w+)', 'name'), (r'我喜欢(\w+)', 'favorite_color' if '颜色' in dialogue_text else 'interest'), (r'我在(.+?)(工作|公司)', 'works_at'), (r'我是(.+?)(程序员|工程师|设计师)', 'occupation'), ] for pattern, predicate in patterns: matches = re.findall(pattern, dialogue_text) for match in matches: facts.append({ 'predicate': predicate, 'object': match if isinstance(match, str) else match[0], 'topic': 'other', 'confidence': 0.5 }) return facts def infer_relationship_strength(self, from_id: str, to_id: str, interaction_count: int = None, interaction_depth: float = 1.0) -> float: """基于交互频率推断关系强度 Args: from_id: 发起方 ID to_id: 目标方 ID interaction_count: 交互次数（可选，默认查数据库统计） interaction_depth: 交互深度权重（0-1） Returns: 推断的关系强度（0-1） """ if not self._conn: self.initialize() # 统计交互次数（从 events 和 change_log） if interaction_count is None: # 从 events 统计共同事件数 count_query = ''' SELECT COUNT(*) FROM events WHERE participant_ids LIKE ? AND participant_ids LIKE ? ''' result = self._conn.execute(count_query, (f'%{from_id}%', f'%{to_id}%')).fetchone() interaction_count = result[0] if result else 0 # 加上变更记录中的互动 change_query = ''' SELECT COUNT(*) FROM change_log WHERE (entity_type = 'relationship' AND entity_id LIKE ?) OR (entity_type = 'event' AND metadata LIKE ?) ''' result = self._conn.execute(change_query, (f'%{from_id}→{to_id}%', f'%{from_id}%')).fetchone() interaction_count += result[0] if result else 0 # 基础公式：强度 = min(0.95, 0.3 + 0.1 * log1p(交互次数) * depth) import math base_strength = 0.3 + 0.1 * math.log1p(interaction_count) * interaction_depth inferred_strength = min(0.95, max(0.1, base_strength)) return round(inferred_strength, 2) # =============== 时间标准化 =============== _TIME_PATTERNS = [ # 月级相对时间 → YYYY-MM（具体优先于模糊） (r'(\d+)月(\d+)号?', lambda m, now: f"{now.year}-{int(m.group(1)):02d}-{int(m.group(2)):02d}"), (r'(\d+)月(\d+)日', lambda m, now: f"{now.year}-{int(m.group(1)):02d}-{int(m.group(2)):02d}"), (r'今年(\d+)月', lambda m, now: f"{now.year}-{int(m.group(1)):02d}"), (r'(明年|去年)(\d+)月', lambda m, now: f"{(now.year + (1 if m.group(1) == '明年' else -1))}-{int(m.group(2)):02d}"), (r'上个月', lambda m, now: (now.replace(day=1) - timedelta(days=1)).strftime('%Y-%m')), (r'下个月', lambda m, now: (now.replace(day=28) + timedelta(days=4)).strftime('%Y-%m')), (r'这个月', lambda m, now: now.strftime('%Y-%m')), (r'本月', lambda m, now: now.strftime('%Y-%m')), (r'(\d+)个月后', lambda m, now: (now + timedelta(days=int(m.group(1)) * 30)).strftime('%Y-%m')), (r'(\d+)个月前', lambda m, now: (now - timedelta(days=int(m.group(1)) * 30)).strftime('%Y-%m')), # 日期级相对时间 → YYYY-MM-DD (r'上周', lambda m, now: (now - timedelta(days=now.weekday() + 7)).strftime('%Y-%m-%d')), (r'这周', lambda m, now: now.strftime('%Y-%m-%d')), (r'下周', lambda m, now: (now + timedelta(days=7 - now.weekday())).strftime('%Y-%m-%d')), (r'(\d+)周前', lambda m, now: (now - timedelta(weeks=int(m.group(1)))).strftime('%Y-%m-%d')), (r'(\d+)周后', lambda m, now: (now + timedelta(weeks=int(m.group(1)))).strftime('%Y-%m-%d')), (r'(\d+)天前', lambda m, now: (now - timedelta(days=int(m.group(1)))).strftime('%Y-%m-%d')), (r'(\d+)天后', lambda m, now: (now + timedelta(days=int(m.group(1)))).strftime('%Y-%m-%d')), (r'上(周一|周二|周三|周四|周五|周六|周日)', lambda m, now: (now - timedelta(days=(now.weekday() - ['一','二','三','四','五','六','日'].index(m.group(1)) + 7) % 7)).strftime('%Y-%m-%d')), (r'这(周一|周二|周三|周四|周五|周六|周日)', lambda m, now: (now - timedelta(days=now.weekday() % 7)).strftime('%Y-%m-%d')), (r'下(周一|周二|周三|周四|周五|周六|周日)', lambda m, now: (now + timedelta(days=(7 - now.weekday() + ['一','二','三','四','五','六','日'].index(m.group(1))) % 7)).strftime('%Y-%m-%d')), # 年级相对时间 → YYYY（最通用，放最后） (r'去年', lambda m, now: str(now.year - 1)), (r'今年', lambda m, now: str(now.year)), (r'明年', lambda m, now: str(now.year + 1)), (r'(\d+)年前', lambda m, now: str(now.year - int(m.group(1)))), ] def normalize_time(time_str: str, now: datetime = None) -> str: """将相对时间转换为绝对日期 YYYY-MM-DD 或 YYYY-MM""" if not time_str or time_str in ('null', 'None', ''): return None now = now or datetime.now() original = time_str # 先尝试直接解析 for fmt in ('%Y-%m-%d', '%Y-%m', '%Y'): try: datetime.strptime(time_str, fmt) return time_str except ValueError: pass # 尝试匹配相对时间模式 for pattern, func in _TIME_PATTERNS: match = re.search(pattern, time_str) if match: try: return func(match, now) except Exception: pass # 兜底：如果无法解析，返回原始字符串 return original def extract_fact_prompt(text: str) -> str: """构建事实提取提示词（含主题分类）""" return f'''分析以下对话，提取用户的关键信息作为事实三元组，并为每条事实分类主题。 对话内容：{text} 请提取以下类型的事实（如果存在）： - works_at: 工作公司 - education: 教育背景 - favorite_color/hobby/food: 喜好 - location: 所在地 - family: 家庭关系 - role: 职位角色 - skill: 技能专长 - project: 正在进行的项目 - purchase: 购买/拥有物品 - reminder: 需要提醒的重要日期 - travel: 旅行经历、目的地、去过哪里 每条事实必须包含 topic 字段，从以下主题中选择： profile（基础档案）、work（工作）、study（学习）、hobby（爱好）、 family（家庭）、location（地点）、travel（旅行）、event（事件）、 reminder（提醒）、skill（技能）、project（项目）、purchase（消费）、other（其他） 返回 JSON 格式（如无事实返回空数组 []）： [{"predicate": "works_at", "object": "谷歌", "topic": "work", "time": null, "confidence": 0.9}, {"predicate": "wedding_date", "object": "下个月", "topic": "reminder", "time": "2024-10-15", "confidence": 0.95}] 只输出 JSON，不要解释。''' async def _extract_and_store_facts(llm_client, profile_manager, speaker_id: str, text: str) -> List[Dict]: """ 从对话文本中自动提取事实并存储（含主题分类） Args: llm_client: LLM 客户端 profile_manager: ProfileManager 实例 speaker_id: 说话人 ID text: 对话文本 Returns: 提取的事实列表 """ try: prompt = extract_fact_prompt(text) # 调用 LLM 提取事实 response = await llm_client.chat([ {"role": "user", "content": prompt} ]) # 解析 JSON 响应 import re json_match = re.search(r'\[.*\]', response, re.DOTALL) if json_match: facts = json.loads(json_match.group()) # 存储事实 stored = [] for fact in facts: if isinstance(fact, dict) and 'predicate' in fact and 'object' in fact: profile_manager.add_fact( subject_id=speaker_id, predicate=fact['predicate'], object=fact['object'], topic=fact.get('topic', 'other'), confidence=fact.get('confidence', 0.8) ) stored.append(fact) logger.info(f"📝 自动提取事实: {speaker_id} [{fact.get('topic', 'other')}] -> {fact['predicate']} -> {fact['object']}") return stored except Exception as e: logger.warning(f"⚠️ 事实提取失败: {e}") return [] # =============== 使用示例 =============== def demo(): """演示用法""" import tempfile import os db_path = tempfile.mktemp(suffix='.db') pm = ProfileManager(db_path) pm.initialize() # 1. 注册人物 print("=== 1. 注册人物档案 ===") pm.register_profile('zhangsan', name='张三', bio='产品经理，喜欢运动和科技') pm.register_profile('lisi', name='李四', bio='工程师，热爱编程和开源') pm.register_profile('wangwu', name='王五', bio='设计师，关注用户体验') # 2. 添加事实 print("\n=== 2. 添加事实 ===") pm.add_fact('zhangsan', 'works_at', '某科技公司', source_dialogue_id=1) pm.add_fact('zhangsan', 'favorite_color', '蓝色', source_dialogue_id=1) pm.add_fact('zhangsan', 'hates', '早起开会', source_dialogue_id=2) pm.add_fact('lisi', 'works_at', '某科技公司', source_dialogue_id=1) pm.add_fact('lisi', 'expertise', '后端开发', source_dialogue_id=3) # 3. 建立关系 print("\n=== 3. 建立关系 ===") pm.update_relationship('zhangsan', 'lisi', 'colleague', strength=0.8, note='项目组同事') pm.update_relationship('zhangsan', 'wangwu', 'colleague', strength=0.6, note='产品&设计协作') pm.update_relationship('lisi', 'wangwu', 'friend', strength=0.9, note='好朋友') # 4. 添加事件 print("\n=== 4. 添加事件 ===") pm.add_event(time.time(), ['zhangsan', 'lisi'], 'meeting', description='项目评审会议', importance=0.7) pm.add_event(time.time(), ['zhangsan'], 'social', description='午餐聊天', importance=0.3) # 5. 查询 print("\n=== 5. 查询 ===") # 获取人物摘要 summary = pm.get_summary('zhangsan') print(f"张三档案:") print(f" - 名字: {summary['profile']['name']}") print(f" - 简介: {summary['profile']['bio']}") print(f" - 事实数: {summary['facts_count']}") print(f" - 关系数: {summary['relations_count']}") # 获取事实 facts = pm.get_facts(subject_id='zhangsan') print(f"\n张三的事实:") for f in facts: print(f" - {f['predicate']}: {f['object']}") # 获取关系 relations = pm.get_relationships('zhangsan') print(f"\n张三的关系:") for r in relations: direction = '→' if r['from'] == 'zhangsan' else '←' print(f" {direction} {r['to']}: {r['type']} (强度: {r['strength']:.1f})") # 搜索 print("\n=== 6. 搜索 ===") results = pm.search_profiles('科技') print(f"搜索'科技': {[r['name'] for r in results]}") pm.close() os.unlink(db_path) print("\n✅ 演示完成") if __name__ == '__main__': demo()