From 5c76b196241c933a9d903254577f61ac8537a4ff Mon Sep 17 00:00:00 2001 From: root Date: Sun, 12 Apr 2026 20:47:07 +0800 Subject: [PATCH] refactor: packet-based IPC with core/ui separation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - All logic in core layer (BackendServer runs independently) - Packet protocol: {id, type, body} for all communications - UI simplified: just input→backend→display - Config, history, tools all handled by core - BackendServer can run standalone without UI --- core/__init__.py | 220 +++++++++++++++++++++++++++++++++++++- tests/test_ui.py | 33 +++--- trulymem_entry.py | 22 +--- ui/app.py | 266 ++++++++++------------------------------------ 4 files changed, 294 insertions(+), 247 deletions(-) diff --git a/core/__init__.py b/core/__init__.py index 0ae06f2..7e7979d 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -1,5 +1,219 @@ -from .server import BackendServer -from .client import BackendClient +import threading +import queue +import time +from typing import Any, Dict +from dataclasses import dataclass, field +from enum import Enum + + +class PacketType(Enum): + MESSAGE = "message" + CONFIG = "config" + TOOL = "tool" + STATUS = "status" + HISTORY = "history" + + +@dataclass +class Packet: + id: str + type: PacketType + body: Dict[str, Any] + created_at: float = field(default_factory=time.time) + + from .embedded_db import EmbeddedGraphDB -__all__ = ["BackendServer", "BackendClient", "EmbeddedGraphDB"] \ No newline at end of file + +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._running = False + self._thread = None + self._input_queue: queue.Queue[Packet] = queue.Queue() + self._response_queues: Dict[str, queue.Queue] = {} + self._lock = threading.Lock() + self._config = {"api_key": "", "base_url": "https://api.deepseek.com"} + + def start(self, api_key: str = "", base_url: str = "https://api.deepseek.com") -> None: + if self._running: + return + self._init_graph() + self._config = {"api_key": api_key, "base_url": base_url} + if api_key: + from .graph_client import GraphMemoryClient + 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: + packet = self._input_queue.get(timeout=0.1) + except queue.Empty: + continue + self._process_packet(packet) + + def _process_packet(self, packet: Packet) -> None: + body = {"success": False, "error": "not implemented"} + try: + if packet.type == PacketType.MESSAGE: + body = self._handle_message(packet.body) + elif packet.type == PacketType.CONFIG: + body = self._handle_config(packet.body) + elif packet.type == PacketType.TOOL: + body = self._handle_tool(packet.body) + elif packet.type == PacketType.STATUS: + body = self._handle_status() + elif packet.type == PacketType.HISTORY: + body = self._handle_history(packet.body) + body["success"] = True + except Exception as e: + body["error"] = str(e) + finally: + self._send_response(packet.id, Packet(id=packet.id, type=packet.type, body=body)) + + def _handle_message(self, body: Dict) -> Dict: + from .tool_executor import execute_tool + from .tool_limiter import ToolLimiter + + limiter = ToolLimiter() + user_input = body.get("message", "") + if not self._client: + return {"error": "API Key not configured", "content": "请先配置 API Key"} + + messages = [{"role": "user", "content": user_input}] + response = self._client.send_message_with_history(messages) + message = response.choices[0].message + content = message.content or "(无��复)" + tool_calls = [] + + while message.tool_calls: + args = {} + try: + import json + args = json.loads(message.tool_calls[0].function.arguments) + except: + pass + + allowed, reason = limiter.can_call(message.tool_calls[0].function.name, args) + if not allowed: + tool_calls.append({"name": message.tool_calls[0].function.name, "result": f"工具调用被拒绝: {reason}"}) + continue + + limiter.record_call(message.tool_calls[0].function.name, args) + result = execute_tool(self._graph, message.tool_calls[0].function.name, args) + tool_calls.append({"name": message.tool_calls[0].function.name, "result": result}) + messages.append({"role": "assistant", "tool_calls": [{"id": "1", "function": {"name": message.tool_calls[0].function.name, "arguments": message.tool_calls[0].function.arguments}}]}) + messages.append({"role": "tool", "tool_call_id": "1", "content": result}) + response = self._client.send_message_with_history(messages) + message = response.choices[0].message + content = message.content or content + + return {"content": content, "tool_calls": tool_calls} + + def _handle_config(self, body: Dict) -> Dict: + api_key = body.get("api_key", "") + base_url = body.get("base_url", "https://api.deepseek.com") + self._config = {"api_key": api_key, "base_url": base_url} + if api_key and self._graph: + from .graph_client import GraphMemoryClient + self._client = GraphMemoryClient(api_key=api_key, base_url=base_url, graph=self._graph) + return {"status": "config_updated"} + + def _handle_tool(self, body: Dict) -> Dict: + from .tool_executor import execute_tool + name = body.get("name", "") + args = body.get("arguments", {}) + result = execute_tool(self._graph, name, args) + return {"result": result} + + def _handle_status(self) -> Dict: + return {"running": self._running, "config": self._config, "client_ready": self._client is not None} + + def _handle_history(self, body: Dict) -> Dict: + action = body.get("action", "get") + if action == "save": + self._message_history = body.get("messages", []) + return {"status": "saved"} + return {"history": getattr(self, "_message_history", [])} + + def send(self, packet: Packet) -> Packet: + resp_q = queue.Queue() + with self._lock: + self._response_queues[packet.id] = resp_q + self._input_queue.put(packet) + try: + return resp_q.get(timeout=30.0) + except queue.Empty: + return Packet(id=packet.id, type=packet.type, body={"success": False, "error": "timeout"}) + finally: + with self._lock: + self._response_queues.pop(packet.id, None) + + def _send_response(self, request_id: str, packet: Packet) -> None: + with self._lock: + q = self._response_queues.pop(request_id, None) + if q: + q.put(packet) + + def shutdown(self) -> None: + self._running = False + if self._thread: + self._thread.join(timeout=2.0) + if self._graph: + self._graph.close() + self._graph = None + + +class BackendClient: + def __init__(self, server: BackendServer): + self._server = server + self._counter = 0 + self._lock = threading.Lock() + + def _next_id(self) -> str: + with self._lock: + self._counter += 1 + return f"{time.time()}_{self._counter}" + + def send_message(self, message: str) -> Dict: + packet = Packet(id=self._next_id(), type=PacketType.MESSAGE, body={"message": message}) + return self._server.send(packet).body + + def update_config(self, api_key: str, base_url: str = "https://api.deepseek.com") -> Dict: + packet = Packet(id=self._next_id(), type=PacketType.CONFIG, body={"api_key": api_key, "base_url": base_url}) + return self._server.send(packet).body + + def execute_tool(self, name: str, arguments: Dict) -> str: + packet = Packet(id=self._next_id(), type=PacketType.TOOL, body={"name": name, "arguments": arguments}) + return self._server.send(packet).body.get("result", "") + + def get_status(self) -> Dict: + packet = Packet(id=self._next_id(), type=PacketType.STATUS, body={}) + return self._server.send(packet).body + + def save_history(self, messages: list) -> None: + packet = Packet(id=self._next_id(), type=PacketType.HISTORY, body={"action": "save", "messages": messages}) + self._server.send(packet) + + def get_history(self) -> list: + packet = Packet(id=self._next_id(), type=PacketType.HISTORY, body={"action": "get"}) + return self._server.send(packet).body.get("history", []) + + def shutdown(self) -> None: + self._server.shutdown() + + +__all__ = ["BackendServer", "BackendClient", "EmbeddedGraphDB", "Packet", "PacketType"] \ No newline at end of file diff --git a/tests/test_ui.py b/tests/test_ui.py index b35b213..216ff42 100644 --- a/tests/test_ui.py +++ b/tests/test_ui.py @@ -10,28 +10,15 @@ class TestUIImport: from ui import GraphMemoryApp assert GraphMemoryApp is not None - def test_import_config(self): - from ui import AppConfig - assert AppConfig is not None - class TestUIApp: - def test_create_app(self): + def test_create_app_without_backend(self): from ui import GraphMemoryApp app = GraphMemoryApp() assert app is not None assert app._backend_server is None assert app._backend_client is None - def test_create_app_with_config(self): - from ui import GraphMemoryApp, AppConfig - - config = AppConfig(api_key="test-key", base_url="https://api.test.com") - app = GraphMemoryApp(config=config) - - assert app._config is config - assert app._config.api_key == "test-key" - def test_create_app_with_backend(self): from ui import GraphMemoryApp from core import BackendServer @@ -57,14 +44,26 @@ class TestUIApp: class TestAppConfig: def test_config_from_env(self): from ui import AppConfig - config = AppConfig.from_env() assert config is not None def test_config_default_values(self): from ui import AppConfig - config = AppConfig() assert config.api_key == "" assert config.model == "deepseek-chat" - assert config.base_url == "https://api.deepseek.com" \ No newline at end of file + assert config.base_url == "https://api.deepseek.com" + + +class TestPacketProtocol: + def test_import_packet(self): + from core import Packet, PacketType + assert Packet is not None + assert PacketType is not None + + def test_packet_creation(self): + from core import Packet, PacketType + packet = Packet(id="test-1", type=PacketType.MESSAGE, body={"message": "hello"}) + assert packet.id == "test-1" + assert packet.type == PacketType.MESSAGE + assert packet.body["message"] == "hello" \ No newline at end of file diff --git a/trulymem_entry.py b/trulymem_entry.py index 3374049..4060352 100644 --- a/trulymem_entry.py +++ b/trulymem_entry.py @@ -1,6 +1,5 @@ #!/usr/bin/env python3 import sys -import os from pathlib import Path if getattr(sys, 'frozen', False): @@ -8,37 +7,26 @@ if getattr(sys, 'frozen', False): else: application_path = Path(__file__).parent +sys.path.insert(0, str(application_path)) +import os os.chdir(application_path) -if str(application_path) not in sys.path: - sys.path.insert(0, str(application_path)) - from core import BackendServer -from ui import GraphMemoryApp, AppConfig +from ui import GraphMemoryApp def main(): backend_server = BackendServer(db_path="graph_memory.db", use_embedded_db=True) - - try: - config = AppConfig.from_env() - backend_server.start(api_key=config.api_key, base_url=config.base_url) - except Exception as e: - print(f"后端启动失败: {e}") + backend_server.start(api_key="", base_url="https://api.deepseek.com") app = GraphMemoryApp(backend_server=backend_server) try: app.run() except KeyboardInterrupt: - print("\n应用已退出") - except Exception as e: - print(f"应用启动失败: {e}") - import traceback - traceback.print_exc() + print("\n退出") finally: backend_server.shutdown() - sys.exit(0) if __name__ == "__main__": diff --git a/ui/app.py b/ui/app.py index 989d08f..29b0c1c 100644 --- a/ui/app.py +++ b/ui/app.py @@ -2,17 +2,12 @@ import asyncio from pathlib import Path from textual.app import App, ComposeResult from textual.binding import Binding -from datetime import datetime from core import BackendServer, BackendClient -from .models.message import Message, ToolCall, ToolResult -from .models.config import AppConfig -from .models.log_entry import LogEntry -from .services.config_manager import ConfigManager -from core import BackendClient +from .models.message import Message -class GraphMemoryApp(App[None]): +class GraphMemoryApp(App): CSS_PATH = [ Path(__file__).parent / "styles" / "app.css", Path(__file__).parent / "styles" / "messages.css", @@ -23,257 +18,108 @@ class GraphMemoryApp(App[None]): Binding("f1", "show_help", "帮助"), Binding("f2", "toggle_sidebar", "侧边栏"), Binding("f3", "toggle_tool_details", "工具详情"), - Binding("f4", "focus_query", "查询"), Binding("f5", "clear_history", "清屏"), Binding("f6", "quit", "退出"), ] - def __init__(self, config: AppConfig | None = None, backend_server: BackendServer | None = None, **kwargs): + def __init__(self, backend_server: BackendServer = None, **kwargs): super().__init__(**kwargs) - self._config_manager = ConfigManager() - - if config: - self._config = config - elif self._config_manager.exists(): - self._config = self._config_manager.load() - else: - self._config = AppConfig.from_env() - - if backend_server: - self._backend_server = backend_server - self._backend_client = BackendClient(backend_server) - else: - self._backend_server: BackendServer | None = None - self._backend_client: BackendClient | None = None + self._backend_server = backend_server + self._backend_client = BackendClient(backend_server) if backend_server else None def compose(self) -> ComposeResult: from .widgets.left_panel import LeftPanel from .widgets.right_panel import RightPanel from .widgets.status_bar import StatusBar - yield LeftPanel() - yield RightPanel(self._config, use_embedded_db=True) + yield RightPanel() yield StatusBar() def on_mount(self) -> None: + if not self._backend_server: + from .widgets.message_history import MessageHistory + history = self.query_one(MessageHistory) + error = Message(role="assistant", content="后端未初始化") + history.add_message(error) + return + + status = self._backend_client.get_status() + api_configured = status.get("config", {}).get("api_key", "") != "" + from .widgets.message_history import MessageHistory history = self.query_one(MessageHistory) - - try: - self._backend_server = BackendServer( - db_path="graph_memory.db", - use_embedded_db=True - ) - self._backend_server.start( - api_key=self._config.api_key, - base_url=self._config.base_url - ) - - self._backend_client = BackendClient(self._backend_server) - - welcome = Message( - role="assistant", - content="系统初始化成功！\n\n" - f"数据库: 内嵌SQLite (graph_memory.db)\n" - f"API Key: {'已配置' if self._config.api_key else '未配置'}\n\n" - "现在可以开始对话了！", - ) - history.add_message(welcome) - - except Exception as e: - error = Message( - role="assistant", - content=f"初始化失败: {str(e)}\n\n" - "请检查：\n" - "1. API Key 是否配置\n" - "2. 网络连接是否正常\n\n" - "按F2打开侧边栏配置API Key", - ) - history.add_message(error) + welcome = Message( + role="assistant", + content=f"系统就绪\nAPI Key: {'已配置' if api_configured else '未配置'}\n\n输入消息开始对话" + ) + history.add_message(welcome) def on_unmount(self) -> None: - if self._backend_server: - self._backend_server.shutdown() + if self._backend_client: + self._backend_client.shutdown() def action_show_help(self) -> None: - help_text = """ -快捷键： -F1 - 帮助 -F2 - 切换侧边栏 -F3 - 工具详情 -F4 - 查询框 -F5 - 清屏 -F6 - 退出 - -输入消��后按 Enter 发送 - """ - self.notify(help_text, title="帮助", timeout=10) + self.notify("F1-帮助 F2-侧边栏 F3-工具详情 F5-清屏 F6-退出", title="快捷键", timeout=10) def action_toggle_sidebar(self) -> None: from .widgets.right_panel import RightPanel sidebar = self.query_one(RightPanel) sidebar.toggle() - sidebar.update_title() def action_toggle_tool_details(self) -> None: from .widgets.message_history import MessageHistory history = self.query_one(MessageHistory) history.toggle_latest_tool_details() - def action_focus_query(self) -> None: - from .widgets.right_panel import RightPanel - sidebar = self.query_one(RightPanel) - - if not sidebar.has_cypher_query_box(): - self.notify("查询框仅在 Neo4j 模式下可用", title="提示", timeout=3) - return - - if sidebar.is_collapsed(): - sidebar.toggle() - sidebar.update_title() - - query_box = sidebar.get_cypher_query_box() - if query_box: - query_box.focus() - def action_clear_history(self) -> None: from .widgets.message_history import MessageHistory history = self.query_one(MessageHistory) history.clear_messages() def on_input_box_send_message(self, event) -> None: - from .widgets.input_box import InputBox - from .widgets.message_history import MessageHistory - from .widgets.right_panel import RightPanel + if not self._backend_client: + return - try: - history = self.query_one(MessageHistory) - user_message = Message(role="user", content=event.content) - history.add_message(user_message) - - if not self._config.api_key: - response_msg = Message( - role="assistant", - content="请先配置API Key。\n\n按F2打开侧边栏，输入API Key后按Enter保存。", - ) - history.add_message(response_msg) - return - - processing_msg = Message(role="assistant", content="正在处理...") - history.add_message(processing_msg) - - asyncio.create_task(self._process_message_async(event.content)) - - except Exception as e: - error_msg = Message(role="assistant", content=f"错误: {str(e)}") - history.add_message(error_msg) - - async def _process_message_async(self, user_input: str) -> None: + user_input = event.content from .widgets.message_history import MessageHistory - from .widgets.right_panel import RightPanel - history = self.query_one(MessageHistory) - log = self.query_one(RightPanel).get_operation_log() + + history.add_message(Message(role="user", content=user_input)) + history.add_message(Message(role="assistant", content="处理中...")) + + asyncio.create_task(self._process(user_input)) + async def _process(self, user_input: str) -> None: try: - if not self._backend_client: - raise Exception("后端未初始化") - result = await asyncio.get_event_loop().run_in_executor( None, - lambda: self._backend_client.process_message(user_input) + lambda: self._backend_client.send_message(user_input) ) - - content = result.get("content", "(无回复)") - tool_calls_data = result.get("tool_calls", []) - rejected_tools = result.get("rejected_tools", []) - - tool_calls = [] - tool_results = [] - for tc in tool_calls_data: - tc_obj = ToolCall( - id=tc.get("id", ""), - name=tc.get("name", ""), - arguments=tc.get("arguments", {}) - ) - tool_calls.append(tc_obj) - - tr = ToolResult( - tool_call_id=tc_obj.id, - name=tc_obj.name, - arguments=tc_obj.arguments, - result=tc.get("result", ""), - success=not tc.get("result", "").startswith("工具执行���误") - ) - tool_results.append(tr) - - log_entry = LogEntry( - tool_name=tc_obj.name, - arguments=tc_obj.arguments, - result=tc.get("result", ""), - ) - log.add_log(log_entry) - - assistant_message = Message( - role="assistant", - content=content, - tool_calls=tool_calls if tool_calls else None, - tool_results=tool_results if tool_results else None - ) - - history.add_message(assistant_message) - self.refresh() - - except Exception as e: - error_msg = str(e) + from .widgets.message_history import MessageHistory + history = self.query_one(MessageHistory) + history.clear_messages() - if "Connection error" in error_msg or "connection" in error_msg.lower(): - help_text = """ -网络连接错误！可能的原因： -1. API Key 未配置或无效 -2. 网络无法访问 API 服务器 -3. API 服务器暂时不可用 - -解决方法： -按 F2 展开侧边栏，检查并配置 API Key -检查网络连接 -""" - elif "API Key" in error_msg: - help_text = """ -API Key 未配置！ - -请按以下步骤配置： -1. 按 F2 展开右侧边栏 -2. 点击"配置"展开配置区 -3. 在 API Key 输入框输入你的密钥 -4. 按 Enter 键保存配置 - -获取 API Key: https://platform.deepseek.com/ -""" + history.add_message(Message(role="user", content=user_input)) + + if result.get("success"): + content = result.get("content", "(无回复)") + history.add_message(Message(role="assistant", content=content)) else: - help_text = f"\n详细错误: {error_msg}" - - error_message = Message(role="assistant", content=f"错误: {error_msg}\n{help_text}") - history.add_message(error_message) + error = result.get("error", "未知错误") + history.add_message(Message(role="assistant", content=f"错误: {error}")) + + except Exception as e: + from .widgets.message_history import MessageHistory + history = self.query_one(MessageHistory) + history.add_message(Message(role="assistant", content=f"错误: {str(e)}")) - def on_config_section_config_changed(self, event) -> None: - from .widgets.right_panel import RightPanel + def on_config_changed(self, event) -> None: + if not self._backend_client: + return - self._config = event.config - self._config_manager.save(self._config) + api_key = event.api_key + base_url = event.base_url - try: - right_panel = self.query_one(RightPanel) - right_panel._config = self._config - except Exception: - pass - - if self._backend_client: - self._backend_client.update_config( - api_key=self._config.api_key, - base_url=self._config.base_url - ) - self.notify("配置已保存并应用", title="配置") - else: - self.notify("配置已保存，但后端未初始化", title="警告") \ No newline at end of file + result = self._backend_client.update_config(api_key=api_key, base_url=base_url) + self.notify("配置已保存" if result.get("success") else "配置失败", title="配置") \ No newline at end of file