feat: 增加两个alter的支持

讲真这个实现得略抽象 但勉强能用,,
This commit is contained in:
ChensenCHX
2025-03-17 20:14:59 +08:00
parent ab32464609
commit c1bed2f2c4

View File

@@ -111,7 +111,7 @@ def remove_mem_node(hippocampus: Hippocampus):
console.print(f"[yellow]存在边“{edge['source']} -> {edge['target']}”, 请慎重考虑[/yellow]")
console.print(f"[yellow]确定要移除名为“{concept}”的节点以及其相关边吗[/yellow]")
destory = console.input(f"[orange]请输入“{concept}”以删除节点 其他输入将被视为取消操作[/orange]\n")
destory = console.input(f"[red]请输入“{concept}”以删除节点 其他输入将被视为取消操作[/red]\n")
if destory == concept:
hippocampus.memory_graph.G.remove_node(concept)
else:
@@ -122,16 +122,16 @@ def add_mem_edge(hippocampus: Hippocampus):
source = input("请输入 **第一个节点** 名称(输入'退出'以结束):\n")
if source.lower() == "退出": break
if db.graph_data.nodes.count_documents({'concept': source}) == 0:
console.print("[yellow]“{source}”节点不存在,操作已取消。[/yellow]")
console.print(f"[yellow]“{source}”节点不存在,操作已取消。[/yellow]")
continue
target = input("请输入 **第二个节点** 名称:\n")
if db.graph_data.nodes.count_documents({'concept': target}) == 0:
console.print("[yellow]“{target}”节点不存在,操作已取消。[/yellow]")
console.print(f"[yellow]“{target}”节点不存在,操作已取消。[/yellow]")
continue
if source == target:
console.print("[yellow]试图创建“{source} <-> {target}”自环,操作已取消。[/yellow]")
console.print(f"[yellow]试图创建“{source} <-> {target}”自环,操作已取消。[/yellow]")
continue
hippocampus.memory_graph.connect_dot(source, target)
@@ -167,16 +167,109 @@ def remove_mem_edge(hippocampus: Hippocampus):
if accept.lower() == "确认":
hippocampus.memory_graph.G.remove_edge(source, target)
console.print(f"[green]边“{source} <-> {target}”已删除。[green]")
# 修改节点信息
def alter_mem_node(hippocampus: Hippocampus):
#todo...
#需要允许修改memory_items, last_modified
return
batchEnviroment = dict()
while True:
concept = input("请输入节点概念名(输入'终止'以结束):\n")
if concept.lower() == "终止": break
_, node = hippocampus.memory_graph.get_dot(concept)
if node is None:
console.print(f"[yellow]“{concept}”节点不存在,操作已取消。[/yellow]")
continue
console.print("[yellow]注意,请确保你知道自己在做什么[/yellow]")
console.print("[yellow]你将获得一个执行任意代码的环境[/yellow]")
console.print("[red]你已经被警告过了。[/red]\n")
nodeEnviroment = {"concept": '<节点名>', 'memory_items': '<记忆文本数组>'}
console.print("[green]环境变量中会有env与batchEnv两个dict, env在切换节点时会清空, batchEnv在操作终止时才会清空[/green]")
console.print(f"[green] env 会被初始化为[/green]\n{nodeEnviroment}\n[green]且会在用户代码执行完毕后被提交 [/green]")
console.print("[yellow]为便于书写临时脚本请手动在输入代码通过Ctrl+C等方式触发KeyboardInterrupt来结束代码执行[/yellow]")
# 拷贝数据以防操作炸了
nodeEnviroment = dict(node)
nodeEnviroment['concept'] = concept
while True:
userexec = lambda script, env, batchEnv: eval(script)
try:
command = console.input()
except KeyboardInterrupt:
# 稍微防一下小天才
try:
if isinstance(nodeEnviroment['memory_items'], list):
node['memory_items'] = nodeEnviroment['memory_items']
else:
raise Exception
except:
console.print("[red]我不知道你做了什么但显然nodeEnviroment['memory_items']已经不是个数组了,操作已取消[/red]")
break
try:
userexec(command, nodeEnviroment, batchEnviroment)
except Exception as e:
console.print(e)
console.print("[red]自定义代码执行时发生异常,已捕获,请重试(可通过 console.print(locals()) 检查环境状态)[/red]")
# 修改边信息
def alter_mem_edge(hippocampus: Hippocampus):
#todo...
#需要允许修改strength, last_modified
return
batchEnviroment = dict()
while True:
source = input("请输入 **第一个节点** 名称(输入'终止'以结束):\n")
if source.lower() == "终止": break
if hippocampus.memory_graph.get_dot(source) is None:
console.print(f"[yellow]“{source}”节点不存在,操作已取消。[/yellow]")
continue
target = input("请输入 **第二个节点** 名称:\n")
if hippocampus.memory_graph.get_dot(target) is None:
console.print(f"[yellow]“{target}”节点不存在,操作已取消。[/yellow]")
continue
edge = hippocampus.memory_graph.G.get_edge_data(source, target)
if edge is None:
console.print(f"[yellow]边“{source} <-> {target}”不存在,操作已取消。[/yellow]")
continue
console.print("[yellow]注意,请确保你知道自己在做什么[/yellow]")
console.print("[yellow]你将获得一个执行任意代码的环境[/yellow]")
console.print("[red]你已经被警告过了。[/red]\n")
edgeEnviroment = {"source": '<节点名>', "target": '<节点名>', 'strength': '<强度值,装在一个list里>'}
console.print("[green]环境变量中会有env与batchEnv两个dict, env在切换节点时会清空, batchEnv在操作终止时才会清空[/green]")
console.print(f"[green] env 会被初始化为[/green]\n{edgeEnviroment}\n[green]且会在用户代码执行完毕后被提交 [/green]")
console.print("[yellow]为便于书写临时脚本请手动在输入代码通过Ctrl+C等方式触发KeyboardInterrupt来结束代码执行[/yellow]")
# 拷贝数据以防操作炸了
edgeEnviroment['strength'] = [edge["strength"]]
edgeEnviroment['source'] = source
edgeEnviroment['target'] = target
while True:
userexec = lambda script, env, batchEnv: eval(script)
try:
command = console.input()
except KeyboardInterrupt:
# 稍微防一下小天才
try:
if isinstance(edgeEnviroment['strength'][0], int):
edge['strength'] = edgeEnviroment['strength'][0]
else:
raise Exception
except:
console.print("[red]我不知道你做了什么但显然edgeEnviroment['strength']已经不是个int了操作已取消[/red]")
break
try:
userexec(command, edgeEnviroment, batchEnviroment)
except Exception as e:
console.print(e)
console.print("[red]自定义代码执行时发生异常,已捕获,请重试(可通过 console.print(locals()) 检查环境状态)[/red]")
async def main():
start_time = time.time()
@@ -194,7 +287,10 @@ async def main():
logger.info(f"\033[32m[加载海马体耗时: {end_time - start_time:.2f} 秒]\033[0m")
while True:
query = int(input("请输入操作类型\n0 -> 查询节点; 1 -> 增加节点; 2 -> 移除节点; 3 -> 增加边; 4 -> 移除边;\n其他任意输入 -> 退出\n"))
try:
query = int(input("请输入操作类型\n0 -> 查询节点; 1 -> 增加节点; 2 -> 移除节点; 3 -> 增加边; 4 -> 移除边;\n5 -> 修改节点; 6 -> 修改边; 其他任意输入 -> 退出\n"))
except:
query = -1
if query == 0:
query_mem_info(memory_graph)
@@ -207,9 +303,9 @@ async def main():
elif query == 4:
remove_mem_edge(hippocampus)
elif query == 5:
continue
alter_mem_node(hippocampus)
elif query == 6:
continue
alter_mem_edge(hippocampus)
else:
print("已结束操作")
break