feat: knowledge base for long-term memory (#1099)

backend/app/agent/toolkit/knowledge_base_toolkit.py

@@ -0,0 +1,162 @@
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+
+"""
+Toolkit for long-term memory using markdown files (issue #1099).
+
+Agents can both read and write to the project's memory file (.eigent/memory.md).
+This provides a simple, file-based approach for persistent knowledge storage.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Final
+
+from camel.toolkits.base import BaseToolkit
+from camel.toolkits.function_tool import FunctionTool
+
+from app.agent.toolkit.abstract_toolkit import AbstractToolkit
+from app.component.environment import env
+from app.utils.memory_file import append_memory, read_memory
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_WORKING_DIR: Final[str] = "~/.eigent"
+_NO_MEMORY_MESSAGE: Final[str] = (
+    "No project memory exists yet. Use remember_this to save information."
+)
+_SUCCESS_MESSAGE: Final[str] = (
+    "Saved to project memory (.eigent/memory.md). "
+    "This will be available in future conversations."
+)
+_FAILURE_MESSAGE: Final[str] = "Failed to save to project memory. Please try again."
+
+
+def _resolve_working_directory(working_directory: str | None) -> str:
+    """Resolve and validate the working directory path."""
+    if working_directory is None:
+        working_directory = env("file_save_path", os.path.expanduser(_DEFAULT_WORKING_DIR))
+    resolved = os.path.expanduser(working_directory)
+    os.makedirs(resolved, exist_ok=True)
+    return resolved
+
+
+class KnowledgeBaseToolkit(BaseToolkit, AbstractToolkit):
+    """Toolkit for reading and writing project long-term memory.
+
+    Uses a simple markdown file (.eigent/memory.md) in the project's working
+    directory. Agents can both read existing memories and write new ones.
+    """
+
+    def __init__(
+        self,
+        api_task_id: str,
+        working_directory: str | None = None,
+        agent_name: str | None = None,
+        timeout: float | None = None,
+    ) -> None:
+        if not api_task_id or not api_task_id.strip():
+            raise ValueError("api_task_id cannot be empty")
+
+        super().__init__(timeout=timeout)
+        self.api_task_id = api_task_id.strip()
+        self.working_directory = _resolve_working_directory(working_directory)
+        self.agent_name = agent_name.strip() if agent_name else "agent"
+
+        logger.debug(
+            "KnowledgeBaseToolkit initialized",
+            extra={
+                "api_task_id": self.api_task_id,
+                "working_directory": self.working_directory,
+                "agent_name": self.agent_name,
+            },
+        )
+
+    def read_project_memory(self) -> str:
+        """Read the project's long-term memory file.
+
+        Returns the content of .eigent/memory.md which contains facts,
+        preferences, and decisions that should persist across sessions.
+
+        Returns:
+            str: The memory file content, or a message if no memory exists yet.
+        """
+        try:
+            content = read_memory(self.working_directory)
+            if content is None:
+                return _NO_MEMORY_MESSAGE
+            return content
+        except Exception as e:
+            logger.error(
+                f"Error reading project memory: {e}",
+                extra={"working_directory": self.working_directory},
+            )
+            return _NO_MEMORY_MESSAGE
+
+    def remember_this(self, content: str) -> str:
+        """Save a fact or piece of information to the project's long-term memory.
+
+        Use this when the user or task establishes something that should be
+        remembered for future conversations (e.g. preferences, decisions,
+        project-specific facts). The content will be appended to .eigent/memory.md.
+
+        Args:
+            content (str): The information to remember (clear, self-contained text).
+
+        Returns:
+            str: Confirmation message indicating success or failure.
+        """
+        if not content or not content.strip():
+            return "Cannot save empty content. Please provide information to remember."
+
+        try:
+            success = append_memory(self.working_directory, content)
+            if success:
+                logger.info(
+                    "Memory saved successfully",
+                    extra={
+                        "api_task_id": self.api_task_id,
+                        "content_length": len(content),
+                    },
+                )
+                return _SUCCESS_MESSAGE
+            return _FAILURE_MESSAGE
+        except Exception as e:
+            logger.error(
+                f"Error saving to project memory: {e}",
+                extra={"working_directory": self.working_directory},
+            )
+            return _FAILURE_MESSAGE
+
+    def get_tools(self) -> list[FunctionTool]:
+        """Return the list of tools provided by this toolkit."""
+        return [
+            FunctionTool(self.read_project_memory),
+            FunctionTool(self.remember_this),
+        ]
+
+
+def get_tools(
+    api_task_id: str,
+    working_directory: str | None = None,
+    agent_name: str | None = None,
+) -> list[FunctionTool]:
+    """Return the memory tools for use by an agent."""
+    return KnowledgeBaseToolkit(
+        api_task_id=api_task_id,
+        working_directory=working_directory,
+        agent_name=agent_name,
+    ).get_tools()

backend/app/agent/tools.py

@@ -27,6 +27,7 @@
 from app.agent.toolkit.google_drive_mcp_toolkit import GoogleDriveMCPToolkit
 from app.agent.toolkit.google_gmail_mcp_toolkit import GoogleGmailMCPToolkit
 from app.agent.toolkit.image_analysis_toolkit import ImageAnalysisToolkit
+from app.agent.toolkit.knowledge_base_toolkit import KnowledgeBaseToolkit
 from app.agent.toolkit.lark_toolkit import LarkToolkit
 from app.agent.toolkit.linkedin_toolkit import LinkedInToolkit
 from app.agent.toolkit.mcp_search_toolkit import McpSearchToolkit
@@ -63,6 +64,7 @@ async def get_toolkits(tools: list[str], agent_name: str, api_task_id: str):
         "google_drive_mcp_toolkit": GoogleDriveMCPToolkit,
         "google_gmail_mcp_toolkit": GoogleGmailMCPToolkit,
         "image_analysis_toolkit": ImageAnalysisToolkit,
+        "knowledge_base_toolkit": KnowledgeBaseToolkit,
         "linkedin_toolkit": LinkedInToolkit,
         "lark_toolkit": LarkToolkit,
         "mcp_search_toolkit": McpSearchToolkit,

backend/app/router.py

@@ -13,8 +13,7 @@
 # ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
 """
 Centralized router registration for the Eigent API.
-All routers are explicitly registered here
-for better visibility and maintainability.
+All routers are explicitly registered here for better visibility and maintainability.
 """
 
 import logging
@@ -79,9 +78,7 @@ def register_routers(app: FastAPI, prefix: str = "") -> None:
         )
         route_count = len(config["router"].routes)
         logger.info(
-            f"Registered {config['tags'][0]} router:"
-            f" {route_count} routes -"
-            f" {config['description']}"
+            f"Registered {config['tags'][0]} router: {route_count} routes - {config['description']}"
         )
 
     logger.info(f"Total routers registered: {len(routers_config)}")

backend/app/service/chat_service.py

@@ -235,7 +235,8 @@ def check_conversation_history_length(
 
 
 def build_conversation_context(
-    task_lock: TaskLock, header: str = "=== CONVERSATION HISTORY ==="
+    task_lock: TaskLock,
+    header: str = "=== CONVERSATION HISTORY ===",
 ) -> str:
     """Build conversation context from task_lock history
     with files listed only once at the end.
@@ -245,14 +246,13 @@ def build_conversation_context(
         header: Header text for the context section
 
     Returns:
-        Formatted context string with task history
-        and files listed once at the end
+        Formatted context string with task history and files listed once at the end
     """
     context = ""
     working_directories = set()  # Collect all unique working directories
 
     if task_lock.conversation_history:
-        context = f"{header}\n"
+        context += f"{header}\n"
 
         for entry in task_lock.conversation_history:
             if entry["role"] == "task_result":
@@ -553,7 +553,8 @@ async def step_solve(options: Chat, request: Request, task_lock: TaskLock):
                         "without workforce"
                     )
                     conv_ctx = build_conversation_context(
-                        task_lock, header="=== Previous Conversation ==="
+                        task_lock,
+                        header="=== Previous Conversation ===",
                     )
                     simple_answer_prompt = (
                         f"{conv_ctx}"
@@ -1967,7 +1968,8 @@ async def question_confirm(
     context_prompt = ""
     if task_lock:
         context_prompt = build_conversation_context(
-            task_lock, header="=== Previous Conversation ==="
+            task_lock,
+            header="=== Previous Conversation ===",
         )
 
     full_prompt = f"""{context_prompt}User Query: {prompt}