feat: add memreader to tackle with internet

Author: CaralHsi

src/memos/configs/internet_retriever.py

@@ -7,6 +7,7 @@
 from memos.chunkers.factory import ChunkerConfigFactory
 from memos.configs.base import BaseConfig
 from memos.exceptions import ConfigurationError
+from memos.mem_reader.factory import MemReaderConfigFactory
 
 
 class BaseInternetRetrieverConfig(BaseConfig):
@@ -53,6 +54,11 @@ class XinyuSearchConfig(BaseInternetRetrieverConfig):
         default_factory=ChunkerConfigFactory,
         description="Chunker configuration",
     )
+    reader: MemReaderConfigFactory = Field(
+        ...,
+        default_factory=MemReaderConfigFactory,
+        description="Reader configuration",
+    )
 
 
 class InternetRetrieverConfigFactory(BaseConfig):

src/memos/mem_reader/simple_struct.py

@@ -180,8 +180,12 @@ def get_scene_data_info(self, scene_data: list, type: str) -> list[str]:
         elif type == "doc":
             for item in scene_data:
                 try:
-                    parsed_text = parser.parse(item)
-                    results.append({"file": item, "text": parsed_text})
+                    if not isinstance(item, str):
+                        parsed_text = parser.parse(item)
+                        results.append({"file": "pure_text", "text": parsed_text})
+                    else:
+                        parsed_text = item
+                        results.append({"file": item, "text": parsed_text})
                 except Exception as e:
                     print(f"Error parsing file {item}: {e!s}")
 

src/memos/memories/textual/tree_text_memory/retrieve/internet_retriever_factory.py

@@ -5,6 +5,7 @@
 from memos.chunkers.factory import ChunkerFactory
 from memos.configs.internet_retriever import InternetRetrieverConfigFactory
 from memos.embedders.base import BaseEmbedder
+from memos.mem_reader.factory import MemReaderFactory
 from memos.memories.textual.tree_text_memory.retrieve.internet_retriever import (
     InternetGoogleRetriever,
 )
@@ -68,6 +69,7 @@ def from_config(
                 search_engine_id=config.search_engine_id,
                 embedder=embedder,
                 chunker=ChunkerFactory.from_config(config.chunker),
+                reader=MemReaderFactory.from_config(config.reader),
                 max_results=config.max_results,
             )
         else:

src/memos/memories/textual/tree_text_memory/retrieve/searcher.py

@@ -141,7 +141,7 @@ def retrieve_from_internet():
             if memory_type not in ["All"]:
                 return []
             internet_items = self.internet_retriever.retrieve_from_internet(
-                query=query, top_k=top_k, parsed_goal=parsed_goal
+                query=query, top_k=top_k, parsed_goal=parsed_goal, info=info
             )
 
             # Convert to the format expected by reranker

src/memos/memories/textual/tree_text_memory/retrieve/xinyusearch.py

@@ -11,7 +11,8 @@
 from memos.chunkers.base import BaseChunker
 from memos.embedders.factory import OllamaEmbedder
 from memos.log import get_logger
-from memos.memories.textual.item import TextualMemoryItem, TreeNodeTextualMemoryMetadata
+from memos.mem_reader.base import BaseMemReader
+from memos.memories.textual.item import TextualMemoryItem
 
 
 logger = get_logger(__name__)
@@ -115,6 +116,7 @@ def __init__(
         search_engine_id: str,
         embedder: OllamaEmbedder,
         chunker: BaseChunker,
+        reader: BaseMemReader,
         max_results: int = 20,
     ):
         """
@@ -128,9 +130,10 @@ def __init__(
         self.xinyu_api = XinyuSearchAPI(access_key, search_engine_id, max_results=max_results)
         self.embedder = embedder
         self.chunker = chunker
+        self.reader = reader
 
     def retrieve_from_internet(
-        self, query: str, top_k: int = 10, parsed_goal=None
+        self, query: str, top_k: int = 10, parsed_goal=None, info=None
     ) -> list[TextualMemoryItem]:
         """
         Retrieve information from Xinyu search and convert to TextualMemoryItem format
@@ -139,7 +142,7 @@ def retrieve_from_internet(
             query: Search query
             top_k: Number of results to return
             parsed_goal: Parsed task goal (optional)
-
+            info (dict): Leave a record of memory consumption.
         Returns:
             List of TextualMemoryItem
         """
@@ -151,7 +154,7 @@ def retrieve_from_internet(
 
         with ThreadPoolExecutor(max_workers=8) as executor:
             futures = [
-                executor.submit(self._process_result, result, query, parsed_goal)
+                executor.submit(self._process_result, result, query, parsed_goal, info)
                 for result in search_results
             ]
             for future in as_completed(futures):
@@ -301,7 +304,7 @@ def _extract_tags(self, title: str, content: str, summary: str, parsed_goal=None
         return list(set(tags))[:15]  # Limit to 15 tags
 
     def _process_result(
-        self, result: dict, query: str, parsed_goal: str
+        self, result: dict, query: str, parsed_goal: str, info: dict
     ) -> list[TextualMemoryItem]:
         title = result.get("title", "")
         content = result.get("content", "")
@@ -318,55 +321,19 @@ def _process_result(
                 publish_time = datetime.now().strftime("%Y-%m-%d")
         else:
             publish_time = datetime.now().strftime("%Y-%m-%d")
-        source = result.get("source", "")
-        site = result.get("site", "")
-        if site:
-            site = site.split("|")[0]
 
-        qualified_chunks = self._chunk(content)
+        read_items = self.reader.get_memory([content], type="doc", info=info)
 
         memory_items = []
-        for chunk_text, chunk_emb, score in qualified_chunks:
-            memory_content = (
+        for read_item_i in read_items[0]:
+            read_item_i.memory = (
                 f"Title: {title}\nNewsTime: {publish_time}\nSummary: {summary}\n"
-                f"Content: {chunk_text}\nSource: {url}"
-            )
-            metadata = TreeNodeTextualMemoryMetadata(
-                user_id=None,
-                session_id=None,
-                status="activated",
-                type="fact",
-                source="web",
-                confidence=score,
-                entities=self._extract_entities(title, content, summary),
-                tags=self._extract_tags(title, content, summary, parsed_goal),
-                visibility="public",
-                memory_type="OuterMemory",
-                key=f"[{source}]" + title,
-                sources=[url] if url else [],
-                embedding=chunk_emb,
-                created_at=datetime.now().isoformat(),
-                usage=[],
-                background=f"Xinyu search result from {site or source}",
-            )
-            memory_items.append(
-                TextualMemoryItem(id=str(uuid.uuid4()), memory=memory_content, metadata=metadata)
+                f"Content: {read_item_i.memory}"
             )
+            read_item_i.metadata.source = "web"
+            read_item_i.metadata.memory_type = "OuterMemory"
+            read_item_i.metadata.sources = [url] if url else []
+            read_item_i.metadata.visibility = "public"
 
+            memory_items.append(read_item_i)
         return memory_items
-
-    def _chunk(self, content: str) -> list[tuple[str, list[float], float]]:
-        """
-        Use SentenceChunker to split content into chunks and embed each.
-
-        Returns:
-            List of (chunk_text, chunk_embedding, dummy_score)
-        """
-        chunks = self.chunker.chunk(content)
-        if not chunks:
-            return []
-
-        chunk_texts = [c.text for c in chunks]
-        chunk_embeddings = self.embedder.embed(chunk_texts)
-
-        return [(text, emb, 1.0) for text, emb in zip(chunk_texts, chunk_embeddings, strict=False)]