feat(llm): vector-db

.vibedev/spec/hugegraph-llm/fixed_flow/requirements.md

@@ -0,0 +1,24 @@
+## 需求列表
+
+### 核心框架设计
+
+**核心**：Scheduler类中的schedule_flow设计与实现
+
+**验收标准**：
+1.1. 核心框架尽可能复用资源，避免资源的重复分配和释放
+1.2. 应该保证正常的请求处理指标要求
+1.3. 应该能够配置框架整体使用的资源上限
+
+### 固定工作流移植
+
+**核心**：移植Web Demo中的所有用例
+2.1. 保证使用核心框架移植后的工作流的程序行为和移植之前保持一致即可
+
+**已完成的工作流类型**：
+- build_vector_index: 向量索引构建工作流
+- graph_extract: 图抽取工作流
+- import_graph_data: 图数据导入工作流
+- update_vid_embeddings: 向量更新工作流
+- get_graph_index_info: 图索引信息获取工作流
+- build_schema: 模式构建工作流
+- prompt_generate: 提示词生成工作流

.vibedev/spec/hugegraph-llm/fixed_flow/tasks.md

@@ -0,0 +1,36 @@
+# HugeGraph-ai 固定工作流框架设计和用例移植
+
+本文档将 HugeGraph 固定工作流框架设计和用例移植转换为一系列可执行的编码任务。
+
+## 1. schedule_flow设计与实现
+
+- [x] **1.1 构建Scheduler框架1.0**
+    -   需要能够复用已经创建过的Pipeline(Pipeline Pooling)
+    -   使用CGraph(Graph-based engine)作为底层执行引擎
+    -   不同Node之间松耦合
+
+- [ ] **1.2 优化Scheduler框架资源配置**
+    -   支持用户配置底层线程池参数
+    -   现有的workflow可能会根据输入有细小的变化，导致相同的用例得到不同的workflow，怎么解决这个问题呢？
+    -   Node/Operator解耦，Node负责生命周期和上下文，Operator只关注业务逻辑
+    -   Flow只负责组装Node，所有业务逻辑下沉到Node/Operator
+    -   Scheduler支持多类型Flow注册，注册方式更灵活
+
+- [ ] **1.3 优化Scheduler框架资源使用**
+    -   根据负载控制每个PipelineManager管理的Pipeline数量，实现动态扩缩容
+    -   Node层支持参数区自动绑定和并发安全
+    -   Operator只需实现run(data_json)方法，Node负责调度和结果写回
+
+## 2. 固定工作流用例移植
+
+- [x] **2.1 build_vector_index workflow移植**
+- [x] **2.2 graph_extract workflow移植**
+- [x] **2.3 import_graph_data workflow移植**
+    -   基于Node/Operator机制实现import_graph_data工作流
+- [x] **2.4 update_vid_embeddings workflow移植**
+    -   基于Node/Operator机制实现update_vid_embeddings工作流
+- [x] **2.5 get_graph_index_info workflow移植**
+- [x] **2.6 build_schema workflow移植**
+    -   基于Node/Operator机制实现build_schema工作流
+- [x] **2.7 prompt_generate workflow移植**
+    -   基于Node/Operator机制实现prompt_generate工作流

hugegraph-llm/README.md

@@ -113,6 +113,9 @@ python -m hugegraph_llm.demo.rag_demo.app --host 127.0.0.1 --port 18001
 > The following commands assume you're in the activated virtual environment from step 4 above
 
 ```bash
+# To use vector database backends (e.g., Milvus, Qdrant), sync the optional dependencies:
+uv sync --extra vectordb
+
 # Download NLTK stopwords for better text processing
 python ./src/hugegraph_llm/operators/common_op/nltk_helper.py
 

hugegraph-llm/config.md

@@ -16,6 +16,7 @@
   - [LiteLLM 配置](#litellm-配置)
   - [重排序配置](#重排序配置)
   - [HugeGraph 数据库配置](#hugegraph-数据库配置)
+  - [向量数据库配置](#向量数据库配置)
   - [管理员配置](#管理员配置)
 - [配置使用示例](#配置使用示例)
 - [配置文件位置](#配置文件位置)
@@ -127,6 +128,18 @@
 | `TOPK_PER_KEYWORD`     | Optional[Integer] | 1              | 每个关键词返回的 TopK 数量   |
 | `TOPK_RETURN_RESULTS`  | Optional[Integer] | 20             | 返回结果数量             |
 
+### 向量数据库配置
+
+| 配置项              | 类型               | 默认值  | 说明                     |
+|------------------|------------------|-------|------------------------|
+| `QDRANT_HOST`    | Optional[String] | None  | Qdrant 服务器主机地址         |
+| `QDRANT_PORT`    | Integer          | 6333  | Qdrant 服务器端口           |
+| `QDRANT_API_KEY` | Optional[String] | None  | Qdrant API 密钥（如果设置了的话） |
+| `MILVUS_HOST`    | Optional[String] | None  | Milvus 服务器主机地址         |
+| `MILVUS_PORT`    | Integer          | 19530 | Milvus 服务器端口           |
+| `MILVUS_USER`    | String           | ""    | Milvus 用户名              |
+| `MILVUS_PASSWORD`| String           | ""    | Milvus 密码               |
+
 ### 管理员配置
 
 | 配置项            | 类型               | 默认值     | 说明                 |

hugegraph-llm/pyproject.toml

@@ -22,11 +22,12 @@ description = "A tool for the implementation and research related to large langu
 authors = [
     { name = "Apache HugeGraph Contributors", email = "dev@hugegraph.apache.org" },
 ]
+maintainers = [
+    { name = "Apache HugeGraph Contributors", email = "dev@hugegraph.apache.org" },
+]
 readme = "README.md"
 license = "Apache-2.0"
 requires-python = ">=3.10,<3.12"
-
-
 dependencies = [
     # Common dependencies
     "decorator",
@@ -39,6 +40,7 @@ dependencies = [
     "numpy",
     "pandas",
     "pydantic",
+    "tqdm",
 
     # LLM specific dependencies
     "openai",
@@ -58,7 +60,15 @@ dependencies = [
     "apscheduler",
     "litellm",
     "hugegraph-python-client",
+    "pycgraph",
+]
+
+[project.optional-dependencies]
+vectordb = [
+    "pymilvus==2.5.9",
+    "qdrant-client==1.14.2",
 ]
+
 [project.urls]
 homepage = "https://hugegraph.apache.org/"
 repository = "https://github.com/apache/incubator-hugegraph-ai"
@@ -85,3 +95,23 @@ allow-direct-references = true
 
 [tool.uv.sources]
 hugegraph-python-client = { workspace = true }
+pycgraph = { git = "https://github.com/ChunelFeng/CGraph.git", subdirectory = "python", rev = "main", marker = "sys_platform == 'linux'"  }
+
+[tool.mypy]
+disable_error_code = ["import-untyped"]
+check_untyped_defs = true
+disallow_untyped_defs = false
+
+[tool.ruff]
+line-length = 120
+indent-width = 4
+extend-exclude = []
+
+# TODO: move this config in the root pyproject.toml & add more rules for it
+[tool.ruff.lint]
+extend-select = ["I"]
+
+[tool.ruff.format]
+quote-style = "preserve"
+indent-style = "space"
+line-ending = "auto"

hugegraph-llm/src/hugegraph_llm/api/admin_api.py

@@ -31,8 +31,11 @@ def admin_http_api(router: APIRouter, log_stream):
     @router.post("/logs", status_code=status.HTTP_200_OK)
     async def log_stream_api(req: LogStreamRequest):
         if admin_settings.admin_token != req.admin_token:
-            raise generate_response(RAGResponse(status_code=status.HTTP_403_FORBIDDEN, #pylint: disable=E0702
-                                                message="Invalid admin_token"))
+            raise generate_response(
+                RAGResponse(
+                    status_code=status.HTTP_403_FORBIDDEN, message="Invalid admin_token"
+                )  # pylint: disable=E0702
+            )
         log_path = os.path.join("logs", req.log_file)
 
         # Create a StreamingResponse that reads from the log stream generator

hugegraph-llm/src/hugegraph_llm/api/exceptions/rag_exceptions.py

@@ -21,7 +21,9 @@
 
 class ExternalException(HTTPException):
     def __init__(self):
-        super().__init__(status_code=400, detail="Connect failed with error code -1, please check the input.")
+        super().__init__(
+            status_code=400, detail="Connect failed with error code -1, please check the input."
+        )
 
 
 class ConnectionFailedException(HTTPException):

hugegraph-llm/src/hugegraph_llm/api/models/rag_requests.py

@@ -24,10 +24,10 @@
 
 
 class GraphConfigRequest(BaseModel):
-    url: str = Query('127.0.0.1:8080', description="hugegraph client url.")
-    graph: str = Query('hugegraph', description="hugegraph client name.")
-    user: str = Query('', description="hugegraph client user.")
-    pwd: str = Query('', description="hugegraph client pwd.")
+    url: str = Query("127.0.0.1:8080", description="hugegraph client url.")
+    graph: str = Query("hugegraph", description="hugegraph client name.")
+    user: str = Query("", description="hugegraph client user.")
+    pwd: str = Query("", description="hugegraph client pwd.")
     gs: str = None
 
 
@@ -36,22 +36,42 @@ class RAGRequest(BaseModel):
     raw_answer: bool = Query(False, description="Use LLM to generate answer directly")
     vector_only: bool = Query(False, description="Use LLM to generate answer with vector")
     graph_only: bool = Query(True, description="Use LLM to generate answer with graph RAG only")
-    graph_vector_answer: bool = Query(False, description="Use LLM to generate answer with vector & GraphRAG")
+    graph_vector_answer: bool = Query(
+        False, description="Use LLM to generate answer with vector & GraphRAG"
+    )
     graph_ratio: float = Query(0.5, description="The ratio of GraphRAG ans & vector ans")
-    rerank_method: Literal["bleu", "reranker"] = Query("bleu", description="Method to rerank the results.")
-    near_neighbor_first: bool = Query(False, description="Prioritize near neighbors in the search results.")
-    custom_priority_info: str = Query("", description="Custom information to prioritize certain results.")
+    rerank_method: Literal["bleu", "reranker"] = Query(
+        "bleu", description="Method to rerank the results."
+    )
+    near_neighbor_first: bool = Query(
+        False, description="Prioritize near neighbors in the search results."
+    )
+    custom_priority_info: str = Query(
+        "", description="Custom information to prioritize certain results."
+    )
     # Graph Configs
-    max_graph_items: int = Query(30, description="Maximum number of items for GQL queries in graph.")
+    max_graph_items: int = Query(
+        30, description="Maximum number of items for GQL queries in graph."
+    )
     topk_return_results: int = Query(20, description="Number of sorted results to return finally.")
-    vector_dis_threshold: float = Query(0.9, description="Threshold for vector similarity\
-                                         (results greater than this will be ignored).")
-    topk_per_keyword: int = Query(1, description="TopK results returned for each keyword \
-                                   extracted from the query, by default only the most similar one is returned.")
-    client_config: Optional[GraphConfigRequest] = Query(None, description="hugegraph server config.")
+    vector_dis_threshold: float = Query(
+        0.9,
+        description="Threshold for vector similarity\
+                                         (results greater than this will be ignored).",
+    )
+    topk_per_keyword: int = Query(
+        1,
+        description="TopK results returned for each keyword \
+                                   extracted from the query, by default only the most similar one is returned.",
+    )
+    client_config: Optional[GraphConfigRequest] = Query(
+        None, description="hugegraph server config."
+    )
 
     # Keep prompt params in the end
-    answer_prompt: Optional[str] = Query(prompt.answer_prompt, description="Prompt to guide the answer generation.")
+    answer_prompt: Optional[str] = Query(
+        prompt.answer_prompt, description="Prompt to guide the answer generation."
+    )
     keywords_extract_prompt: Optional[str] = Query(
         prompt.keywords_extract_prompt,
         description="Prompt for extracting keywords from query.",
@@ -67,22 +87,39 @@ class RAGRequest(BaseModel):
 class GraphRAGRequest(BaseModel):
     query: str = Query(..., description="Query you want to ask")
     # Graph Configs
-    max_graph_items: int = Query(30, description="Maximum number of items for GQL queries in graph.")
+    max_graph_items: int = Query(
+        30, description="Maximum number of items for GQL queries in graph."
+    )
     topk_return_results: int = Query(20, description="Number of sorted results to return finally.")
-    vector_dis_threshold: float = Query(0.9, description="Threshold for vector similarity \
-                                        (results greater than this will be ignored).")
-    topk_per_keyword: int = Query(1, description="TopK results returned for each keyword extracted\
-                                    from the query, by default only the most similar one is returned.")
+    vector_dis_threshold: float = Query(
+        0.9,
+        description="Threshold for vector similarity \
+                                        (results greater than this will be ignored).",
+    )
+    topk_per_keyword: int = Query(
+        1,
+        description="TopK results returned for each keyword extracted\
+                                    from the query, by default only the most similar one is returned.",
+    )
 
-    client_config: Optional[GraphConfigRequest] = Query(None, description="hugegraph server config.")
+    client_config: Optional[GraphConfigRequest] = Query(
+        None, description="hugegraph server config."
+    )
     get_vertex_only: bool = Query(False, description="return only keywords & vertex (early stop).")
 
     gremlin_tmpl_num: int = Query(
-        1, description="Number of Gremlin templates to use. If num <=0 means template is not provided"
+        1,
+        description="Number of Gremlin templates to use. If num <=0 means template is not provided",
+    )
+    rerank_method: Literal["bleu", "reranker"] = Query(
+        "bleu", description="Method to rerank the results."
+    )
+    near_neighbor_first: bool = Query(
+        False, description="Prioritize near neighbors in the search results."
+    )
+    custom_priority_info: str = Query(
+        "", description="Custom information to prioritize certain results."
     )
-    rerank_method: Literal["bleu", "reranker"] = Query("bleu", description="Method to rerank the results.")
-    near_neighbor_first: bool = Query(False, description="Prioritize near neighbors in the search results.")
-    custom_priority_info: str = Query("", description="Custom information to prioritize certain results.")
     gremlin_prompt: Optional[str] = Query(
         prompt.gremlin_generate_prompt,
         description="Prompt for the Text2Gremlin query.",
@@ -115,39 +152,44 @@ class LogStreamRequest(BaseModel):
     admin_token: Optional[str] = None
     log_file: Optional[str] = "llm-server.log"
 
+
 class GremlinOutputType(str, Enum):
     MATCH_RESULT = "match_result"
     TEMPLATE_GREMLIN = "template_gremlin"
     RAW_GREMLIN = "raw_gremlin"
     TEMPLATE_EXECUTION_RESULT = "template_execution_result"
     RAW_EXECUTION_RESULT = "raw_execution_result"
 
+
 class GremlinGenerateRequest(BaseModel):
     query: str
     example_num: Optional[int] = Query(
-        0,
-        description="Number of Gremlin templates to use.(0 means no templates)"
+        0, description="Number of Gremlin templates to use.(0 means no templates)"
     )
     gremlin_prompt: Optional[str] = Query(
         prompt.gremlin_generate_prompt,
         description="Prompt for the Text2Gremlin query.",
     )
-    client_config: Optional[GraphConfigRequest] = Query(None, description="hugegraph server config.")
+    client_config: Optional[GraphConfigRequest] = Query(
+        None, description="hugegraph server config."
+    )
     output_types: Optional[List[GremlinOutputType]] = Query(
         default=[GremlinOutputType.TEMPLATE_GREMLIN],
         description="""
         a list can contain "match_result","template_gremlin",
         "raw_gremlin","template_execution_result","raw_execution_result"
         You can specify which type of result do you need. Empty means all types.
-        """
+        """,
     )
 
-    @field_validator('gremlin_prompt')
+    @field_validator("gremlin_prompt")
     @classmethod
     def validate_prompt_placeholders(cls, v):
         if v is not None:
-            required_placeholders = ['{query}', '{schema}', '{example}', '{vertices}']
+            required_placeholders = ["{query}", "{schema}", "{example}", "{vertices}"]
             missing = [p for p in required_placeholders if p not in v]
             if missing:
-                raise ValueError(f"Prompt template is missing required placeholders: {', '.join(missing)}")
+                raise ValueError(
+                    f"Prompt template is missing required placeholders: {', '.join(missing)}"
+                )
         return v