Feat/online embeddings support and refactor qdrant code using langchian-qdrant

.env.example

@@ -90,4 +90,5 @@ SPARK_APPID=changethis
 SPARK_APISecret=changethis
 SPARK_APIKey=changethis
 
-ZHIPUAI_API_KEY=changethis
+ZHIPUAI_API_KEY=changethis
+SILICONFLOW_API_KEY=changethis

backend/app/core/celery_app.py

@@ -1,8 +1,10 @@
 import os
-
 from celery import Celery
-
 from app.core.config import settings
+import logging
+
+# 配置基本日志
+logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
 
 os.environ["HUGGINGFACE_HUB_CACHE"] = os.path.join(os.getcwd(), "fastembed_cache")
 os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"
@@ -22,3 +24,18 @@
 celery_app.conf.update(
     result_expires=3600,
 )
+
+celery_app.conf.task_routes = {"app.worker.celery_worker.*": "main-queue"}
+celery_app.conf.update(task_track_started=True)
+
+# 配置 Celery 日志
+celery_app.conf.update(
+    worker_hijack_root_logger=False,
+    worker_log_format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    worker_task_log_format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+
+@celery_app.task(acks_late=True)
+def test_celery(word: str) -> str:
+    logging.info(f"Test task received: {word}")
+    return f"test task return {word}"

backend/app/core/config.py

@@ -142,9 +142,11 @@ def _enforce_non_default_secrets(self) -> Self:
         return self
 
     # Qdrant
-    QDRANT_SERVICE_API_KEY: str | None = None
-    QDRANT_URL: str | None = None
-    QDRANT_COLLECTION: str | None = None
+    QDRANT_SERVICE_API_KEY: str | None = "XMj3HXm5GlBKQLwZuStOlkwZiOWTdd_IwZNDJINFh-w"
+    # QDRANT_URL: str = "http://localhost:6333"
+    QDRANT_URL: str = "http://127.0.0.1:6333"
+
+    QDRANT_COLLECTION: str | None = "kb_uploads"
 
     # LangSmith
     # USE_LANGSMITH: bool = True
@@ -154,9 +156,18 @@ def _enforce_non_default_secrets(self) -> Self:
     # LANGCHAIN_PROJECT: str | None = None
 
     # Embeddings
-    DENSE_EMBEDDING_MODEL: str | None = None
-    SPARSE_EMBEDDING_MODEL: str | None = None
-    FASTEMBED_CACHE_PATH: str | None = None
+    # EMBEDDING_MODEL: str = "local"  # 或者你想使用的其他模型
+    EMBEDDING_MODEL: str = "zhipuai"  # 或者你想使用的其他模型
+
+    DENSE_EMBEDDING_MODEL: str = (
+        "sentence-transformers/all-MiniLM-L6-v2"  # 默认的密集嵌入模型
+    )
+    SPARSE_EMBEDDING_MODEL: str = (
+        "sentence-transformers/all-MiniLM-L6-v2"  # 默认的稀疏嵌入模型
+    )
+    ZHIPUAI_API_KEY: str | None = None
+    SILICONFLOW_API_KEY: str | None = None
+    OLLAMA_BASE_URL: str | None = None
 
     # Celery
     CELERY_BROKER_URL: str | None = None
@@ -166,5 +177,7 @@ def _enforce_non_default_secrets(self) -> Self:
     RECURSION_LIMIT: int = 25
     TAVILY_API_KEY: str | None = None
 
+    OPENAI_API_KEY: str
+
 
 settings = Settings()  # type: ignore

backend/app/core/graph/members.py

@@ -1,8 +1,9 @@
 from collections.abc import Mapping, Sequence
 from typing import Annotated, Any
-
+from app.core.rag.qdrant import QdrantStore
+from langchain_core.tools import BaseTool
+from pydantic import BaseModel, Field
 from langchain.chat_models import init_chat_model
-from langchain.tools.retriever import create_retriever_tool
 from langchain_core.messages import AIMessage, AnyMessage
 from langchain_core.output_parsers.openai_tools import JsonOutputKeyToolsParser
 from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
@@ -18,7 +19,7 @@
 from pydantic import BaseModel, Field
 from typing_extensions import NotRequired, TypedDict
 
-from app.core.graph.rag.qdrant import QdrantStore
+from app.core.rag.qdrant import QdrantStore
 from app.core.tools import managed_tools
 from app.core.tools.api_tool import dynamic_api_tool
 from app.core.tools.retriever_tool import create_retriever_tool
@@ -41,6 +42,18 @@ def tool(self) -> BaseTool:
             raise ValueError("Skill is not managed and no definition provided.")
 
 
+# class GraphUpload(BaseModel):
+#     name: str = Field(description="Name of the upload")
+#     description: str = Field(description="Description of the upload")
+#     owner_id: int = Field(description="Id of the user that owns this upload")
+#     upload_id: int = Field(description="Id of the upload")
+
+#     @property
+#     def tool(self) -> BaseTool:
+#         retriever = QdrantStore().retriever(self.owner_id, self.upload_id)
+#         return create_retriever_tool(retriever)
+
+
 class GraphUpload(BaseModel):
     name: str = Field(description="Name of the upload")
     description: str = Field(description="Description of the upload")
@@ -49,7 +62,8 @@ class GraphUpload(BaseModel):
 
     @property
     def tool(self) -> BaseTool:
-        retriever = QdrantStore().retriever(self.owner_id, self.upload_id)
+        qdrant_store = QdrantStore()
+        retriever = qdrant_store.retriever(self.owner_id, self.upload_id)
         return create_retriever_tool(retriever)
 
 

backend/app/core/graph/messages.py

@@ -83,7 +83,7 @@ def event_to_response(event: StreamEvent) -> ChatResponse | None:
             for doc in docs:
                 documents.append(
                     {
-                        "score": doc.metadata["score"],
+                        # "score": doc.metadata["score"],
                         "content": doc.page_content,
                     }
                 )