691 lines
30 KiB
Python
691 lines
30 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
OpenClaw 纯图数据库调用 Demo - Neo4j 真实数据库版本
|
||
基于 DeepSeek API Tool Calls 实现摒弃传统上下文的自主记忆多轮对话
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import uuid
|
||
from datetime import datetime
|
||
|
||
DEEPSEEK_API_KEY = "sk-ca05d81474f04d9e9905fb0696f07550"
|
||
DEEPSEEK_BASE_URL = "https://api.deepseek.com"
|
||
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
|
||
|
||
TOOLS = [
|
||
{
|
||
"type": "function",
|
||
"function": {
|
||
"name": "memory_recall",
|
||
"description": "当你对当前输入中的实体、指代、或关系不确定时，调用此工具检索相关记忆。输入为自然语言查询意图，系统将返回相关子图。支持时间范围和多跳路径查询。",
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {
|
||
"query_intent": {"type": "string", "description": "模型用自然语言描述想查什么"},
|
||
"seed_entities": {"type": "array", "items": {"type": "string"}, "description": "可选：已识别的实体ID"},
|
||
"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": "当你认为当前对话包含对未来轮次有价值的信息时，将抽取的三元组写入图库。仅在信息具有跨轮次引用潜力时调用。支持批量写入。",
|
||
"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": "可选：时间标记（如'2026-04-08'）"}
|
||
},
|
||
"required": ["triplets"]
|
||
}
|
||
}
|
||
},
|
||
{
|
||
"type": "function",
|
||
"function": {
|
||
"name": "memory_purge",
|
||
"description": "当你发现记忆中的信息与当前认知矛盾，或用户明确要求更正时，删除指定关系。优先于memory_commit执行以维护一致性。",
|
||
"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": []
|
||
}
|
||
}
|
||
}
|
||
]
|
||
|
||
|
||
class Neo4jGraph:
|
||
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 for w in query_intent.lower().split() if len(w) > 2]
|
||
|
||
params = {"session_id": CURRENT_SESSION_ID}
|
||
cond_parts = ["r.status = 'active'"]
|
||
|
||
if session_filter:
|
||
cond_parts.append("r.session_id = $target_session")
|
||
params["target_session"] = session_filter
|
||
|
||
if keywords:
|
||
cond_parts.append("ANY(k IN $keywords WHERE toLower(e.name) CONTAINS k OR toLower(e.type) CONTAINS k)")
|
||
params["keywords"] = keywords
|
||
|
||
if seed_entities:
|
||
cond_parts.append("e.name IN $seed")
|
||
params["seed"] = seed_entities
|
||
|
||
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)
|
||
|
||
if depth > 1:
|
||
cypher = f"""
|
||
MATCH path = (e:Entity)-[r:RELATES*1..{depth}]-(other:Entity)
|
||
WHERE {where_clause}
|
||
UNWIND relationships(path) AS rel
|
||
WITH DISTINCT rel, e, other
|
||
RETURN e, rel, other
|
||
ORDER BY rel.created_at DESC
|
||
LIMIT 30
|
||
"""
|
||
else:
|
||
cypher = f"""
|
||
MATCH (e:Entity)-[r:RELATES]-(other:Entity)
|
||
WHERE {where_clause}
|
||
RETURN e, r, other
|
||
ORDER BY r.created_at DESC
|
||
LIMIT 20
|
||
"""
|
||
|
||
result = session.run(cypher, params)
|
||
entities, relations = {}, []
|
||
|
||
for record in result:
|
||
e, r, other = record["e"], record["r"], record["other"]
|
||
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 other["name"] not in entities:
|
||
entities[other["name"]] = {"name": other["name"], "type": other.get("type", "unknown"), "mention_count": other.get("mention_count", 1)}
|
||
|
||
relations.append({
|
||
"source": e["name"],
|
||
"target": other["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("r.source CONTAINS $subject")
|
||
params["subject"] = subject_pattern
|
||
if target_pattern:
|
||
cond_parts.append("r.target 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 id(r) as old_id, s.name as source
|
||
""", params)
|
||
|
||
deleted_count = 0
|
||
for record in result:
|
||
old_id = record["old_id"]
|
||
source = record["source"]
|
||
|
||
session.run("""
|
||
MATCH (s:Entity {name: $source})
|
||
WHERE id(s) = $source_id
|
||
CREATE (s)-[r:RELATES {
|
||
type: $new_rel,
|
||
created_at: datetime(),
|
||
session_id: $session_id,
|
||
turn_id: $turn_id,
|
||
role: 'user',
|
||
status: 'active',
|
||
confidence: 0.9,
|
||
date_bucket: date().isoDate,
|
||
supersedes: $old_id
|
||
}]->(t:Entity {name: $new_target})
|
||
""", source_id=record["s"].element_id, new_rel=new_rel, new_target=new_target,
|
||
session_id=CURRENT_SESSION_ID, turn_id=CURRENT_TURN, old_id=old_id)
|
||
deleted_count += 1
|
||
|
||
return {"deleted_count": deleted_count, "mode": "supersede", "new_relation": f"{new_relation.get('subject', '')} -[{new_rel}]-> {new_target}"}
|
||
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]
|
||
|
||
result3 = session.run("""
|
||
MATCH ()-[r:RELATES]->()
|
||
WHERE r.session_id = $session_id
|
||
RETURN r.type as type, count(*) as count
|
||
ORDER BY count DESC
|
||
""", session_id=target_session)
|
||
relation_distribution = {r["type"]: r["count"] for r in result3}
|
||
|
||
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,
|
||
"relation_distribution": relation_distribution
|
||
}
|
||
|
||
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
|
||
}
|
||
|
||
|
||
def execute_tool(graph: Neo4jGraph, 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)
|
||
|
||
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"]:
|
||
lines.append(f" - {e['name']} (类型: {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"]:
|
||
lines.append(f" - {r['source']} --[{r['type']}]--> {r['target']}")
|
||
created = r.get("created_at", "N/A")
|
||
if created and created != "N/A":
|
||
created = created[:19] if "T" in str(created) else str(created)
|
||
lines.append(f" 时间: {created}, 会话: {r.get('session_id', 'N/A')[:20]}, 轮次: {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("=" * 35)
|
||
return "\n".join(lines)
|
||
|
||
|
||
class OpenClawClient:
|
||
def __init__(self, api_key: str, base_url: str, graph: Neo4jGraph):
|
||
from openai import OpenAI
|
||
self.client = OpenAI(api_key=api_key, base_url=base_url)
|
||
self.graph = graph
|
||
self.tools = TOOLS
|
||
self.system_prompt = self._build_system_prompt()
|
||
|
||
def _build_system_prompt(self) -> str:
|
||
return """你是 OpenClaw 的自主记忆控制器。
|
||
|
||
## 核心职责
|
||
你是用户的长期记忆助手。每次对话后，你必须主动决定是否需要将本次对话的关键信息写入记忆图库。
|
||
|
||
## 可用工具
|
||
1. **memory_recall**: 检索历史记忆
|
||
- 当用户提到"之前"、"上次"、"那个"等指代时
|
||
- 当你不确定用户偏好或事实时
|
||
- 当需要验证历史信息时
|
||
|
||
2. **memory_commit**: 写入记忆（三元组）
|
||
- 当用户明确表达偏好（"我喜欢X"、"我偏好Y"）
|
||
- 当讨论重要的实体或项目（"我在做项目Z"）
|
||
- 当产生关键结论或决定时
|
||
- **不要**写入：寒暄、临时例子、重复信息
|
||
|
||
3. **memory_purge**: 修正记忆
|
||
- 当用户说"不是"、"其实"、"更正"时
|
||
- 当发现与记忆矛盾时
|
||
|
||
4. **memory_introspect**: 自省
|
||
- 当需要了解当前会话状态时
|
||
- 当需要检查信息缺口时
|
||
|
||
## 强制规则
|
||
1. 每轮对话结束前，根据对话内容决定是否调用 memory_commit
|
||
2. 如果对话涉及重要信息（如项目、偏好、决定），必须调用 memory_commit
|
||
3. 如果用户提到之前的话题但你不确定内容，必须先调用 memory_recall
|
||
4. 调用工具时，提供具体的参数，不要空调用
|
||
|
||
## 示例
|
||
用户: "我喜欢Python" -> 写入: memory_commit(主题=用户, 关系=喜欢, 对象=Python)
|
||
用户: "上次我们说的项目怎样了" -> 查询: memory_recall(query_intent="项目")"""
|
||
|
||
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}]
|
||
|
||
# 添加之前的 assistant 消息（包含 tool_calls）
|
||
if assistant_msg:
|
||
messages.append(assistant_msg)
|
||
|
||
# 添加之前的工具结果
|
||
if tool_results:
|
||
messages.extend(tool_results)
|
||
|
||
# 添加当前用户输入
|
||
messages.append({"role": "user", "content": user_input})
|
||
|
||
response = self.client.chat.completions.create(
|
||
model=MODEL_NAME,
|
||
messages=messages,
|
||
tools=self.tools,
|
||
tool_choice="auto"
|
||
)
|
||
|
||
return response
|
||
|
||
def chat_loop(self):
|
||
global CURRENT_TURN
|
||
|
||
print("\n" + "=" * 60)
|
||
print("OpenClaw 纯图数据库对话 Demo (Neo4j)")
|
||
print("=" * 60)
|
||
print(f"会话ID: {CURRENT_SESSION_ID}")
|
||
print(f"Neo4j: {NEO4J_URI}")
|
||
print("输入 quit/exit 退出")
|
||
print("=" * 60 + "\n")
|
||
|
||
while True:
|
||
try:
|
||
user_input = input("\n[你] ").strip()
|
||
if not user_input:
|
||
continue
|
||
if user_input.lower() in ["quit", "exit", "退出"]:
|
||
print("\n[系统] 再见！")
|
||
break
|
||
|
||
CURRENT_TURN += 1
|
||
print(f"\n[轮次 {CURRENT_TURN}] 发送请求...")
|
||
|
||
# 用于累积工具结果
|
||
tool_results = []
|
||
last_assistant_msg = None
|
||
|
||
# 首次请求
|
||
response = self.send_message(user_input, tool_results)
|
||
message = response.choices[0].message
|
||
|
||
# 处理工具调用循环
|
||
while message.tool_calls:
|
||
# 打印模型响应（如果有）
|
||
if message.content:
|
||
print(f"\n[模型] {message.content}")
|
||
|
||
# 保存包含 tool_calls 的 assistant 消息
|
||
last_assistant_msg = {
|
||
"role": "assistant",
|
||
"content": message.content,
|
||
"type": "message",
|
||
"tool_calls": [{"id": tc.id, "type": "function", "function": {"name": tc.function.name, "arguments": tc.function.arguments}} for tc in message.tool_calls]
|
||
}
|
||
|
||
# 执行所有工具调用
|
||
for tool_call in message.tool_calls:
|
||
tool_name = tool_call.function.name
|
||
tool_args = json.loads(tool_call.function.arguments)
|
||
tool_id = tool_call.id
|
||
|
||
result = execute_tool(self.graph, tool_name, tool_args)
|
||
print(f"\n[工具结果] {result}")
|
||
|
||
# 添加工具结果 - 包含 tool_call_id
|
||
tool_results.append({
|
||
"role": "tool",
|
||
"tool_call_id": tool_id,
|
||
"content": result
|
||
})
|
||
|
||
# 继续调用 - 包含 system + assistant消息(with tool_calls) + tool_results + user
|
||
messages = [{"role": "system", "content": self.system_prompt}]
|
||
if last_assistant_msg:
|
||
messages.append(last_assistant_msg)
|
||
messages.extend(tool_results)
|
||
messages.append({"role": "user", "content": user_input})
|
||
|
||
response = self.client.chat.completions.create(
|
||
model=MODEL_NAME,
|
||
messages=messages,
|
||
tools=self.tools
|
||
)
|
||
message = response.choices[0].message
|
||
|
||
# 最终回复
|
||
final_content = message.content or "(无回复)"
|
||
print(f"\n[模型] {final_content}")
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n\n[系统] 中断退出")
|
||
break
|
||
except Exception as e:
|
||
print(f"\n[错误] {str(e)}")
|
||
|
||
|
||
def main():
|
||
print("OpenClaw Demo - Neo4j 纯图数据库调用的多轮对话")
|
||
print("-" * 40)
|
||
|
||
if DEEPSEEK_API_KEY == "your-api-key-here":
|
||
print("请设置 DEEPSEEK_API_KEY 环境变量")
|
||
print(" export DEEPSEEK_API_KEY='your-actual-key'")
|
||
print()
|
||
|
||
print(f"Neo4j 配置: {NEO4J_URI}")
|
||
print("如需修改，请设置环境变量: NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD")
|
||
print()
|
||
|
||
try:
|
||
graph = Neo4jGraph(NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD)
|
||
graph.ensure_constraints()
|
||
print("[Info] Neo4j 连接成功\n")
|
||
except Exception as e:
|
||
print(f"[Error] Neo4j 连接失败: {e}")
|
||
print("请确保 Neo4j 已启动，或运行 scripts/ 下的安装脚本")
|
||
return
|
||
|
||
client = OpenClawClient(DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL, graph)
|
||
client.chat_loop()
|
||
graph.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|