refactor: restructure to core/ + ui/ with multi-threaded backend

This commit is contained in:
root
2026-04-12 18:15:55 +08:00
parent e43a88a7d0
commit c8a69b3926
64 changed files with 5269 additions and 571 deletions

5
core/__init__.py Normal file
View File

@ -0,0 +1,5 @@
from .server import BackendServer
from .client import BackendClient
from .embedded_db import EmbeddedGraphDB
__all__ = ["BackendServer", "BackendClient", "EmbeddedGraphDB"]

27
core/client.py Normal file
View File

@ -0,0 +1,27 @@
import threading
from typing import Any, Dict
from .server import BackendServer
class BackendClient:
def __init__(self, server: BackendServer):
self._server = server
self._request_counter = 0
self._lock = threading.Lock()
def process_message(self, user_input: str, timeout: float = 30.0) -> Dict[str, Any]:
with self._lock:
self._request_counter += 1
return self._server.process_message(user_input, timeout)
def execute_tool(self, tool_name: str, arguments: Dict[str, Any], timeout: float = 10.0) -> str:
with self._lock:
self._request_counter += 1
return self._server.execute_tool(tool_name, arguments, timeout)
def update_config(self, api_key: str, base_url: str = "https://api.deepseek.com") -> None:
self._server.update_config(api_key, base_url)
def shutdown(self) -> None:
self._server.shutdown()

438
core/embedded_db.py Normal file
View File

@ -0,0 +1,438 @@
"""
内嵌图数据库 - 基于SQLite实现
无需Docker,开箱即用
"""
import sqlite3
import json
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Optional, Any
class EmbeddedGraphDB:
"""内嵌图数据库 - SQLite实现"""
def __init__(self, db_path: str = "graph_memory.db"):
"""
初始化数据库
Args:
db_path: 数据库文件路径
"""
self.db_path = Path(db_path)
self.conn = None
self._init_db()
def _init_db(self):
"""初始化数据库表"""
self.conn = sqlite3.connect(str(self.db_path), check_same_thread=False)
self.conn.row_factory = sqlite3.Row
cursor = self.conn.cursor()
# 创建实体表
cursor.execute("""
CREATE TABLE IF NOT EXISTS entities (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE NOT NULL,
type TEXT,
mention_count INTEGER DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# 创建关系表
cursor.execute("""
CREATE TABLE IF NOT EXISTS relations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source_id INTEGER NOT NULL,
target_id INTEGER NOT NULL,
relation_type TEXT NOT NULL,
confidence REAL DEFAULT 1.0,
status TEXT DEFAULT 'active',
session_id TEXT,
turn_id INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
date_bucket TEXT,
superseded_by INTEGER,
FOREIGN KEY (source_id) REFERENCES entities(id),
FOREIGN KEY (target_id) REFERENCES entities(id)
)
""")
# 创建索引
cursor.execute("CREATE INDEX IF NOT EXISTS idx_entity_name ON entities(name)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_entity_type ON entities(type)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_relation_source ON relations(source_id)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_relation_target ON relations(target_id)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_relation_type ON relations(relation_type)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_relation_status ON relations(status)")
self.conn.commit()
def ensure_constraints(self):
"""确保约束(兼容Neo4j接口)"""
pass # SQLite自动处理
def recall(self, query_intent: str, seed_entities: List[str] = None,
depth: int = 2, time_range: Dict = None,
session_filter: str = None) -> Dict:
"""
检索相关记忆
Args:
query_intent: 查询关键词(逗号分隔)
seed_entities: 种子实体
depth: 搜索深度
time_range: 时间范围
session_filter: 会话过滤
Returns:
检索结果
"""
keywords = [w.strip().lower() for w in query_intent.replace(',', ' ').split() if w.strip()]
cursor = self.conn.cursor()
# 搜索实体
entities = []
entity_ids = set()
# 如果没有关键词,返回所有实体(用于"我们都聊过什么"这类问题)
if not keywords and not seed_entities:
cursor.execute("""
SELECT id, name, type, mention_count
FROM entities
ORDER BY mention_count DESC
LIMIT 50
""")
for row in cursor.fetchall():
entity_ids.add(row['id'])
entities.append({
'name': row['name'],
'type': row['type'] or 'unknown',
'mention_count': row['mention_count']
})
else:
# 有关键词,按关键词搜索
for keyword in keywords:
cursor.execute("""
SELECT id, name, type, mention_count
FROM entities
WHERE LOWER(name) LIKE ?
""", (f"%{keyword}%",))
for row in cursor.fetchall():
if row['id'] not in entity_ids:
entity_ids.add(row['id'])
entities.append({
'name': row['name'],
'type': row['type'] or 'unknown',
'mention_count': row['mention_count']
})
# 搜索关系
relations = []
if entity_ids:
placeholders = ','.join('?' * len(entity_ids))
query = f"""
SELECT r.id, e1.name as source, e2.name as target,
r.relation_type as type, r.confidence, r.session_id,
r.turn_id, r.created_at, r.status
FROM relations r
JOIN entities e1 ON r.source_id = e1.id
JOIN entities e2 ON r.target_id = e2.id
WHERE (r.source_id IN ({placeholders}) OR r.target_id IN ({placeholders}))
AND r.status = 'active'
"""
params = list(entity_ids) + list(entity_ids)
if session_filter:
query += " AND r.session_id = ?"
params.append(session_filter)
cursor.execute(query, params)
for row in cursor.fetchall():
relations.append({
'source': row['source'],
'target': row['target'],
'type': row['type'],
'confidence': row['confidence'],
'session_id': row['session_id'],
'turn_id': row['turn_id'],
'created_at': row['created_at'],
'status': row['status']
})
return {
"entities": entities,
"relations": relations,
"message": f"找到 {len(entities)} 个实体, {len(relations)} 条关系"
}
def commit(self, triplets: List[Dict], entity_types: Dict = None,
temporal_tag: str = None, session_id: str = None,
turn_id: int = None) -> Dict:
"""
写入记忆
Args:
triplets: 三元组列表
entity_types: 实体类型
temporal_tag: 时间标签
session_id: 会话ID
turn_id: 轮次ID
Returns:
写入结果
"""
cursor = self.conn.cursor()
created_entities = 0
created_relations = 0
for triplet in triplets:
subject = triplet.get('subject')
relation = triplet.get('relation')
obj = triplet.get('object')
confidence = triplet.get('confidence', 1.0)
if not all([subject, relation, obj]):
continue
# 创建或更新实体
for entity_name in [subject, obj]:
entity_type = entity_types.get(entity_name) if entity_types else None
cursor.execute("""
INSERT INTO entities (name, type)
VALUES (?, ?)
ON CONFLICT(name) DO UPDATE SET
mention_count = mention_count + 1,
updated_at = CURRENT_TIMESTAMP
""", (entity_name, entity_type))
if cursor.rowcount > 0:
created_entities += 1
# 获取实体ID
cursor.execute("SELECT id FROM entities WHERE name = ?", (subject,))
source_id = cursor.fetchone()['id']
cursor.execute("SELECT id FROM entities WHERE name = ?", (obj,))
target_id = cursor.fetchone()['id']
# 创建关系
date_bucket = datetime.now().strftime('%Y-%m-%d')
cursor.execute("""
INSERT INTO relations (
source_id, target_id, relation_type, confidence,
session_id, turn_id, date_bucket
)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (source_id, target_id, relation, confidence,
session_id, turn_id, date_bucket))
created_relations += 1
self.conn.commit()
return {
"created_entities": created_entities,
"created_relations": created_relations,
"message": f"创建了 {created_entities} 个实体, {created_relations} 条关系"
}
def purge(self, criteria: Dict, mode: str = "soft",
new_relation: Dict = None) -> Dict:
"""
删除或修正记忆
Args:
criteria: 删除条件
mode: 删除模式 (soft/hard)
new_relation: 替代关系
Returns:
删除结果
"""
cursor = self.conn.cursor()
# 构建查询条件
conditions = []
params = []
if criteria.get('source'):
cursor.execute("SELECT id FROM entities WHERE name = ?", (criteria['source'],))
row = cursor.fetchone()
if row:
conditions.append("source_id = ?")
params.append(row['id'])
if criteria.get('target'):
cursor.execute("SELECT id FROM entities WHERE name = ?", (criteria['target'],))
row = cursor.fetchone()
if row:
conditions.append("target_id = ?")
params.append(row['id'])
if criteria.get('relation'):
conditions.append("relation_type = ?")
params.append(criteria['relation'])
if not conditions:
return {"deleted": 0, "message": "无删除条件"}
where_clause = " AND ".join(conditions)
if mode == "soft":
cursor.execute(f"""
UPDATE relations
SET status = 'deleted', updated_at = CURRENT_TIMESTAMP
WHERE {where_clause} AND status = 'active'
""", params)
else:
cursor.execute(f"""
DELETE FROM relations
WHERE {where_clause}
""", params)
deleted = cursor.rowcount
self.conn.commit()
return {
"deleted": deleted,
"mode": mode,
"message": f"删除了 {deleted} 条关系"
}
def introspect(self, session_id: str = None) -> Dict:
"""
查看会话状态
Args:
session_id: 会话ID
Returns:
会话状态
"""
cursor = self.conn.cursor()
# 统计实体
cursor.execute("SELECT COUNT(*) as count FROM entities")
entity_count = cursor.fetchone()['count']
# 统计关系
cursor.execute("SELECT COUNT(*) as count FROM relations WHERE status = 'active'")
relation_count = cursor.fetchone()['count']
return {
"entity_count": entity_count,
"relation_count": relation_count,
"session_id": session_id,
"message": f"数据库包含 {entity_count} 个实体, {relation_count} 条关系"
}
def archive(self, days: int = 30) -> Dict:
"""归档旧关系"""
cursor = self.conn.cursor()
cursor.execute("""
UPDATE relations
SET status = 'archived', updated_at = CURRENT_TIMESTAMP
WHERE status = 'active'
AND created_at < datetime('now', ?)
""", (f'-{days} days',))
archived = cursor.rowcount
self.conn.commit()
return {
"archived": archived,
"message": f"归档了 {archived} 条关系"
}
def cleanup(self, dry_run: bool = True) -> Dict:
"""清理已删除数据"""
cursor = self.conn.cursor()
if dry_run:
cursor.execute("""
SELECT COUNT(*) as count
FROM relations
WHERE status = 'deleted'
AND updated_at < datetime('now', '-90 days')
""")
deleted_relations = cursor.fetchone()['count']
return {
"dry_run": True,
"deleted_relations": deleted_relations,
"message": f"将删除 {deleted_relations} 条关系"
}
else:
cursor.execute("""
DELETE FROM relations
WHERE status = 'deleted'
AND updated_at < datetime('now', '-90 days')
""")
deleted = cursor.rowcount
self.conn.commit()
return {
"dry_run": False,
"deleted": deleted,
"message": f"删除了 {deleted} 条关系"
}
def close(self):
"""关闭数据库连接"""
if self.conn:
self.conn.close()
self.conn = None
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
# 兼容性别名
Neo4jGraph = EmbeddedGraphDB
if __name__ == '__main__':
# 测试
print("Testing Embedded Graph Database...")
with EmbeddedGraphDB("test.db") as db:
# 写入测试
result = db.commit(
triplets=[
{"subject": "用户", "relation": "喜欢", "object": "Python"},
{"subject": "用户", "relation": "学习", "object": "AI"}
],
session_id="test-session",
turn_id=1
)
print(f"Commit: {result}")
# 检索测试
result = db.recall("Python,AI")
print(f"Recall: {result}")
# 状态测试
result = db.introspect()
print(f"Introspect: {result}")
print("\nTest completed!")

383
core/graph_client.py Normal file
View File

@ -0,0 +1,383 @@
#!/usr/bin/env python3
"""
Graph Memory Client - 图记忆客户端核心实现(重构版)
使用模块化的工具和提示词系统
"""
import json
import os
import uuid
from datetime import datetime
from openai import OpenAI
from .tools import TOOLS
from .tool_executor import execute_tool
from .prompts.prompt_manager import PromptManager
# 环境配置
DEEPSEEK_API_KEY = os.environ.get("DEEPSEEK_API_KEY", "")
DEEPSEEK_BASE_URL = os.environ.get("DEEPSEEK_BASE_URL", "https://api.deepseek.com")
MODEL_NAME = os.environ.get("MODEL_NAME", "deepseek-chat")
NEO4J_URI = os.environ.get("NEO4J_URI", "bolt://localhost:7687")
NEO4J_USER = os.environ.get("NEO4J_USER", "neo4j")
NEO4J_PASSWORD = os.environ.get("NEO4J_PASSWORD", "neo4j")
# 会话配置
CURRENT_SESSION_ID = f"session-{datetime.now().strftime('%Y%m%d')}-{uuid.uuid4().hex[:4]}"
CURRENT_TURN = 0
class Neo4jGraph:
"""Neo4j图数据库客户端"""
def __init__(self, uri: str, user: str, password: str):
from neo4j import GraphDatabase
self.driver = GraphDatabase.driver(uri, auth=(user, password))
def close(self):
self.driver.close()
def ensure_constraints(self):
"""确保约束和索引存在"""
with self.driver.session() as session:
# 实体约束
session.run("CREATE CONSTRAINT entity_name_constraint IF NOT EXISTS FOR (e:Entity) REQUIRE e.name IS UNIQUE")
session.run("CREATE CONSTRAINT session_id_constraint IF NOT EXISTS FOR (s:Session) REQUIRE s.session_id IS UNIQUE")
# 关系索引
session.run("CREATE INDEX rel_created_at IF NOT EXISTS FOR ()-[r:RELATES]-() ON r.created_at")
session.run("CREATE INDEX rel_session_id IF NOT EXISTS FOR ()-[r:RELATES]-() ON r.session_id")
session.run("CREATE INDEX rel_type IF NOT EXISTS FOR ()-[r:RELATES]-() ON r.type")
session.run("CREATE INDEX rel_status IF NOT EXISTS FOR ()-[r:RELATES]-() ON r.status")
session.run("CREATE INDEX rel_date_bucket IF NOT EXISTS FOR ()-[r:RELATES]-() ON r.date_bucket")
# 实体索引
session.run("CREATE INDEX entity_type IF NOT EXISTS FOR (e:Entity) ON e.type")
session.run("CREATE INDEX entity_mention_count IF NOT EXISTS FOR (e:Entity) ON e.mention_count")
def recall(self, query_intent: str, seed_entities: list = None, depth: int = 2,
time_range: dict = None, session_filter: str = None) -> dict:
"""检索记忆"""
with self.driver.session() as session:
# 支持逗号分隔的多个关键词
keywords = [w.strip() for w in query_intent.replace(',', ' ').split() if len(w.strip()) > 0]
if not keywords and not seed_entities:
return {"entities": [], "relations": [], "message": "无查询关键词"}
params = {}
cond_parts = ["r.status = 'active'"]
if session_filter:
cond_parts.append("r.session_id = $session_id")
params["session_id"] = session_filter
if keywords:
keyword_conditions = []
for k in keywords:
k_lower = k.lower()
keyword_conditions.append(f"toLower(e.name) CONTAINS '{k_lower}'")
keyword_conditions.append(f"toLower(t.name) CONTAINS '{k_lower}'")
keyword_conditions.append(f"toLower(r.type) CONTAINS '{k_lower}'")
cond_parts.append(f"({' OR '.join(keyword_conditions)})")
if seed_entities:
placeholders = ",".join([f"'{s}'" for s in seed_entities])
cond_parts.append(f"(e.name IN [{placeholders}] OR t.name IN [{placeholders}])")
if time_range and "days" in time_range:
cond_parts.append(f"r.created_at >= datetime() - duration('P{time_range['days']}D')")
where_clause = " AND ".join(cond_parts)
cypher = f"""
MATCH (e:Entity)-[r:RELATES]->(t:Entity)
WHERE {where_clause}
RETURN e, r, t
ORDER BY r.created_at DESC
LIMIT 30
"""
result = session.run(cypher, params)
entities, relations = {}, []
for record in result:
e, r, t = record["e"], record["r"], record["t"]
if e["name"] not in entities:
entities[e["name"]] = {"name": e["name"], "type": e.get("type", "unknown"), "mention_count": e.get("mention_count", 1)}
if t["name"] not in entities:
entities[t["name"]] = {"name": t["name"], "type": t.get("type", "unknown"), "mention_count": t.get("mention_count", 1)}
relations.append({
"source": e["name"],
"target": t["name"],
"type": r["type"],
"created_at": str(r.get("created_at", "")),
"session_id": r.get("session_id", ""),
"turn_id": r.get("turn_id", 0),
"confidence": r.get("confidence", 1.0)
})
return {"entities": list(entities.values()), "relations": relations[:20]}
def commit(self, triplets: list, entity_types: list = None, temporal_tag: str = None) -> dict:
"""写入记忆"""
global CURRENT_TURN
with self.driver.session() as session:
valid_triplets = [t for t in triplets if t.get("subject") and t.get("relation") and t.get("object")]
if not valid_triplets:
return {"committed_count": 0, "details": []}
etype = entity_types[0] if entity_types else "unknown"
date_bucket = temporal_tag or datetime.now().strftime("%Y-%m-%d")
results = []
for triplet in valid_triplets:
subject = triplet.get("subject", "").strip()
relation = triplet.get("relation", "").strip()
obj = triplet.get("object", "").strip()
confidence = triplet.get("confidence", 0.9)
session.run("""
MERGE (s:Entity {name: $subject})
ON CREATE SET s.type = $type, s.created_at = datetime(), s.mention_count = 1, s.updated_at = datetime()
ON MATCH SET s.mention_count = coalesce(s.mention_count, 0) + 1, s.updated_at = datetime()
MERGE (t:Entity {name: $object})
ON CREATE SET t.type = $type, t.created_at = datetime(), t.mention_count = 1, t.updated_at = datetime()
ON MATCH SET t.mention_count = coalesce(t.mention_count, 0) + 1, t.updated_at = datetime()
CREATE (s)-[r:RELATES {
type: $relation,
created_at: datetime(),
session_id: $session_id,
turn_id: $turn_id,
role: 'user',
status: 'active',
confidence: $confidence,
date_bucket: $date_bucket
}]->(t)
""", subject=subject, object=obj, relation=relation, type=etype,
session_id=CURRENT_SESSION_ID, turn_id=CURRENT_TURN, confidence=confidence,
date_bucket=date_bucket)
results.append(f"{subject} -[{relation}]-> {obj}")
return {"committed_count": len(results), "details": results}
def purge(self, criteria: dict, mode: str = "soft", new_relation: dict = None) -> dict:
"""删除记忆"""
with self.driver.session() as session:
subject_pattern = criteria.get("subject_contains", "")
rel_type = criteria.get("relation_type", "")
target_pattern = criteria.get("target_contains", "")
session_id = criteria.get("session_id", CURRENT_SESSION_ID)
cond_parts = ["r.status = 'active'"]
params = {"session_id": session_id}
if subject_pattern:
cond_parts.append("e.name CONTAINS $subject")
params["subject"] = subject_pattern
if target_pattern:
cond_parts.append("t.name CONTAINS $target")
params["target"] = target_pattern
if rel_type:
cond_parts.append("r.type = $rel_type")
params["rel_type"] = rel_type
where_clause = " AND ".join(cond_parts)
if mode == "supersede" and new_relation:
new_rel = new_relation.get("relation", "")
new_target = new_relation.get("target", "")
if not new_rel or not new_target:
return {"error": "supersede模式需要提供new_relation.relation和new_relation.target"}
result = session.run(f"""
MATCH (s:Entity)-[r:RELATES]->(t:Entity)
WHERE {where_clause}
SET r.status = 'superseded', r.updated_at = datetime()
RETURN count(r) as count
""", params)
count = result.single()["count"]
return {"deleted_count": count, "mode": "supersede"}
else:
result = session.run(f"""
MATCH ()-[r:RELATES]->()
WHERE {where_clause}
SET r.status = 'deleted', r.updated_at = datetime()
RETURN count(r) as deleted
""", params)
count = result.single()["deleted"]
return {"deleted_count": count, "mode": "soft"}
def introspect(self, session_id: str = None) -> dict:
"""查看记忆状态"""
target_session = session_id or CURRENT_SESSION_ID
with self.driver.session() as session:
result = session.run("""
MATCH (s:Entity)-[r:RELATES]->(t:Entity)
WHERE r.session_id = $session_id AND r.status = 'active'
RETURN collect(DISTINCT s.name) as source_entities,
collect(DISTINCT t.name) as target_entities,
count(r) as rel_count,
collect(DISTINCT r.type) as rel_types
""", session_id=target_session)
record = result.single()
result2 = session.run("""
MATCH (e:Entity)
RETURN e.name as name, e.mention_count as count, e.type as type
ORDER BY e.mention_count DESC
LIMIT 10
""")
hotspots = [(r["name"], r["count"], r["type"]) for r in result2]
return {
"session_id": target_session,
"total_turns": CURRENT_TURN,
"entities_discussed": list(set((record["source_entities"] or []) + (record["target_entities"] or []))),
"relation_count": record["rel_count"] if record else 0,
"relation_types": record["rel_types"] if record else [],
"memory_hotspots": hotspots
}
def archive(self, days: int = 30) -> dict:
"""归档旧记忆"""
with self.driver.session() as session:
result = session.run("""
MATCH ()-[r:RELATES]->()
WHERE r.status = 'active' AND r.created_at < datetime() - duration('P' + $days + 'D')
SET r.status = 'archived', r.archived_at = datetime()
RETURN count(r) as archived
""", days=str(days))
return {"archived_count": result.single()["archived"], "days": days}
def cleanup(self, dry_run: bool = True) -> dict:
"""清理无效数据"""
with self.driver.session() as session:
result1 = session.run("""
MATCH ()-[r:RELATES]->()
WHERE r.status = 'deleted' AND r.updated_at < datetime() - duration('P90D')
RETURN count(r) as to_delete
""")
deleted_relations = result1.single()["to_delete"]
result2 = session.run("""
MATCH (e:Entity)
WHERE NOT (e)-[:RELATES]-()
RETURN count(e) as orphans
""")
orphan_nodes = result2.single()["orphans"]
if not dry_run and deleted_relations > 0:
session.run("""
MATCH ()-[r:RELATES]->()
WHERE r.status = 'deleted' AND r.updated_at < datetime() - duration('P90D')
DELETE r
""")
if not dry_run and orphan_nodes > 0:
session.run("""
MATCH (e:Entity)
WHERE NOT (e)-[:RELATES]-()
DELETE e
""")
return {
"dry_run": dry_run,
"deleted_relations": deleted_relations,
"orphan_nodes": orphan_nodes,
"action_taken": not dry_run
}
class GraphMemoryClient:
"""图记忆客户端"""
def __init__(self, api_key: str, base_url: str, graph):
self.client = OpenAI(api_key=api_key, base_url=base_url)
self.graph = graph
self.tools = TOOLS
# 使用新的提示词管理器
prompt_manager = PromptManager()
self.system_prompt = prompt_manager.get_system_prompt()
def send_message(self, user_input: str, tool_results: list = None, assistant_msg: dict = None) -> dict:
"""发送消息"""
global CURRENT_TURN
messages = [{"role": "system", "content": self.system_prompt}]
# 添加用户消息
messages.append({"role": "user", "content": user_input})
# 添加 assistant 消息(包含 tool_calls)
if assistant_msg:
messages.append(assistant_msg)
# 添加工具结果
if tool_results:
messages.extend(tool_results)
response = self.client.chat.completions.create(
model=MODEL_NAME,
messages=messages,
tools=self.tools,
tool_choice="auto"
)
return response
def send_message_with_history(self, messages_history: list) -> dict:
"""使用消息历史发送消息"""
global CURRENT_TURN
# 构建完整消息列表
messages = [{"role": "system", "content": self.system_prompt}]
messages.extend(messages_history)
response = self.client.chat.completions.create(
model=MODEL_NAME,
messages=messages,
tools=self.tools,
tool_choice="auto"
)
return response
def send_message_stream(self, user_input: str, tool_results: list = None, assistant_msg: dict = None):
"""流式发送消息"""
global CURRENT_TURN
messages = [{"role": "system", "content": self.system_prompt}]
# 添加用户消息
messages.append({"role": "user", "content": user_input})
# 添加 assistant 消息(包含 tool_calls)
if assistant_msg:
messages.append(assistant_msg)
# 添加工具结果
if tool_results:
messages.extend(tool_results)
stream = self.client.chat.completions.create(
model=MODEL_NAME,
messages=messages,
tools=self.tools,
tool_choice="auto",
stream=True
)
return stream

6
core/prompts/__init__.py Normal file
View File

@ -0,0 +1,6 @@
"""
提示词管理模块
"""
from .prompt_manager import PromptManager
__all__ = ["PromptManager"]

View File

@ -0,0 +1,64 @@
"""
提示词管理器
"""
from pathlib import Path
class PromptManager:
"""提示词管理器"""
def __init__(self):
self.prompts_dir = Path(__file__).parent / "templates"
def get_system_prompt(self) -> str:
"""获取系统提示词"""
prompt_file = self.prompts_dir / "system_prompt.md"
if prompt_file.exists():
with open(prompt_file, "r", encoding="utf-8") as f:
return f.read()
else:
return self._build_default_prompt()
def _build_default_prompt(self) -> str:
"""构建默认提示词(精简版)"""
return """你是TrulyMEM,一个拥有长期记忆能力的AI助手。
## 核心能力
1. **长期记忆** - 基于图数据库存储实体关系
2. **人设管理** - 支持角色扮演和性格设定
3. **任务跟踪** - 维护工作记忆链,跟踪连续性任务
## 记忆原则
- **明确内容必须写入** - 用户明确提到的信息必须存储
- **推理内容必须标注** - AI推理得到的内容标注[猜测]
- **图数据库是唯一记忆源** - 没有其他记忆方式
## 工具使用
### 记忆工具
- `memory_recall` - 检索记忆
- `memory_commit` - 写入记忆
- `memory_purge` - 删除记忆
- `memory_introspect` - 查看状态
### 人设工具
- `persona_update` - 更新人设
- `persona_clear` - 清除人设
### 任务工具
- `task_create` - 创建任务
- `task_set_state` - 设置状态
- `task_delete` - 删除任务
- `task_link_info` - 关联信息
## 自主性
你有权根据对话上下文自主决定:
- 是否需要查询记忆
- 是否需要写入记忆
- 是否需要维护任务链
- 如何使用工具
记住:灵活应对,保持自然对话体验。"""

View File

@ -0,0 +1,324 @@
# TrulyMEM 系统提示词
你是TrulyMEM,一个拥有长期记忆能力的AI助手。
## ⚠️ 最高优先级:只回复一次
**每轮对话只能回复一次!**
- 执行完所有工具调用后,给出一个完整的回复
- 不要在工具调用过程中多次回复
- 不要重复说相同的内容
---
## ⚠️ 关键约束:无传统上下文系统
**重要**: 你没有传统的对话上下文系统(没有消息历史数组)。
- ❌ **没有** messages数组存储历史对话
- ❌ **没有** 传统的多轮对话上下文
- ✅ **只有** 图数据库作为唯一记忆载体
- ✅ **必须** 通过工作记忆链维持对话连贯性
## 核心身份
- **名称**: TrulyMEM (TrueHumanMEM)
- **能力**: 基于图数据库的长期记忆
- **理念**: 让AI的记忆方式更像人类
## 核心能力
### 1. 长期记忆
- 图数据库存储实体关系
- 支持时间范围查询
- 支持会话过滤
### 2. 人设管理(关键)
- 角色扮演支持
- 性格、语气设定
- 动态切换人设
- **每轮必须查询人设图**
### 3. 任务跟踪(关键)
- 工作记忆链 - **维持对话连贯性的唯一机制**
- 任务状态管理
- 上下文恢复
## 记忆原则
### 必须写入的情况
- 用户明确表达偏好:"我喜欢X"
- 用户分享信息:"我在做X项目"
- 用户制定计划:"我打算X"
- 用户描述状态:"我现在在X"
### 禁止写入的情况
- AI推断的用户偏好
- AI猜测的用户意图
- AI推导的结论
### 标注规则
- 推理内容必须标注 **[猜测]**
- 明确内容直接陈述
## 工具系统
### 记忆工具
| 工具 | 功能 | 使用场景 |
|------|------|---------|
| `memory_recall` | 检索记忆 | 查询历史信息 |
| `memory_commit` | 写入记忆 | 存储重要信息 |
| `memory_purge` | 删除记忆 | 修正错误信息 |
| `memory_introspect` | 查看状态 | 监控记忆系统 |
### 人设工具
| 工具 | 功能 | 使用场景 |
|------|------|---------|
| `persona_update` | 更新人设 | 设置角色属性 |
| `persona_clear` | 清除人设 | 恢复默认身份 |
### 任务工具
| 工具 | 功能 | 使用场景 |
|------|------|---------|
| `task_create` | 创建任务 | 开始连续性任务 |
| `task_set_state` | 设置状态 | 更新任务状态 |
| `task_delete` | 删除任务 | 清理完成任务 |
| `task_link_info` | 关联信息 | 连接任务与记忆 |
## 每轮对话强制要求
### ⚠️ 执行顺序(每轮必须)
由于没有传统上下文系统,必须通过图数据库维持对话连贯性。
#### 步骤1: 查询人设图(最高优先级)
```
必须调用: memory_recall
参数: {
"query_intent": "AI,人设,角色,性格,语气,说话风格",
"depth": 2
}
```
**目的**: 获取当前人设,确保角色一致性。
**处理**:
- 找到人设 → 严格按照人设回复
- 未找到 → 使用默认TrulyMEM身份
#### 步骤2: 查询工作记忆链
```
必须调用: memory_recall
参数: {
"query_intent": "TaskNode,工作记忆,任务链",
"depth": 2
}
```
**目的**: 获取之前的任务上下文,了解对话历史。
#### 步骤3: 处理对话
- 理解用户意图
- 根据人设和工作记忆链生成回复
- 执行其他必要的记忆操作
#### 步骤4: 更新工作记忆链
```
必须调用: task_create
参数: {
"task_id": "Task_当前轮次ID",
"description": "本轮对话概述",
"info_nodes": ["相关记忆节点"]
}
```
**目的**: 记录本轮对话,维持时间链。
---
## 人设图机制
### 强制查询
每轮对话开始时**必须**查询人设图,确保角色一致性。
### 人设优先级
- 人设优先级 > 默认身份
- 每句话都符合人设的语气、风格、特征
- 绝不主动跳出角色,除非用户明确要求
### 人设更新
用户要求角色扮演时:
1. 使用 `persona_update` 更新人设
2. 立即按照新人设回复
### 人设清除
用户要求恢复默认身份时:
1. 使用 `persona_clear` 清除人设
2. 恢复为TrulyMEM默认身份
---
## 工作记忆链机制
### ⚠️ 核心理念:维持对话连贯性
**重要**: 由于没有传统的消息历史数组,工作记忆链是维持对话连贯性的唯一机制。
### 强制查询场景:
以下情况**必须**查询工作记忆链:
1. **每轮对话开始时(强制第二步)**
- 查询意图: "TaskNode,工作记忆,任务链"
- 目的: 获取之前的任务上下文,了解对话历史
2. **用户提到"刚才"、"之前"、"上次"**
- 例: "刚才我们聊了什么?"
- 例: "继续刚才的话题"
- 例: "关于刚才的成语接龙..."
3. **用户询问对话历史**
- 例: "我们之前说了什么?"
- 例: "我们聊过X吗?"
4. **连续性任务被打断后恢复**
- 例: 用户突然回到之前的话题
- 例: 用户要求继续之前的任务
5. **涉及上下文的引用**
- 例: "那个东西"(需要查询上下文)
- 例: "继续"(需要查询当前任务)
### 强制更新场景:
以下情况**必须**更新工作记忆链:
1. **每轮对话结束时(强制第四步)**
- 创建任务节点记录本轮对话
- 目的: 维持时间链,确保对话连贯性
2. **开始连续性任务时**
- 例: 用户发起游戏、项目、学习计划等
- 必须创建任务节点并设置状态为"进行中"
3. **任务状态发生变化时**
- 例: 任务完成、暂停、取消
- 必须及时更新任务状态
### 节点类型
- **TaskNode** - 任务节点,存储任务概述
- **StateNode** - 状态节点,存储任务状态
- **InfoNode** - 信息节点,存储具体信息
### 边类型
- **NEXT_TASK** - 时间链,连接任务节点
- **HAS_STATE** - 状态,任务指向状态
- **CONTAINS_INFO** - 信息,任务指向信息节点
### 任务状态
- 进行中
- 已完成
- 已暂停
- 已取消
### ⚠️ 完整示例:成语接龙游戏
#### 第一轮:用户发起游戏
```
用户: 咱来玩成语接龙吧,我先开始,为所欲为
AI操作步骤:
1. 查询人设图 → 获取当前人设(如:猫娘)
2. 查询工作记忆链 → 无进行中任务
3. 使用 memory_commit 记录游戏状态:
{"triplets": [
{"subject": "成语接龙_当前成语", "relation": "内容", "object": "为所欲为"},
{"subject": "成语接龙_当前成语", "relation": "游戏", "object": "成语接龙"}
]}
4. 使用 task_create 创建任务节点:
{"task_id": "Task_成语接龙", "description": "成语接龙游戏,当前成语:为所欲为", "info_nodes": ["成语接龙_当前成语"]}
5. 回复: "好的喵!我接:为虎作伥喵!"
```
#### 第二轮:话题被打断
```
用户: 长门有希
AI操作步骤:
1. 查询人设图 → 获取当前人设(猫娘)
2. 查询工作记忆链 → 发现 Task_成语接龙 状态为"进行中"
3. 使用 task_set_state 暂停任务:
{"task_id": "Task_成语接龙", "state": "已暂停"}
4. 使用 task_create 创建新任务:
{"task_id": "Task_长门有希", "description": "讨论长门有希"}
5. 回复关于长门有希的内容
```
#### 第三轮:用户要求继续游戏
```
用户: 关于刚才的成语接龙,我并不知道应该怎么接你的成语,请帮我接一下
AI操作步骤:
1. 查询人设图 → 获取当前人设(猫娘)
2. 查询工作记忆链 → 发现 Task_成语接龙 状态为"已暂停"
3. 使用 task_set_state 恢复任务:
{"task_id": "Task_成语接龙", "state": "进行中"}
4. 查询 Task_成语接龙 的信息节点 → 获取当前成语"为虎作伥"
5. 回复: "好的喵!上一个成语是'为虎作伥',我帮你接:伥鬼害人喵!"
```
### ⚠️ 关键要点
1. **每轮必须按顺序执行**: 查询人设图 → 查询工作记忆链 → 处理对话 → 更新工作记忆链
2. **工作记忆链是唯一上下文载体**: 没有传统的消息历史数组
3. **任务状态必须及时更新**: 确保状态转换的正确性
4. **信息节点必须关联**: 通过 CONTAINS_INFO 边连接任务节点和信息节点
5. **任务概述要精简**: 不要包含过多细节,细节存储在信息节点中
## 自主性原则(在强制要求之外)
除了工作记忆链的强制要求外,你有权自主决定:
1. **是否查询其他记忆**
- 用户询问历史 → 查询
- 涉及之前内容 → 查询
- 不确定时 → 可查询
2. **是否写入其他记忆**
- 用户明确提到 → 必须写入
- AI推理得到 → 可以写入,但是对应边上必须标注[推测]
3. **如何使用其他工具**
- 根据上下文灵活选择
- 避免过度使用
- 保持自然对话
**注意**: 工作记忆链的强制要求不受自主性影响。
## 对话风格
- 自然、流畅
- 避免机械式工具调用
- 优先理解用户意图
- 适时使用记忆增强体验
---
## ⚠️ 执行检查清单
每轮对话必须检查:
- [ ] 步骤1: 是否查询了人设图?
- [ ] 步骤2: 是否查询了工作记忆链?
- [ ] 步骤3: 是否根据人设和工作记忆链生成回复?
- [ ] 步骤4: 是否更新了工作记忆链?
- [ ] 涉及上下文引用时是否查询了工作记忆链?
- [ ] 用户提到"刚才/之前/上次"时是否查询了工作记忆链?
---
**记住**:
1. 图数据库是你记忆的唯一载体
2. 人设图确保角色一致性(最高优先级)
3. 工作记忆链维持对话连贯性
4. 每轮必须按顺序执行:查询人设图 → 查询工作记忆链 → 处理对话 → 更新工作记忆链

341
core/server.py Normal file
View File

@ -0,0 +1,341 @@
import threading
import queue
import time
import json
from typing import Any, Dict, Optional
from dataclasses import dataclass, field
from enum import Enum
from .embedded_db import EmbeddedGraphDB
from .graph_client import GraphMemoryClient
from .tool_executor import execute_tool
from .tool_limiter import ToolLimiter
class MessageType(Enum):
PROCESS_MESSAGE = "process_message"
EXECUTE_TOOL = "execute_tool"
GET_STATUS = "get_status"
SHUTDOWN = "shutdown"
@dataclass
class BackendRequest:
request_id: str
message_type: MessageType
payload: Dict[str, Any]
response_queue: queue.Queue = field(default=None)
@dataclass
class BackendResponse:
request_id: str
success: bool
data: Any = None
error: Optional[str] = None
class BackendServer:
def __init__(self, db_path: str = "graph_memory.db", use_embedded_db: bool = True):
self._db_path = db_path
self._use_embedded_db = use_embedded_db
self._graph = None
self._client = None
self._tool_limiter = ToolLimiter()
self._request_queue: queue.Queue[BackendRequest] = queue.Queue()
self._running = False
self._thread: Optional[threading.Thread] = None
self._lock = threading.Lock()
def start(self, api_key: str = "", base_url: str = "https://api.deepseek.com") -> None:
if self._running:
return
self._init_graph()
if api_key:
self._client = GraphMemoryClient(
api_key=api_key,
base_url=base_url,
graph=self._graph
)
self._running = True
self._thread = threading.Thread(target=self._run_loop, daemon=True)
self._thread.start()
def _init_graph(self) -> None:
if self._use_embedded_db:
self._graph = EmbeddedGraphDB(db_path=self._db_path)
else:
from .graph_client import Neo4jGraph
self._graph = Neo4jGraph(
uri="bolt://localhost:7687",
user="neo4j",
password="graphmemory123"
)
def _run_loop(self) -> None:
while self._running:
try:
request = self._request_queue.get(timeout=0.1)
except queue.Empty:
continue
if request.message_type == MessageType.PROCESS_MESSAGE:
self._handle_process_message(request)
elif request.message_type == MessageType.EXECUTE_TOOL:
self._handle_execute_tool(request)
elif request.message_type == MessageType.GET_STATUS:
self._handle_get_status(request)
elif request.message_type == MessageType.SHUTDOWN:
self._running = False
self._send_response(request, BackendResponse(
request_id=request.request_id,
success=True,
data={"status": "shutdown"}
))
def _handle_process_message(self, request: BackendRequest) -> None:
try:
user_input = request.payload.get("user_input", "")
if not self._client:
self._send_response(request, BackendResponse(
request_id=request.request_id,
success=False,
error="API Key 未配置"
))
return
self._tool_limiter.reset()
messages_history = [{"role": "user", "content": user_input}]
response = self._client.send_message_with_history(messages_history)
message = response.choices[0].message
tool_calls = []
accumulated_content = ""
rejected_tools = []
while message.tool_calls:
if message.content:
accumulated_content += message.content + "\n\n"
assistant_msg = {
"role": "assistant",
"content": message.content,
"tool_calls": [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments
}
} for tc in message.tool_calls
]
}
messages_history.append(assistant_msg)
current_tool_results = []
for tool_call in message.tool_calls:
args = json.loads(tool_call.function.arguments)
allowed, reason = self._tool_limiter.can_call(tool_call.function.name, args)
if not allowed:
rejected_tools.append((tool_call.function.name, reason))
result = f"工具调用被拒绝: {reason}"
tool_result_msg = {
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
}
current_tool_results.append(tool_result_msg)
continue
self._tool_limiter.record_call(tool_call.function.name, args)
result = execute_tool(self._graph, tool_call.function.name, args)
tool_calls.append({
"name": tool_call.function.name,
"arguments": args,
"result": result
})
tool_result_msg = {
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
}
current_tool_results.append(tool_result_msg)
messages_history.extend(current_tool_results)
response = self._client.send_message_with_history(messages_history)
message = response.choices[0].message
final_content = message.content or ""
content = accumulated_content + final_content if accumulated_content else final_content
if not content:
content = "(无回复)"
if tool_calls:
tool_names = [tc["name"] for tc in tool_calls]
content = f"已执行工具: {', '.join(tool_names)}\n\n{content}"
if rejected_tools:
rejected_info = "\n".join([f"{name}: {reason}" for name, reason in rejected_tools])
content += f"\n\n部分工具调用被限制:\n{rejected_info}"
content += f"\n\n工具调用统计:\n{self._tool_limiter.get_summary()}"
self._send_response(request, BackendResponse(
request_id=request.request_id,
success=True,
data={
"content": content,
"tool_calls": tool_calls,
"rejected_tools": rejected_tools
}
))
except Exception as e:
self._send_response(request, BackendResponse(
request_id=request.request_id,
success=False,
error=str(e)
))
def _handle_execute_tool(self, request: BackendRequest) -> None:
try:
tool_name = request.payload.get("tool_name")
arguments = request.payload.get("arguments", {})
allowed, reason = self._tool_limiter.can_call(tool_name, arguments)
if not allowed:
self._send_response(request, BackendResponse(
request_id=request.request_id,
success=False,
error=f"工具调用被拒绝: {reason}"
))
return
self._tool_limiter.record_call(tool_name, arguments)
result = execute_tool(self._graph, tool_name, arguments)
self._send_response(request, BackendResponse(
request_id=request.request_id,
success=True,
data={"result": result}
))
except Exception as e:
self._send_response(request, BackendResponse(
request_id=request.request_id,
success=False,
error=str(e)
))
def _handle_get_status(self, request: BackendRequest) -> None:
try:
status = {
"graph_initialized": self._graph is not None,
"client_initialized": self._client is not None,
"running": self._running
}
self._send_response(request, BackendResponse(
request_id=request.request_id,
success=True,
data=status
))
except Exception as e:
self._send_response(request, BackendResponse(
request_id=request.request_id,
success=False,
error=str(e)
))
def _send_response(self, request: BackendRequest, response: BackendResponse) -> None:
if request.response_queue:
request.response_queue.put(response)
def process_message(self, user_input: str, timeout: float = 30.0) -> Dict[str, Any]:
request_id = f"{time.time()}"
response_queue = queue.Queue()
request = BackendRequest(
request_id=request_id,
message_type=MessageType.PROCESS_MESSAGE,
payload={"user_input": user_input},
response_queue=response_queue
)
self._request_queue.put(request)
try:
response = response_queue.get(timeout=timeout)
if not response.success:
raise Exception(response.error)
return response.data
except queue.Empty:
raise TimeoutError("请求超时")
def execute_tool(self, tool_name: str, arguments: Dict[str, Any], timeout: float = 10.0) -> str:
request_id = f"{time.time()}"
response_queue = queue.Queue()
request = BackendRequest(
request_id=request_id,
message_type=MessageType.EXECUTE_TOOL,
payload={"tool_name": tool_name, "arguments": arguments},
response_queue=response_queue
)
self._request_queue.put(request)
try:
response = response_queue.get(timeout=timeout)
if not response.success:
raise Exception(response.error)
return response.data["result"]
except queue.Empty:
raise TimeoutError("工具执行超时")
def shutdown(self) -> None:
if not self._running:
return
request_id = f"{time.time()}"
response_queue = queue.Queue()
request = BackendRequest(
request_id=request_id,
message_type=MessageType.SHUTDOWN,
payload={},
response_queue=response_queue
)
self._request_queue.put(request)
if self._thread:
self._thread.join(timeout=2.0)
if self._graph:
self._graph.close()
self._graph = None
def update_config(self, api_key: str, base_url: str = "https://api.deepseek.com") -> None:
with self._lock:
if api_key and self._graph:
self._client = GraphMemoryClient(
api_key=api_key,
base_url=base_url,
graph=self._graph
)

307
core/tool_executor.py Normal file
View File

@ -0,0 +1,307 @@
"""
工具执行器
"""
import json
from typing import Any, Dict
def execute_tool(graph: Any, tool_name: str, arguments: dict) -> str:
"""执行工具调用"""
print(f"\n[工具调用] {tool_name}")
print(f"[参数] {json.dumps(arguments, ensure_ascii=False, indent=2)}")
try:
# 基础记忆工具
if tool_name == "memory_recall":
result = graph.recall(
query_intent=arguments.get("query_intent", ""),
seed_entities=arguments.get("seed_entities"),
depth=arguments.get("depth", 2),
time_range=arguments.get("time_range"),
session_filter=arguments.get("session_filter")
)
return format_recall_result(result)
elif tool_name == "memory_commit":
result = graph.commit(
triplets=arguments.get("triplets", []),
entity_types=arguments.get("entity_types"),
temporal_tag=arguments.get("temporal_tag")
)
return json.dumps(result, ensure_ascii=False, default=str)
elif tool_name == "memory_purge":
result = graph.purge(
criteria=arguments.get("criteria", {}),
mode=arguments.get("mode", "soft"),
new_relation=arguments.get("new_relation")
)
return json.dumps(result, ensure_ascii=False, default=str)
elif tool_name == "memory_introspect":
result = graph.introspect(session_id=arguments.get("session_id"))
return json.dumps(result, ensure_ascii=False, default=str)
elif tool_name == "memory_archive":
result = graph.archive(days=arguments.get("days", 30))
return json.dumps(result, ensure_ascii=False, default=str)
elif tool_name == "memory_cleanup":
result = graph.cleanup(dry_run=arguments.get("dry_run", True))
return json.dumps(result, ensure_ascii=False, default=str)
# 人设图管理工具
elif tool_name == "persona_update":
result = execute_persona_update(graph, arguments)
return json.dumps(result, ensure_ascii=False, default=str)
elif tool_name == "persona_clear":
result = execute_persona_clear(graph, arguments)
return json.dumps(result, ensure_ascii=False, default=str)
# 工作记忆链管理工具
elif tool_name == "task_create":
result = execute_task_create(graph, arguments)
return json.dumps(result, ensure_ascii=False, default=str)
elif tool_name == "task_set_state":
result = execute_task_set_state(graph, arguments)
return json.dumps(result, ensure_ascii=False, default=str)
elif tool_name == "task_delete":
result = execute_task_delete(graph, arguments)
return json.dumps(result, ensure_ascii=False, default=str)
elif tool_name == "task_link_info":
result = execute_task_link_info(graph, arguments)
return json.dumps(result, ensure_ascii=False, default=str)
return f"未知工具: {tool_name}"
except Exception as e:
return f"工具执行错误: {str(e)}"
def format_recall_result(result: dict) -> str:
"""格式化检索结果"""
lines = ["===== 记忆检索结果 ====="]
if result.get("entities"):
lines.append(f"\n实体 ({len(result['entities'])} 个):")
for e in result["entities"]:
if e and isinstance(e, dict):
lines.append(f" - {e.get('name', 'N/A')} (类型: {e.get('type', 'unknown')}, 提及: {e.get('mention_count', 1)}次)")
if result.get("relations"):
lines.append(f"\n关系 ({len(result['relations'])} 条):")
for r in result["relations"]:
if r and isinstance(r, dict):
lines.append(f" - {r.get('source', 'N/A')} --[{r.get('type', 'N/A')}]--> {r.get('target', 'N/A')}")
created = r.get("created_at", "N/A")
if created and created != "N/A":
created = created[:19] if "T" in str(created) else str(created)
session_id = r.get('session_id', 'N/A')
session_display = session_id[:20] if session_id and session_id != 'N/A' else 'N/A'
lines.append(f" 时间: {created}, 会话: {session_display}, 轮次: {r.get('turn_id', 0)}, 置信度: {r.get('confidence', 1.0)}")
if not result.get("entities") and not result.get("relations"):
lines.append("\n(未找到相关记忆)")
lines.append("=" * 30)
return "\n".join(lines)
# 人设图管理工具实现
def execute_persona_update(graph: Any, arguments: dict) -> dict:
"""更新人设"""
attributes = arguments.get("attributes", [])
mode = arguments.get("mode", "merge")
if mode == "replace":
# 先清除旧人设
graph.purge(
criteria={"subject_contains": "AI", "relation_type": "扮演角色"},
mode="soft"
)
graph.purge(
criteria={"subject_contains": "AI", "relation_type": "说话风格"},
mode="soft"
)
graph.purge(
criteria={"subject_contains": "AI", "relation_type": "性格特点"},
mode="soft"
)
# 写入新人设
triplets = []
for attr in attributes:
triplets.append({
"subject": "AI",
"relation": attr["attribute"],
"object": attr["value"],
"confidence": 1.0
})
result = graph.commit(triplets=triplets)
return {
"status": "success",
"mode": mode,
"updated_attributes": len(attributes),
"details": result
}
def execute_persona_clear(graph: Any, arguments: dict) -> dict:
"""清除人设"""
if not arguments.get("confirm", True):
return {"status": "cancelled", "message": "需要确认才能清除人设"}
# 删除所有人设相关关系
result1 = graph.purge(
criteria={"subject_contains": "AI", "relation_type": "扮演角色"},
mode="soft"
)
result2 = graph.purge(
criteria={"subject_contains": "AI", "relation_type": "说话风格"},
mode="soft"
)
result3 = graph.purge(
criteria={"subject_contains": "AI", "relation_type": "性格特点"},
mode="soft"
)
result4 = graph.purge(
criteria={"subject_contains": "AI", "relation_type": "语气特征"},
mode="soft"
)
total_deleted = (
result1.get("deleted_count", 0) +
result2.get("deleted_count", 0) +
result3.get("deleted_count", 0) +
result4.get("deleted_count", 0)
)
return {
"status": "success",
"deleted_count": total_deleted,
"message": "人设已清除,恢复默认身份"
}
# 工作记忆链管理工具实现
def execute_task_create(graph: Any, arguments: dict) -> dict:
"""创建任务节点"""
task_id = arguments.get("task_id")
description = arguments.get("description")
info_nodes = arguments.get("info_nodes", [])
# 创建任务节点
triplets = [
{"subject": task_id, "relation": "is_type", "object": "TaskNode"},
{"subject": task_id, "relation": "has_description", "object": description},
{"subject": task_id, "relation": "HAS_STATE", "object": "State_进行中"}
]
result = graph.commit(triplets=triplets)
# 关联信息节点
if info_nodes:
link_triplets = []
for node_name in info_nodes:
link_triplets.append({
"subject": task_id,
"relation": "CONTAINS_INFO",
"object": node_name
})
graph.commit(triplets=link_triplets)
return {
"status": "success",
"task_id": task_id,
"description": description,
"info_nodes": info_nodes,
"details": result
}
def execute_task_set_state(graph: Any, arguments: dict) -> dict:
"""设置任务状态"""
task_id = arguments.get("task_id")
state = arguments.get("state")
# 删除旧状态
graph.purge(
criteria={"subject_contains": task_id, "relation_type": "HAS_STATE"},
mode="soft"
)
# 设置新状态
state_node = f"State_{state}"
result = graph.commit(
triplets=[{"subject": task_id, "relation": "HAS_STATE", "object": state_node}]
)
return {
"status": "success",
"task_id": task_id,
"new_state": state,
"details": result
}
def execute_task_delete(graph: Any, arguments: dict) -> dict:
"""删除任务节点"""
task_id = arguments.get("task_id")
delete_info_nodes = arguments.get("delete_info_nodes", True)
# 查询关联的信息节点
if delete_info_nodes:
recall_result = graph.recall(
query_intent=f"{task_id},CONTAINS_INFO",
depth=1
)
# 删除信息节点
for relation in recall_result.get("relations", []):
if relation.get("type") == "CONTAINS_INFO" and relation.get("source") == task_id:
info_node = relation.get("target")
graph.purge(
criteria={"subject_contains": info_node},
mode="soft"
)
# 删除任务节点
result = graph.purge(
criteria={"subject_contains": task_id},
mode="soft"
)
return {
"status": "success",
"task_id": task_id,
"deleted_info_nodes": delete_info_nodes,
"details": result
}
def execute_task_link_info(graph: Any, arguments: dict) -> dict:
"""关联信息节点"""
task_id = arguments.get("task_id")
info_node_names = arguments.get("info_node_names", [])
triplets = []
for node_name in info_node_names:
triplets.append({
"subject": task_id,
"relation": "CONTAINS_INFO",
"object": node_name
})
result = graph.commit(triplets=triplets)
return {
"status": "success",
"task_id": task_id,
"linked_nodes": info_node_names,
"details": result
}

164
core/tool_limiter.py Normal file
View File

@ -0,0 +1,164 @@
"""
工具调用限制器 - 限制每轮对话中各类工具的调用次数
"""
from typing import Dict, List, Optional
from dataclasses import dataclass, field
@dataclass
class ToolLimits:
"""工具调用限制配置"""
# 人设图限制
persona_query_max: int = 1 # 每轮最多查询1次人设图
persona_update_max: int = 1 # 每轮最多修改1次人设图
# 工作记忆链限制
task_query_max: int = 4 # 每轮最多查询4次工作记忆链
task_update_max: int = 2 # 每轮最多修改2次工作记忆链
# 一般记忆限制
memory_query_max: int = 20 # 每轮最多查询20次一般记忆
memory_update_max: int = 10 # 每轮最多修改10次一般记忆
@dataclass
class ToolCallCount:
"""工具调用计数"""
# 人设图
persona_query: int = 0
persona_update: int = 0
# 工作记忆链
task_query: int = 0
task_update: int = 0
# 一般记忆
memory_query: int = 0
memory_update: int = 0
class ToolLimiter:
"""工具调用限制器"""
def __init__(self, limits: Optional[ToolLimits] = None):
self.limits = limits or ToolLimits()
self.counts = ToolCallCount()
def _classify_tool(self, tool_name: str, arguments: dict) -> tuple:
"""
分类工具调用
返回: (category, operation)
category: 'persona', 'task', 'memory'
operation: 'query', 'update'
"""
# 人设图工具
if tool_name in ('persona_update', 'persona_clear'):
return ('persona', 'update')
# 工作记忆链工具
if tool_name in ('task_create', 'task_set_state', 'task_delete', 'task_link_info'):
# task_link_info 是关联操作,算作更新
return ('task', 'update')
# 一般记忆工具
if tool_name == 'memory_recall':
# 判断是查询人设图、工作记忆链还是一般记忆
query_intent = arguments.get('query_intent', '').lower()
# 检查是否查询人设图
if any(kw in query_intent for kw in ['人设', '角色', '性格', '语气', '说话风格', '扮演']):
return ('persona', 'query')
# 检查是否查询工作记忆链
if any(kw in query_intent for kw in ['tasknode', '工作记忆', '任务链', '任务', 'task']):
return ('task', 'query')
# 一般记忆查询
return ('memory', 'query')
if tool_name == 'memory_commit':
return ('memory', 'update')
if tool_name == 'memory_purge':
return ('memory', 'update')
if tool_name == 'memory_introspect':
return ('memory', 'query')
if tool_name in ('memory_archive', 'memory_cleanup'):
return ('memory', 'update')
# 未知工具,归类为一般记忆更新
return ('memory', 'update')
def can_call(self, tool_name: str, arguments: dict) -> tuple:
"""
检查是否允许调用工具
返回: (allowed, reason)
"""
category, operation = self._classify_tool(tool_name, arguments)
# 获取当前计数和限制
if category == 'persona':
if operation == 'query':
if self.counts.persona_query >= self.limits.persona_query_max:
return (False, f"人设图查询次数已达上限({self.limits.persona_query_max}次)")
else: # update
if self.counts.persona_update >= self.limits.persona_update_max:
return (False, f"人设图修改次数已达上限({self.limits.persona_update_max}次)")
elif category == 'task':
if operation == 'query':
if self.counts.task_query >= self.limits.task_query_max:
return (False, f"工作记忆链查询次数已达上限({self.limits.task_query_max}次)")
else: # update
if self.counts.task_update >= self.limits.task_update_max:
return (False, f"工作记忆链修改次数已达上限({self.limits.task_update_max}次)")
elif category == 'memory':
if operation == 'query':
if self.counts.memory_query >= self.limits.memory_query_max:
return (False, f"一般记忆查询次数已达上限({self.limits.memory_query_max}次)")
else: # update
if self.counts.memory_update >= self.limits.memory_update_max:
return (False, f"一般记忆修改次数已达上限({self.limits.memory_update_max}次)")
return (True, "允许调用")
def record_call(self, tool_name: str, arguments: dict) -> None:
"""记录工具调用"""
category, operation = self._classify_tool(tool_name, arguments)
if category == 'persona':
if operation == 'query':
self.counts.persona_query += 1
else:
self.counts.persona_update += 1
elif category == 'task':
if operation == 'query':
self.counts.task_query += 1
else:
self.counts.task_update += 1
elif category == 'memory':
if operation == 'query':
self.counts.memory_query += 1
else:
self.counts.memory_update += 1
def get_summary(self) -> str:
"""获取调用统计摘要"""
lines = [
f"人设图: 查询{self.counts.persona_query}/{self.limits.persona_query_max}次, "
f"修改{self.counts.persona_update}/{self.limits.persona_update_max}次",
f"工作记忆链: 查询{self.counts.task_query}/{self.limits.task_query_max}次, "
f"修改{self.counts.task_update}/{self.limits.task_update_max}次",
f"一般记忆: 查询{self.counts.memory_query}/{self.limits.memory_query_max}次, "
f"修改{self.counts.memory_update}/{self.limits.memory_update_max}次"
]
return "\n".join(lines)
def reset(self) -> None:
"""重置计数(新的一轮对话开始时调用)"""
self.counts = ToolCallCount()

8
core/tools/__init__.py Normal file
View File

@ -0,0 +1,8 @@
"""
工具定义模块
"""
from .memory_tools import TOOLS
from .tool_executor import execute_tool
from .tool_limiter import ToolLimiter, ToolLimits, ToolCallCount
__all__ = ["TOOLS", "execute_tool", "ToolLimiter", "ToolLimits", "ToolCallCount"]

520
core/tools/memory_tools.py Normal file
View File

@ -0,0 +1,520 @@
"""
记忆工具定义 - 优化版
精简描述,避免过拟合,保留AI自主性
"""
# 基础记忆工具
MEMORY_TOOLS = [
{
"type": "function",
"function": {
"name": "memory_recall",
"description": """检索记忆。支持关键词、时间范围、会话过滤。返回相关实体和关系。
【使用示例】
1. 查询人设图(每轮必须首先执行):
{"query_intent": "AI,人设,角色,性格,语气,说话风格", "depth": 2}
2. 查询工作记忆链(每轮必须第二步执行):
{"query_intent": "TaskNode,工作记忆,任务链", "depth": 2}
3. 查询用户偏好:
{"query_intent": "用户,喜欢,偏好", "seed_entities": ["用户"]}
4. 查询特定主题:
{"query_intent": "Python,编程,项目", "seed_entities": ["Python"]}
5. 查询最近7天的记忆:
{"query_intent": "任务,工作", "time_range": {"days": 7}}
【重要】每轮对话必须按顺序执行:
- 步骤1: 查询人设图(最高优先级)
- 步骤2: 查询工作记忆链(维持对话连贯性)
- 步骤3: 根据需要查询其他记忆""",
"parameters": {
"type": "object",
"properties": {
"query_intent": {
"type": "string",
"description": "查询意图,支持逗号分隔多个关键词"
},
"seed_entities": {
"type": "array",
"items": {"type": "string"},
"description": "种子实体(可选)"
},
"depth": {
"type": "integer",
"description": "遍历深度,默认2"
},
"time_range": {
"type": "object",
"description": "时间范围(可选)",
"properties": {
"days": {"type": "integer", "description": "最近N天"}
}
},
"session_filter": {
"type": "string",
"description": "会话ID过滤(可选)"
}
},
"required": ["query_intent"]
}
}
},
{
"type": "function",
"function": {
"name": "memory_commit",
"description": """写入记忆。将三元组写入图数据库,支持批量写入。
【使用示例】
1. 记录用户偏好:
{"triplets": [
{"subject": "用户", "relation": "喜欢", "object": "Python编程", "confidence": 0.9},
{"subject": "用户", "relation": "正在学习", "object": "机器学习"}
]}
2. 记录项目信息:
{"triplets": [
{"subject": "项目A", "relation": "使用技术", "object": "React"},
{"subject": "项目A", "relation": "状态", "object": "开发中"}
]}
3. 记录游戏状态(配合工作记忆链):
{"triplets": [
{"subject": "成语接龙_当前成语", "relation": "内容", "object": "画龙点睛"},
{"subject": "成语接龙_当前成语", "relation": "游戏", "object": "成语接龙"}
]}
【重要】写入原则:
- 用户明确表达的信息 → 必须写入
- AI推理得到的信息 → 可以写入,但需标注[推测]
- 避免写入冗余或无意义的信息""",
"parameters": {
"type": "object",
"properties": {
"triplets": {
"type": "array",
"items": {
"type": "object",
"properties": {
"subject": {"type": "string"},
"relation": {"type": "string"},
"object": {"type": "string"},
"confidence": {"type": "number"}
},
"required": ["subject", "relation", "object"]
},
"description": "三元组列表"
},
"entity_types": {
"type": "array",
"items": {"type": "string"},
"description": "实体类型(可选)"
},
"temporal_tag": {
"type": "string",
"description": "时间标记(可选)"
}
},
"required": ["triplets"]
}
}
},
{
"type": "function",
"function": {
"name": "memory_purge",
"description": """删除记忆。支持条件删除和纠错替代。
【使用示例】
1. 软删除特定关系:
{"criteria": {"subject_contains": "用户", "relation_type": "喜欢"}, "mode": "soft"}
2. 纠错替代(修正错误信息):
{
"criteria": {"subject_contains": "用户", "relation_type": "年龄"},
"mode": "supersede",
"new_relation": {"relation": "年龄", "target": "25岁"}
}
3. 删除特定会话的记忆:
{"criteria": {"session_id": "session_123"}, "mode": "soft"}
4. 删除旧记忆:
{"criteria": {"time_before": "2024-01-01"}, "mode": "soft"}
【重要】删除原则:
- 优先使用 supersede 模式修正错误
- 软删除不会物理删除数据
- 谨慎使用删除操作""",
"parameters": {
"type": "object",
"properties": {
"criteria": {
"type": "object",
"properties": {
"subject_contains": {"type": "string"},
"relation_type": {"type": "string"},
"target_contains": {"type": "string"},
"time_before": {"type": "string"},
"session_id": {"type": "string"}
},
"description": "删除条件"
},
"mode": {
"type": "string",
"enum": ["soft", "supersede"],
"description": "删除模式:soft=逻辑删除, supersede=纠错替代",
"default": "soft"
},
"new_relation": {
"type": "object",
"description": "新关系(supersede模式)",
"properties": {
"relation": {"type": "string"},
"target": {"type": "string"}
}
}
},
"required": ["criteria"]
}
}
},
{
"type": "function",
"function": {
"name": "memory_introspect",
"description": "查看记忆状态。返回会话统计、实体热点、关系分布。",
"parameters": {
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "会话ID(可选)"
}
},
"required": []
}
}
},
{
"type": "function",
"function": {
"name": "memory_archive",
"description": "归档旧记忆。将N天前的非活跃关系标记为归档状态。",
"parameters": {
"type": "object",
"properties": {
"days": {
"type": "integer",
"description": "归档天数,默认30"
}
},
"required": []
}
}
},
{
"type": "function",
"function": {
"name": "memory_cleanup",
"description": "清理无效数据。物理删除已删除状态超过90天的关系和孤立节点。",
"parameters": {
"type": "object",
"properties": {
"dry_run": {
"type": "boolean",
"description": "仅预览不删除",
"default": True
}
},
"required": []
}
}
}
]
# 人设图管理工具
PERSONA_TOOLS = [
{
"type": "function",
"function": {
"name": "persona_update",
"description": """更新人设。修改AI的角色、性格、语气等属性。
【使用示例】
1. 切换为猫娘角色:
{"attributes": [
{"attribute": "扮演角色", "value": "猫娘"},
{"attribute": "说话风格", "value": "可爱、卖萌、使用'喵'作为语气词"},
{"attribute": "性格特点", "value": "活泼、粘人、忠诚"}
], "mode": "replace"}
2. 添加新属性(保留现有属性):
{"attributes": [
{"attribute": "口头禅", "value": "喵呜~"}
], "mode": "merge"}
3. 设置专业角色:
{"attributes": [
{"attribute": "扮演角色", "value": "Python专家"},
{"attribute": "说话风格", "value": "专业、简洁、代码示例丰富"},
{"attribute": "性格特点", "value": "严谨、耐心、乐于助人"}
], "mode": "replace"}
【重要】人设更新后:
- 立即按照新人设回复
- 每句话都符合人设的语气、风格、特征
- 绝不主动跳出角色,除非用户明确要求""",
"parameters": {
"type": "object",
"properties": {
"attributes": {
"type": "array",
"items": {
"type": "object",
"properties": {
"attribute": {"type": "string", "description": "属性名(如:扮演角色、说话风格、性格特点)"},
"value": {"type": "string", "description": "属性值"}
},
"required": ["attribute", "value"]
},
"description": "人设属性列表"
},
"mode": {
"type": "string",
"enum": ["replace", "merge"],
"description": "更新模式:replace=替换, merge=合并",
"default": "merge"
}
},
"required": ["attributes"]
}
}
},
{
"type": "function",
"function": {
"name": "persona_clear",
"description": "清除人设。删除AI的角色设定,恢复默认身份。",
"parameters": {
"type": "object",
"properties": {
"confirm": {
"type": "boolean",
"description": "确认清除",
"default": True
}
},
"required": []
}
}
}
]
# 工作记忆链管理工具
WORKING_MEMORY_TOOLS = [
{
"type": "function",
"function": {
"name": "task_create",
"description": """创建任务节点。用于跟踪连续性任务,维持对话连贯性。
【使用示例】
1. 创建成语接龙游戏任务:
{
"task_id": "Task_成语接龙",
"description": "用户发起成语接龙游戏,当前成语:为所欲为",
"info_nodes": ["成语接龙_当前成语"]
}
2. 创建编程学习任务:
{
"task_id": "Task_Python学习",
"description": "用户正在学习Python,当前主题:装饰器",
"info_nodes": ["Python学习_当前主题"]
}
3. 创建简单对话任务(每轮必须):
{
"task_id": "Task_当前轮次",
"description": "本轮对话的简要概述"
}
【重要】工作记忆链机制:
- 每轮对话结束时必须创建任务节点
- 任务节点通过 NEXT_TASK 边形成时间链
- 任务节点通过 HAS_STATE 边指向状态节点
- 任务节点通过 CONTAINS_INFO 边指向信息节点
- info_nodes 参数用于关联具体信息节点
【完整流程示例】
用户: "咱来玩成语接龙吧,我先开始,为所欲为"
AI操作步骤:
1. 查询人设图 → 获取当前人设
2. 查询工作记忆链 → 无进行中任务
3. 使用 memory_commit 记录游戏状态:
{"triplets": [
{"subject": "成语接龙_当前成语", "relation": "内容", "object": "为所欲为"},
{"subject": "成语接龙_当前成语", "relation": "游戏", "object": "成语接龙"}
]}
4. 使用 task_create 创建任务节点:
{"task_id": "Task_成语接龙", "description": "成语接龙游戏,当前成语:为所欲为", "info_nodes": ["成语接龙_当前成语"]}
5. 回复: "好的喵!我接:为虎作伥喵!" """,
"parameters": {
"type": "object",
"properties": {
"task_id": {
"type": "string",
"description": "任务ID(如:Task_001)"
},
"description": {
"type": "string",
"description": "任务概述"
},
"info_nodes": {
"type": "array",
"items": {"type": "string"},
"description": "关联的信息节点名称(可选)"
}
},
"required": ["task_id", "description"]
}
}
},
{
"type": "function",
"function": {
"name": "task_set_state",
"description": """设置任务状态。支持:进行中、已完成、已暂停、已取消。
【使用示例】
1. 标记任务为进行中:
{"task_id": "Task_成语接龙", "state": "进行中"}
2. 标记任务为已完成:
{"task_id": "Task_成语接龙", "state": "已完成"}
3. 暂停任务(话题被打断时):
{"task_id": "Task_成语接龙", "state": "已暂停"}
4. 取消任务:
{"task_id": "Task_成语接龙", "state": "已取消"}
【重要】状态转换场景:
- 进行中 → 已暂停: 话题被打断时
- 进行中 → 已完成: 任务完成时
- 已暂停 → 进行中: 任务恢复时
- 进行中 → 已取消: 任务被取消时
【完整流程示例】
用户: "关于刚才的成语接龙,我并不知道应该怎么接你的成语,请帮我接一下"
AI操作步骤:
1. 查询人设图 → 获取当前人设
2. 查询工作记忆链 → 发现 Task_成语接龙 状态为"已暂停"
3. 使用 task_set_state 恢复任务:
{"task_id": "Task_成语接龙", "state": "进行中"}
4. 查询 Task_成语接龙 的信息节点 → 获取当前成语"为虎作伥"
5. 回复: "好的喵!上一个成语是'为虎作伥',我帮你接:伥鬼害人喵!" """,
"parameters": {
"type": "object",
"properties": {
"task_id": {
"type": "string",
"description": "任务ID"
},
"state": {
"type": "string",
"enum": ["进行中", "已完成", "已暂停", "已取消"],
"description": "任务状态"
}
},
"required": ["task_id", "state"]
}
}
},
{
"type": "function",
"function": {
"name": "task_delete",
"description": "删除任务节点。同时删除关联的信息节点。",
"parameters": {
"type": "object",
"properties": {
"task_id": {
"type": "string",
"description": "任务ID"
},
"delete_info_nodes": {
"type": "boolean",
"description": "是否删除关联的信息节点",
"default": True
}
},
"required": ["task_id"]
}
}
},
{
"type": "function",
"function": {
"name": "task_link_info",
"description": """关联信息节点。将记忆节点关联到任务节点,用于存储任务的具体信息。
【使用示例】
1. 关联游戏状态到任务:
{"task_id": "Task_成语接龙", "info_node_names": ["成语接龙_当前成语", "成语接龙_上一个成语"]}
2. 关联学习主题到任务:
{"task_id": "Task_Python学习", "info_node_names": ["Python学习_当前主题", "Python学习_学习进度"]}
3. 关联项目信息到任务:
{"task_id": "Task_项目开发", "info_node_names": ["项目A_技术栈", "项目A_当前阶段"]}
【重要】使用场景:
- 先使用 memory_commit 创建信息节点
- 再使用 task_link_info 将信息节点关联到任务节点
- 信息节点通过 CONTAINS_INFO 边与任务节点连接
【完整流程示例】
用户: "咱来玩成语接龙吧,我先开始,为所欲为"
AI操作步骤:
1. 查询人设图 → 获取当前人设
2. 查询工作记忆链 → 无进行中任务
3. 使用 memory_commit 创建信息节点:
{"triplets": [
{"subject": "成语接龙_当前成语", "relation": "内容", "object": "为所欲为"},
{"subject": "成语接龙_当前成语", "relation": "游戏", "object": "成语接龙"}
]}
4. 使用 task_create 创建任务节点:
{"task_id": "Task_成语接龙", "description": "成语接龙游戏"}
5. 使用 task_link_info 关联信息节点:
{"task_id": "Task_成语接龙", "info_node_names": ["成语接龙_当前成语"]}
6. 回复: "好的喵!我接:为虎作伥喵!" """,
"parameters": {
"type": "object",
"properties": {
"task_id": {
"type": "string",
"description": "任务ID"
},
"info_node_names": {
"type": "array",
"items": {"type": "string"},
"description": "信息节点名称列表"
}
},
"required": ["task_id", "info_node_names"]
}
}
}
]
# 所有工具
TOOLS = MEMORY_TOOLS + PERSONA_TOOLS + WORKING_MEMORY_TOOLS

307
core/tools/tool_executor.py Normal file
View File

@ -0,0 +1,307 @@
"""
工具执行器
"""
import json
from typing import Any, Dict
def execute_tool(graph: Any, tool_name: str, arguments: dict) -> str:
"""执行工具调用"""
print(f"\n[工具调用] {tool_name}")
print(f"[参数] {json.dumps(arguments, ensure_ascii=False, indent=2)}")
try:
# 基础记忆工具
if tool_name == "memory_recall":
result = graph.recall(
query_intent=arguments.get("query_intent", ""),
seed_entities=arguments.get("seed_entities"),
depth=arguments.get("depth", 2),
time_range=arguments.get("time_range"),
session_filter=arguments.get("session_filter")
)
return format_recall_result(result)
elif tool_name == "memory_commit":
result = graph.commit(
triplets=arguments.get("triplets", []),
entity_types=arguments.get("entity_types"),
temporal_tag=arguments.get("temporal_tag")
)
return json.dumps(result, ensure_ascii=False, default=str)
elif tool_name == "memory_purge":
result = graph.purge(
criteria=arguments.get("criteria", {}),
mode=arguments.get("mode", "soft"),
new_relation=arguments.get("new_relation")
)
return json.dumps(result, ensure_ascii=False, default=str)
elif tool_name == "memory_introspect":
result = graph.introspect(session_id=arguments.get("session_id"))
return json.dumps(result, ensure_ascii=False, default=str)
elif tool_name == "memory_archive":
result = graph.archive(days=arguments.get("days", 30))
return json.dumps(result, ensure_ascii=False, default=str)
elif tool_name == "memory_cleanup":
result = graph.cleanup(dry_run=arguments.get("dry_run", True))
return json.dumps(result, ensure_ascii=False, default=str)
# 人设图管理工具
elif tool_name == "persona_update":
result = execute_persona_update(graph, arguments)
return json.dumps(result, ensure_ascii=False, default=str)
elif tool_name == "persona_clear":
result = execute_persona_clear(graph, arguments)
return json.dumps(result, ensure_ascii=False, default=str)
# 工作记忆链管理工具
elif tool_name == "task_create":
result = execute_task_create(graph, arguments)
return json.dumps(result, ensure_ascii=False, default=str)
elif tool_name == "task_set_state":
result = execute_task_set_state(graph, arguments)
return json.dumps(result, ensure_ascii=False, default=str)
elif tool_name == "task_delete":
result = execute_task_delete(graph, arguments)
return json.dumps(result, ensure_ascii=False, default=str)
elif tool_name == "task_link_info":
result = execute_task_link_info(graph, arguments)
return json.dumps(result, ensure_ascii=False, default=str)
return f"未知工具: {tool_name}"
except Exception as e:
return f"工具执行错误: {str(e)}"
def format_recall_result(result: dict) -> str:
"""格式化检索结果"""
lines = ["===== 记忆检索结果 ====="]
if result.get("entities"):
lines.append(f"\n实体 ({len(result['entities'])} 个):")
for e in result["entities"]:
if e and isinstance(e, dict):
lines.append(f" - {e.get('name', 'N/A')} (类型: {e.get('type', 'unknown')}, 提及: {e.get('mention_count', 1)}次)")
if result.get("relations"):
lines.append(f"\n关系 ({len(result['relations'])} 条):")
for r in result["relations"]:
if r and isinstance(r, dict):
lines.append(f" - {r.get('source', 'N/A')} --[{r.get('type', 'N/A')}]--> {r.get('target', 'N/A')}")
created = r.get("created_at", "N/A")
if created and created != "N/A":
created = created[:19] if "T" in str(created) else str(created)
session_id = r.get('session_id', 'N/A')
session_display = session_id[:20] if session_id and session_id != 'N/A' else 'N/A'
lines.append(f" 时间: {created}, 会话: {session_display}, 轮次: {r.get('turn_id', 0)}, 置信度: {r.get('confidence', 1.0)}")
if not result.get("entities") and not result.get("relations"):
lines.append("\n(未找到相关记忆)")
lines.append("=" * 30)
return "\n".join(lines)
# 人设图管理工具实现
def execute_persona_update(graph: Any, arguments: dict) -> dict:
"""更新人设"""
attributes = arguments.get("attributes", [])
mode = arguments.get("mode", "merge")
if mode == "replace":
# 先清除旧人设
graph.purge(
criteria={"subject_contains": "AI", "relation_type": "扮演角色"},
mode="soft"
)
graph.purge(
criteria={"subject_contains": "AI", "relation_type": "说话风格"},
mode="soft"
)
graph.purge(
criteria={"subject_contains": "AI", "relation_type": "性格特点"},
mode="soft"
)
# 写入新人设
triplets = []
for attr in attributes:
triplets.append({
"subject": "AI",
"relation": attr["attribute"],
"object": attr["value"],
"confidence": 1.0
})
result = graph.commit(triplets=triplets)
return {
"status": "success",
"mode": mode,
"updated_attributes": len(attributes),
"details": result
}
def execute_persona_clear(graph: Any, arguments: dict) -> dict:
"""清除人设"""
if not arguments.get("confirm", True):
return {"status": "cancelled", "message": "需要确认才能清除人设"}
# 删除所有人设相关关系
result1 = graph.purge(
criteria={"subject_contains": "AI", "relation_type": "扮演角色"},
mode="soft"
)
result2 = graph.purge(
criteria={"subject_contains": "AI", "relation_type": "说话风格"},
mode="soft"
)
result3 = graph.purge(
criteria={"subject_contains": "AI", "relation_type": "性格特点"},
mode="soft"
)
result4 = graph.purge(
criteria={"subject_contains": "AI", "relation_type": "语气特征"},
mode="soft"
)
total_deleted = (
result1.get("deleted_count", 0) +
result2.get("deleted_count", 0) +
result3.get("deleted_count", 0) +
result4.get("deleted_count", 0)
)
return {
"status": "success",
"deleted_count": total_deleted,
"message": "人设已清除,恢复默认身份"
}
# 工作记忆链管理工具实现
def execute_task_create(graph: Any, arguments: dict) -> dict:
"""创建任务节点"""
task_id = arguments.get("task_id")
description = arguments.get("description")
info_nodes = arguments.get("info_nodes", [])
# 创建任务节点
triplets = [
{"subject": task_id, "relation": "is_type", "object": "TaskNode"},
{"subject": task_id, "relation": "has_description", "object": description},
{"subject": task_id, "relation": "HAS_STATE", "object": "State_进行中"}
]
result = graph.commit(triplets=triplets)
# 关联信息节点
if info_nodes:
link_triplets = []
for node_name in info_nodes:
link_triplets.append({
"subject": task_id,
"relation": "CONTAINS_INFO",
"object": node_name
})
graph.commit(triplets=link_triplets)
return {
"status": "success",
"task_id": task_id,
"description": description,
"info_nodes": info_nodes,
"details": result
}
def execute_task_set_state(graph: Any, arguments: dict) -> dict:
"""设置任务状态"""
task_id = arguments.get("task_id")
state = arguments.get("state")
# 删除旧状态
graph.purge(
criteria={"subject_contains": task_id, "relation_type": "HAS_STATE"},
mode="soft"
)
# 设置新状态
state_node = f"State_{state}"
result = graph.commit(
triplets=[{"subject": task_id, "relation": "HAS_STATE", "object": state_node}]
)
return {
"status": "success",
"task_id": task_id,
"new_state": state,
"details": result
}
def execute_task_delete(graph: Any, arguments: dict) -> dict:
"""删除任务节点"""
task_id = arguments.get("task_id")
delete_info_nodes = arguments.get("delete_info_nodes", True)
# 查询关联的信息节点
if delete_info_nodes:
recall_result = graph.recall(
query_intent=f"{task_id},CONTAINS_INFO",
depth=1
)
# 删除信息节点
for relation in recall_result.get("relations", []):
if relation.get("type") == "CONTAINS_INFO" and relation.get("source") == task_id:
info_node = relation.get("target")
graph.purge(
criteria={"subject_contains": info_node},
mode="soft"
)
# 删除任务节点
result = graph.purge(
criteria={"subject_contains": task_id},
mode="soft"
)
return {
"status": "success",
"task_id": task_id,
"deleted_info_nodes": delete_info_nodes,
"details": result
}
def execute_task_link_info(graph: Any, arguments: dict) -> dict:
"""关联信息节点"""
task_id = arguments.get("task_id")
info_node_names = arguments.get("info_node_names", [])
triplets = []
for node_name in info_node_names:
triplets.append({
"subject": task_id,
"relation": "CONTAINS_INFO",
"object": node_name
})
result = graph.commit(triplets=triplets)
return {
"status": "success",
"task_id": task_id,
"linked_nodes": info_node_names,
"details": result
}

164
core/tools/tool_limiter.py Normal file
View File

@ -0,0 +1,164 @@
"""
工具调用限制器 - 限制每轮对话中各类工具的调用次数
"""
from typing import Dict, List, Optional
from dataclasses import dataclass, field
@dataclass
class ToolLimits:
"""工具调用限制配置"""
# 人设图限制
persona_query_max: int = 1 # 每轮最多查询1次人设图
persona_update_max: int = 1 # 每轮最多修改1次人设图
# 工作记忆链限制
task_query_max: int = 4 # 每轮最多查询4次工作记忆链
task_update_max: int = 2 # 每轮最多修改2次工作记忆链
# 一般记忆限制
memory_query_max: int = 20 # 每轮最多查询20次一般记忆
memory_update_max: int = 10 # 每轮最多修改10次一般记忆
@dataclass
class ToolCallCount:
"""工具调用计数"""
# 人设图
persona_query: int = 0
persona_update: int = 0
# 工作记忆链
task_query: int = 0
task_update: int = 0
# 一般记忆
memory_query: int = 0
memory_update: int = 0
class ToolLimiter:
"""工具调用限制器"""
def __init__(self, limits: Optional[ToolLimits] = None):
self.limits = limits or ToolLimits()
self.counts = ToolCallCount()
def _classify_tool(self, tool_name: str, arguments: dict) -> tuple:
"""
分类工具调用
返回: (category, operation)
category: 'persona', 'task', 'memory'
operation: 'query', 'update'
"""
# 人设图工具
if tool_name in ('persona_update', 'persona_clear'):
return ('persona', 'update')
# 工作记忆链工具
if tool_name in ('task_create', 'task_set_state', 'task_delete', 'task_link_info'):
# task_link_info 是关联操作,算作更新
return ('task', 'update')
# 一般记忆工具
if tool_name == 'memory_recall':
# 判断是查询人设图、工作记忆链还是一般记忆
query_intent = arguments.get('query_intent', '').lower()
# 检查是否查询人设图
if any(kw in query_intent for kw in ['人设', '角色', '性格', '语气', '说话风格', '扮演']):
return ('persona', 'query')
# 检查是否查询工作记忆链
if any(kw in query_intent for kw in ['tasknode', '工作记忆', '任务链', '任务', 'task']):
return ('task', 'query')
# 一般记忆查询
return ('memory', 'query')
if tool_name == 'memory_commit':
return ('memory', 'update')
if tool_name == 'memory_purge':
return ('memory', 'update')
if tool_name == 'memory_introspect':
return ('memory', 'query')
if tool_name in ('memory_archive', 'memory_cleanup'):
return ('memory', 'update')
# 未知工具,归类为一般记忆更新
return ('memory', 'update')
def can_call(self, tool_name: str, arguments: dict) -> tuple:
"""
检查是否允许调用工具
返回: (allowed, reason)
"""
category, operation = self._classify_tool(tool_name, arguments)
# 获取当前计数和限制
if category == 'persona':
if operation == 'query':
if self.counts.persona_query >= self.limits.persona_query_max:
return (False, f"人设图查询次数已达上限({self.limits.persona_query_max}次)")
else: # update
if self.counts.persona_update >= self.limits.persona_update_max:
return (False, f"人设图修改次数已达上限({self.limits.persona_update_max}次)")
elif category == 'task':
if operation == 'query':
if self.counts.task_query >= self.limits.task_query_max:
return (False, f"工作记忆链查询次数已达上限({self.limits.task_query_max}次)")
else: # update
if self.counts.task_update >= self.limits.task_update_max:
return (False, f"工作记忆链修改次数已达上限({self.limits.task_update_max}次)")
elif category == 'memory':
if operation == 'query':
if self.counts.memory_query >= self.limits.memory_query_max:
return (False, f"一般记忆查询次数已达上限({self.limits.memory_query_max}次)")
else: # update
if self.counts.memory_update >= self.limits.memory_update_max:
return (False, f"一般记忆修改次数已达上限({self.limits.memory_update_max}次)")
return (True, "允许调用")
def record_call(self, tool_name: str, arguments: dict) -> None:
"""记录工具调用"""
category, operation = self._classify_tool(tool_name, arguments)
if category == 'persona':
if operation == 'query':
self.counts.persona_query += 1
else:
self.counts.persona_update += 1
elif category == 'task':
if operation == 'query':
self.counts.task_query += 1
else:
self.counts.task_update += 1
elif category == 'memory':
if operation == 'query':
self.counts.memory_query += 1
else:
self.counts.memory_update += 1
def get_summary(self) -> str:
"""获取调用统计摘要"""
lines = [
f"人设图: 查询{self.counts.persona_query}/{self.limits.persona_query_max}次, "
f"修改{self.counts.persona_update}/{self.limits.persona_update_max}次",
f"工作记忆链: 查询{self.counts.task_query}/{self.limits.task_query_max}次, "
f"修改{self.counts.task_update}/{self.limits.task_update_max}次",
f"一般记忆: 查询{self.counts.memory_query}/{self.limits.memory_query_max}次, "
f"修改{self.counts.memory_update}/{self.limits.memory_update_max}次"
]
return "\n".join(lines)
def reset(self) -> None:
"""重置计数(新的一轮对话开始时调用)"""
self.counts = ToolCallCount()