diff --git a/README.md b/README.md index 65e8001dd2..efb37d6083 100644 --- a/README.md +++ b/README.md @@ -54,7 +54,7 @@ README in বাংলা

-Dify is an open-source LLM app development platform. Its intuitive interface combines agentic AI workflow, RAG pipeline, agent capabilities, model management, observability features and more, letting you quickly go from prototype to production. +Dify is an open-source LLM app development platform. Its intuitive interface combines agentic AI workflow, RAG pipeline, agent capabilities, model management, observability features, and more, allowing you to quickly move from prototype to production. ## Quick start @@ -188,7 +188,7 @@ All of Dify's offerings come with corresponding APIs, so you could effortlessly - **Dify for enterprise / organizations
** We provide additional enterprise-centric features. [Log your questions for us through this chatbot](https://udify.app/chat/22L1zSxg6yW1cWQg) or [send us an email](mailto:business@dify.ai?subject=[GitHub]Business%20License%20Inquiry) to discuss enterprise needs.
- > For startups and small businesses using AWS, check out [Dify Premium on AWS Marketplace](https://aws.amazon.com/marketplace/pp/prodview-t22mebxzwjhu6) and deploy it to your own AWS VPC with one-click. It's an affordable AMI offering with the option to create apps with custom logo and branding. + > For startups and small businesses using AWS, check out [Dify Premium on AWS Marketplace](https://aws.amazon.com/marketplace/pp/prodview-t22mebxzwjhu6) and deploy it to your own AWS VPC with one click. It's an affordable AMI offering with the option to create apps with custom logo and branding. ## Staying ahead @@ -233,7 +233,7 @@ Deploy Dify to AWS with [CDK](https://aws.amazon.com/cdk/) For those who'd like to contribute code, see our [Contribution Guide](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md). At the same time, please consider supporting Dify by sharing it on social media and at events and conferences. -> We are looking for contributors to help with translating Dify to languages other than Mandarin or English. If you are interested in helping, please see the [i18n README](https://github.com/langgenius/dify/blob/main/web/i18n/README.md) for more information, and leave us a comment in the `global-users` channel of our [Discord Community Server](https://discord.gg/8Tpq4AcN9c). +> We are looking for contributors to help translate Dify into languages other than Mandarin or English. If you are interested in helping, please see the [i18n README](https://github.com/langgenius/dify/blob/main/web/i18n/README.md) for more information, and leave us a comment in the `global-users` channel of our [Discord Community Server](https://discord.gg/8Tpq4AcN9c). ## Community & contact diff --git a/api/.env.example b/api/.env.example index b5820fcdc2..2cc6410cdd 100644 --- a/api/.env.example +++ b/api/.env.example @@ -297,6 +297,7 @@ LINDORM_URL=http://ld-*******************-proxy-search-pub.lindorm.aliyuncs.com: LINDORM_USERNAME=admin LINDORM_PASSWORD=admin USING_UGC_INDEX=False +LINDORM_QUERY_TIMEOUT=1 # OceanBase Vector configuration OCEANBASE_VECTOR_HOST=127.0.0.1 diff --git a/api/commands.py b/api/commands.py index e70d6e0b49..c5394c6f87 100644 --- a/api/commands.py +++ b/api/commands.py @@ -271,6 +271,7 @@ def migrate_knowledge_vector_database(): upper_collection_vector_types = { VectorType.MILVUS, VectorType.PGVECTOR, + VectorType.VASTBASE, VectorType.RELYT, VectorType.WEAVIATE, VectorType.ORACLE, diff --git a/api/configs/middleware/__init__.py b/api/configs/middleware/__init__.py index c2ad24094a..d285515998 100644 --- a/api/configs/middleware/__init__.py +++ b/api/configs/middleware/__init__.py @@ -39,6 +39,7 @@ from .vdb.tencent_vector_config import TencentVectorDBConfig from .vdb.tidb_on_qdrant_config import TidbOnQdrantConfig from .vdb.tidb_vector_config import TiDBVectorConfig from .vdb.upstash_config import UpstashConfig +from .vdb.vastbase_vector_config import VastbaseVectorConfig from .vdb.vikingdb_config import VikingDBConfig from .vdb.weaviate_config import WeaviateConfig @@ -270,6 +271,7 @@ class MiddlewareConfig( OpenSearchConfig, OracleConfig, PGVectorConfig, + VastbaseVectorConfig, PGVectoRSConfig, QdrantConfig, RelytConfig, diff --git a/api/configs/middleware/vdb/lindorm_config.py b/api/configs/middleware/vdb/lindorm_config.py index 95e1d1cfca..e80e3f4a35 100644 --- a/api/configs/middleware/vdb/lindorm_config.py +++ b/api/configs/middleware/vdb/lindorm_config.py @@ -32,3 +32,4 @@ class LindormConfig(BaseSettings): description="Using UGC index will store the same type of Index in a single index but can retrieve separately.", default=False, ) + LINDORM_QUERY_TIMEOUT: Optional[float] = Field(description="The lindorm search request timeout (s)", default=2.0) diff --git a/api/configs/middleware/vdb/vastbase_vector_config.py b/api/configs/middleware/vdb/vastbase_vector_config.py new file mode 100644 index 0000000000..816d6df90a --- /dev/null +++ b/api/configs/middleware/vdb/vastbase_vector_config.py @@ -0,0 +1,45 @@ +from typing import Optional + +from pydantic import Field, PositiveInt +from pydantic_settings import BaseSettings + + +class VastbaseVectorConfig(BaseSettings): + """ + Configuration settings for Vector (Vastbase with vector extension) + """ + + VASTBASE_HOST: Optional[str] = Field( + description="Hostname or IP address of the Vastbase server with Vector extension (e.g., 'localhost')", + default=None, + ) + + VASTBASE_PORT: PositiveInt = Field( + description="Port number on which the Vastbase server is listening (default is 5432)", + default=5432, + ) + + VASTBASE_USER: Optional[str] = Field( + description="Username for authenticating with the Vastbase database", + default=None, + ) + + VASTBASE_PASSWORD: Optional[str] = Field( + description="Password for authenticating with the Vastbase database", + default=None, + ) + + VASTBASE_DATABASE: Optional[str] = Field( + description="Name of the Vastbase database to connect to", + default=None, + ) + + VASTBASE_MIN_CONNECTION: PositiveInt = Field( + description="Min connection of the Vastbase database", + default=1, + ) + + VASTBASE_MAX_CONNECTION: PositiveInt = Field( + description="Max connection of the Vastbase database", + default=5, + ) diff --git a/api/controllers/console/datasets/datasets.py b/api/controllers/console/datasets/datasets.py index 752d124735..43615af709 100644 --- a/api/controllers/console/datasets/datasets.py +++ b/api/controllers/console/datasets/datasets.py @@ -657,6 +657,7 @@ class DatasetRetrievalSettingApi(Resource): | VectorType.ELASTICSEARCH | VectorType.ELASTICSEARCH_JA | VectorType.PGVECTOR + | VectorType.VASTBASE | VectorType.TIDB_ON_QDRANT | VectorType.LINDORM | VectorType.COUCHBASE @@ -706,6 +707,7 @@ class DatasetRetrievalSettingMockApi(Resource): | VectorType.ELASTICSEARCH_JA | VectorType.COUCHBASE | VectorType.PGVECTOR + | VectorType.VASTBASE | VectorType.LINDORM | VectorType.OPENGAUSS | VectorType.OCEANBASE diff --git a/api/controllers/service_api/app/conversation.py b/api/controllers/service_api/app/conversation.py index 334f2c5620..55600a3fd0 100644 --- a/api/controllers/service_api/app/conversation.py +++ b/api/controllers/service_api/app/conversation.py @@ -14,6 +14,9 @@ from fields.conversation_fields import ( conversation_infinite_scroll_pagination_fields, simple_conversation_fields, ) +from fields.conversation_variable_fields import ( + conversation_variable_infinite_scroll_pagination_fields, +) from libs.helper import uuid_value from models.model import App, AppMode, EndUser from services.conversation_service import ConversationService @@ -93,6 +96,31 @@ class ConversationRenameApi(Resource): raise NotFound("Conversation Not Exists.") +class ConversationVariablesApi(Resource): + @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.QUERY)) + @marshal_with(conversation_variable_infinite_scroll_pagination_fields) + def get(self, app_model: App, end_user: EndUser, c_id): + # conversational variable only for chat app + app_mode = AppMode.value_of(app_model.mode) + if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}: + raise NotChatAppError() + + conversation_id = str(c_id) + + parser = reqparse.RequestParser() + parser.add_argument("last_id", type=uuid_value, location="args") + parser.add_argument("limit", type=int_range(1, 100), required=False, default=20, location="args") + args = parser.parse_args() + + try: + return ConversationService.get_conversational_variable( + app_model, conversation_id, end_user, args["limit"], args["last_id"] + ) + except services.errors.conversation.ConversationNotExistsError: + raise NotFound("Conversation Not Exists.") + + api.add_resource(ConversationRenameApi, "/conversations//name", endpoint="conversation_name") api.add_resource(ConversationApi, "/conversations") api.add_resource(ConversationDetailApi, "/conversations/", endpoint="conversation_detail") +api.add_resource(ConversationVariablesApi, "/conversations//variables", endpoint="conversation_variables") diff --git a/api/core/app/apps/advanced_chat/generate_task_pipeline.py b/api/core/app/apps/advanced_chat/generate_task_pipeline.py index 3bf6c330db..baefca0c3f 100644 --- a/api/core/app/apps/advanced_chat/generate_task_pipeline.py +++ b/api/core/app/apps/advanced_chat/generate_task_pipeline.py @@ -684,7 +684,9 @@ class AdvancedChatAppGenerateTaskPipeline: ) elif isinstance(event, QueueMessageReplaceEvent): # published by moderation - yield self._message_cycle_manager._message_replace_to_stream_response(answer=event.text) + yield self._message_cycle_manager._message_replace_to_stream_response( + answer=event.text, reason=event.reason + ) elif isinstance(event, QueueAdvancedChatMessageEndEvent): if not graph_runtime_state: raise ValueError("graph runtime state not initialized.") @@ -695,7 +697,8 @@ class AdvancedChatAppGenerateTaskPipeline: if output_moderation_answer: self._task_state.answer = output_moderation_answer yield self._message_cycle_manager._message_replace_to_stream_response( - answer=output_moderation_answer + answer=output_moderation_answer, + reason=QueueMessageReplaceEvent.MessageReplaceReason.OUTPUT_MODERATION, ) # Save message diff --git a/api/core/app/entities/queue_entities.py b/api/core/app/entities/queue_entities.py index 3702326406..7228020e9b 100644 --- a/api/core/app/entities/queue_entities.py +++ b/api/core/app/entities/queue_entities.py @@ -264,8 +264,16 @@ class QueueMessageReplaceEvent(AppQueueEvent): QueueMessageReplaceEvent entity """ + class MessageReplaceReason(StrEnum): + """ + Reason for message replace event + """ + + OUTPUT_MODERATION = "output_moderation" + event: QueueEvent = QueueEvent.MESSAGE_REPLACE text: str + reason: str class QueueRetrieverResourcesEvent(AppQueueEvent): diff --git a/api/core/app/entities/task_entities.py b/api/core/app/entities/task_entities.py index f23ee1b9fd..817699bd20 100644 --- a/api/core/app/entities/task_entities.py +++ b/api/core/app/entities/task_entities.py @@ -148,6 +148,7 @@ class MessageReplaceStreamResponse(StreamResponse): event: StreamEvent = StreamEvent.MESSAGE_REPLACE answer: str + reason: str class AgentThoughtStreamResponse(StreamResponse): diff --git a/api/core/app/task_pipeline/based_generate_task_pipeline.py b/api/core/app/task_pipeline/based_generate_task_pipeline.py index a2e06d4e1f..5331c0cc94 100644 --- a/api/core/app/task_pipeline/based_generate_task_pipeline.py +++ b/api/core/app/task_pipeline/based_generate_task_pipeline.py @@ -126,12 +126,12 @@ class BasedGenerateTaskPipeline: if self._output_moderation_handler: self._output_moderation_handler.stop_thread() - completion = self._output_moderation_handler.moderation_completion( + completion, flagged = self._output_moderation_handler.moderation_completion( completion=completion, public_event=False ) self._output_moderation_handler = None - - return completion + if flagged: + return completion return None diff --git a/api/core/app/task_pipeline/message_cycle_manage.py b/api/core/app/task_pipeline/message_cycle_manage.py index 6223b33b67..fde506639f 100644 --- a/api/core/app/task_pipeline/message_cycle_manage.py +++ b/api/core/app/task_pipeline/message_cycle_manage.py @@ -182,10 +182,12 @@ class MessageCycleManage: from_variable_selector=from_variable_selector, ) - def _message_replace_to_stream_response(self, answer: str) -> MessageReplaceStreamResponse: + def _message_replace_to_stream_response(self, answer: str, reason: str = "") -> MessageReplaceStreamResponse: """ Message replace to stream response. :param answer: answer :return: """ - return MessageReplaceStreamResponse(task_id=self._application_generate_entity.task_id, answer=answer) + return MessageReplaceStreamResponse( + task_id=self._application_generate_entity.task_id, answer=answer, reason=reason + ) diff --git a/api/core/moderation/output_moderation.py b/api/core/moderation/output_moderation.py index e595be126c..2ec315417f 100644 --- a/api/core/moderation/output_moderation.py +++ b/api/core/moderation/output_moderation.py @@ -46,14 +46,14 @@ class OutputModeration(BaseModel): if not self.thread: self.thread = self.start_thread() - def moderation_completion(self, completion: str, public_event: bool = False) -> str: + def moderation_completion(self, completion: str, public_event: bool = False) -> tuple[str, bool]: self.buffer = completion self.is_final_chunk = True result = self.moderation(tenant_id=self.tenant_id, app_id=self.app_id, moderation_buffer=completion) if not result or not result.flagged: - return completion + return completion, False if result.action == ModerationAction.DIRECT_OUTPUT: final_output = result.preset_response @@ -61,9 +61,14 @@ class OutputModeration(BaseModel): final_output = result.text if public_event: - self.queue_manager.publish(QueueMessageReplaceEvent(text=final_output), PublishFrom.TASK_PIPELINE) + self.queue_manager.publish( + QueueMessageReplaceEvent( + text=final_output, reason=QueueMessageReplaceEvent.MessageReplaceReason.OUTPUT_MODERATION + ), + PublishFrom.TASK_PIPELINE, + ) - return final_output + return final_output, True def start_thread(self) -> threading.Thread: buffer_size = dify_config.MODERATION_BUFFER_SIZE @@ -112,7 +117,12 @@ class OutputModeration(BaseModel): # trigger replace event if self.thread_running: - self.queue_manager.publish(QueueMessageReplaceEvent(text=final_output), PublishFrom.TASK_PIPELINE) + self.queue_manager.publish( + QueueMessageReplaceEvent( + text=final_output, reason=QueueMessageReplaceEvent.MessageReplaceReason.OUTPUT_MODERATION + ), + PublishFrom.TASK_PIPELINE, + ) if result.action == ModerationAction.DIRECT_OUTPUT: break diff --git a/api/core/rag/datasource/vdb/lindorm/lindorm_vector.py b/api/core/rag/datasource/vdb/lindorm/lindorm_vector.py index 643ac2df4e..e9ff1ce43d 100644 --- a/api/core/rag/datasource/vdb/lindorm/lindorm_vector.py +++ b/api/core/rag/datasource/vdb/lindorm/lindorm_vector.py @@ -32,6 +32,7 @@ class LindormVectorStoreConfig(BaseModel): username: Optional[str] = None password: Optional[str] = None using_ugc: Optional[bool] = False + request_timeout: Optional[float] = 1.0 # timeout units: s @model_validator(mode="before") @classmethod @@ -251,9 +252,9 @@ class LindormVectorStore(BaseVector): query = default_vector_search_query(query_vector=query_vector, k=top_k, filters=filters, **kwargs) try: - params = {} + params = {"timeout": self._client_config.request_timeout} if self._using_ugc: - params["routing"] = self._routing + params["routing"] = self._routing # type: ignore response = self._client.search(index=self._collection_name, body=query, params=params) except Exception: logger.exception(f"Error executing vector search, query: {query}") @@ -304,8 +305,8 @@ class LindormVectorStore(BaseVector): routing=routing, routing_field=self._routing_field, ) - - response = self._client.search(index=self._collection_name, body=full_text_query) + params = {"timeout": self._client_config.request_timeout} + response = self._client.search(index=self._collection_name, body=full_text_query, params=params) docs = [] for hit in response["hits"]["hits"]: docs.append( @@ -554,6 +555,7 @@ class LindormVectorStoreFactory(AbstractVectorFactory): username=dify_config.LINDORM_USERNAME, password=dify_config.LINDORM_PASSWORD, using_ugc=dify_config.USING_UGC_INDEX, + request_timeout=dify_config.LINDORM_QUERY_TIMEOUT, ) using_ugc = dify_config.USING_UGC_INDEX if using_ugc is None: diff --git a/api/core/rag/datasource/vdb/pyvastbase/__init__.py b/api/core/rag/datasource/vdb/pyvastbase/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/api/core/rag/datasource/vdb/pyvastbase/vastbase_vector.py b/api/core/rag/datasource/vdb/pyvastbase/vastbase_vector.py new file mode 100644 index 0000000000..a61d571e16 --- /dev/null +++ b/api/core/rag/datasource/vdb/pyvastbase/vastbase_vector.py @@ -0,0 +1,243 @@ +import json +import uuid +from contextlib import contextmanager +from typing import Any + +import psycopg2.extras # type: ignore +import psycopg2.pool # type: ignore +from pydantic import BaseModel, model_validator + +from configs import dify_config +from core.rag.datasource.vdb.vector_base import BaseVector +from core.rag.datasource.vdb.vector_factory import AbstractVectorFactory +from core.rag.datasource.vdb.vector_type import VectorType +from core.rag.embedding.embedding_base import Embeddings +from core.rag.models.document import Document +from extensions.ext_redis import redis_client +from models.dataset import Dataset + + +class VastbaseVectorConfig(BaseModel): + host: str + port: int + user: str + password: str + database: str + min_connection: int + max_connection: int + + @model_validator(mode="before") + @classmethod + def validate_config(cls, values: dict) -> dict: + if not values["host"]: + raise ValueError("config VASTBASE_HOST is required") + if not values["port"]: + raise ValueError("config VASTBASE_PORT is required") + if not values["user"]: + raise ValueError("config VASTBASE_USER is required") + if not values["password"]: + raise ValueError("config VASTBASE_PASSWORD is required") + if not values["database"]: + raise ValueError("config VASTBASE_DATABASE is required") + if not values["min_connection"]: + raise ValueError("config VASTBASE_MIN_CONNECTION is required") + if not values["max_connection"]: + raise ValueError("config VASTBASE_MAX_CONNECTION is required") + if values["min_connection"] > values["max_connection"]: + raise ValueError("config VASTBASE_MIN_CONNECTION should less than VASTBASE_MAX_CONNECTION") + return values + + +SQL_CREATE_TABLE = """ +CREATE TABLE IF NOT EXISTS {table_name} ( + id UUID PRIMARY KEY, + text TEXT NOT NULL, + meta JSONB NOT NULL, + embedding floatvector({dimension}) NOT NULL +); +""" + +SQL_CREATE_INDEX = """ +CREATE INDEX IF NOT EXISTS embedding_cosine_v1_idx ON {table_name} +USING hnsw (embedding floatvector_cosine_ops) WITH (m = 16, ef_construction = 64); +""" + + +class VastbaseVector(BaseVector): + def __init__(self, collection_name: str, config: VastbaseVectorConfig): + super().__init__(collection_name) + self.pool = self._create_connection_pool(config) + self.table_name = f"embedding_{collection_name}" + + def get_type(self) -> str: + return VectorType.VASTBASE + + def _create_connection_pool(self, config: VastbaseVectorConfig): + return psycopg2.pool.SimpleConnectionPool( + config.min_connection, + config.max_connection, + host=config.host, + port=config.port, + user=config.user, + password=config.password, + database=config.database, + ) + + @contextmanager + def _get_cursor(self): + conn = self.pool.getconn() + cur = conn.cursor() + try: + yield cur + finally: + cur.close() + conn.commit() + self.pool.putconn(conn) + + def create(self, texts: list[Document], embeddings: list[list[float]], **kwargs): + dimension = len(embeddings[0]) + self._create_collection(dimension) + return self.add_texts(texts, embeddings) + + def add_texts(self, documents: list[Document], embeddings: list[list[float]], **kwargs): + values = [] + pks = [] + for i, doc in enumerate(documents): + if doc.metadata is not None: + doc_id = doc.metadata.get("doc_id", str(uuid.uuid4())) + pks.append(doc_id) + values.append( + ( + doc_id, + doc.page_content, + json.dumps(doc.metadata), + embeddings[i], + ) + ) + with self._get_cursor() as cur: + psycopg2.extras.execute_values( + cur, f"INSERT INTO {self.table_name} (id, text, meta, embedding) VALUES %s", values + ) + return pks + + def text_exists(self, id: str) -> bool: + with self._get_cursor() as cur: + cur.execute(f"SELECT id FROM {self.table_name} WHERE id = %s", (id,)) + return cur.fetchone() is not None + + def get_by_ids(self, ids: list[str]) -> list[Document]: + with self._get_cursor() as cur: + cur.execute(f"SELECT meta, text FROM {self.table_name} WHERE id IN %s", (tuple(ids),)) + docs = [] + for record in cur: + docs.append(Document(page_content=record[1], metadata=record[0])) + return docs + + def delete_by_ids(self, ids: list[str]) -> None: + # Avoiding crashes caused by performing delete operations on empty lists in certain scenarios + # Scenario 1: extract a document fails, resulting in a table not being created. + # Then clicking the retry button triggers a delete operation on an empty list. + if not ids: + return + with self._get_cursor() as cur: + cur.execute(f"DELETE FROM {self.table_name} WHERE id IN %s", (tuple(ids),)) + + def delete_by_metadata_field(self, key: str, value: str) -> None: + with self._get_cursor() as cur: + cur.execute(f"DELETE FROM {self.table_name} WHERE meta->>%s = %s", (key, value)) + + def search_by_vector(self, query_vector: list[float], **kwargs: Any) -> list[Document]: + """ + Search the nearest neighbors to a vector. + + :param query_vector: The input vector to search for similar items. + :param top_k: The number of nearest neighbors to return, default is 5. + :return: List of Documents that are nearest to the query vector. + """ + top_k = kwargs.get("top_k", 4) + + if not isinstance(top_k, int) or top_k <= 0: + raise ValueError("top_k must be a positive integer") + with self._get_cursor() as cur: + cur.execute( + f"SELECT meta, text, embedding <=> %s AS distance FROM {self.table_name}" + f" ORDER BY distance LIMIT {top_k}", + (json.dumps(query_vector),), + ) + docs = [] + score_threshold = float(kwargs.get("score_threshold") or 0.0) + for record in cur: + metadata, text, distance = record + score = 1 - distance + metadata["score"] = score + if score > score_threshold: + docs.append(Document(page_content=text, metadata=metadata)) + return docs + + def search_by_full_text(self, query: str, **kwargs: Any) -> list[Document]: + top_k = kwargs.get("top_k", 5) + + if not isinstance(top_k, int) or top_k <= 0: + raise ValueError("top_k must be a positive integer") + with self._get_cursor() as cur: + cur.execute( + f"""SELECT meta, text, ts_rank(to_tsvector(coalesce(text, '')), plainto_tsquery(%s)) AS score + FROM {self.table_name} + WHERE to_tsvector(text) @@ plainto_tsquery(%s) + ORDER BY score DESC + LIMIT {top_k}""", + # f"'{query}'" is required in order to account for whitespace in query + (f"'{query}'", f"'{query}'"), + ) + + docs = [] + + for record in cur: + metadata, text, score = record + metadata["score"] = score + docs.append(Document(page_content=text, metadata=metadata)) + + return docs + + def delete(self) -> None: + with self._get_cursor() as cur: + cur.execute(f"DROP TABLE IF EXISTS {self.table_name}") + + def _create_collection(self, dimension: int): + cache_key = f"vector_indexing_{self._collection_name}" + lock_name = f"{cache_key}_lock" + with redis_client.lock(lock_name, timeout=20): + collection_exist_cache_key = f"vector_indexing_{self._collection_name}" + if redis_client.get(collection_exist_cache_key): + return + + with self._get_cursor() as cur: + cur.execute(SQL_CREATE_TABLE.format(table_name=self.table_name, dimension=dimension)) + # Vastbase 支持的向量维度取值范围为 [1,16000] + if dimension <= 16000: + cur.execute(SQL_CREATE_INDEX.format(table_name=self.table_name)) + redis_client.set(collection_exist_cache_key, 1, ex=3600) + + +class VastbaseVectorFactory(AbstractVectorFactory): + def init_vector(self, dataset: Dataset, attributes: list, embeddings: Embeddings) -> VastbaseVector: + if dataset.index_struct_dict: + class_prefix: str = dataset.index_struct_dict["vector_store"]["class_prefix"] + collection_name = class_prefix + else: + dataset_id = dataset.id + collection_name = Dataset.gen_collection_name_by_id(dataset_id) + dataset.index_struct = json.dumps(self.gen_index_struct_dict(VectorType.VASTBASE, collection_name)) + + return VastbaseVector( + collection_name=collection_name, + config=VastbaseVectorConfig( + host=dify_config.VASTBASE_HOST or "localhost", + port=dify_config.VASTBASE_PORT, + user=dify_config.VASTBASE_USER or "dify", + password=dify_config.VASTBASE_PASSWORD or "", + database=dify_config.VASTBASE_DATABASE or "dify", + min_connection=dify_config.VASTBASE_MIN_CONNECTION, + max_connection=dify_config.VASTBASE_MAX_CONNECTION, + ), + ) diff --git a/api/core/rag/datasource/vdb/vector_factory.py b/api/core/rag/datasource/vdb/vector_factory.py index 05158cc7ca..66e002312a 100644 --- a/api/core/rag/datasource/vdb/vector_factory.py +++ b/api/core/rag/datasource/vdb/vector_factory.py @@ -74,6 +74,10 @@ class Vector: from core.rag.datasource.vdb.pgvector.pgvector import PGVectorFactory return PGVectorFactory + case VectorType.VASTBASE: + from core.rag.datasource.vdb.pyvastbase.vastbase_vector import VastbaseVectorFactory + + return VastbaseVectorFactory case VectorType.PGVECTO_RS: from core.rag.datasource.vdb.pgvecto_rs.pgvecto_rs import PGVectoRSFactory diff --git a/api/core/rag/datasource/vdb/vector_type.py b/api/core/rag/datasource/vdb/vector_type.py index 0421be3458..7a81565e37 100644 --- a/api/core/rag/datasource/vdb/vector_type.py +++ b/api/core/rag/datasource/vdb/vector_type.py @@ -7,7 +7,9 @@ class VectorType(StrEnum): MILVUS = "milvus" MYSCALE = "myscale" PGVECTOR = "pgvector" + VASTBASE = "vastbase" PGVECTO_RS = "pgvecto-rs" + QDRANT = "qdrant" RELYT = "relyt" TIDB_VECTOR = "tidb_vector" diff --git a/api/core/tools/builtin_tool/provider.py b/api/core/tools/builtin_tool/provider.py index 4f733f0ea1..cf75bd3d7e 100644 --- a/api/core/tools/builtin_tool/provider.py +++ b/api/core/tools/builtin_tool/provider.py @@ -35,8 +35,9 @@ class BuiltinToolProviderController(ToolProviderController): provider_yaml["credentials_for_provider"][credential_name]["name"] = credential_name credentials_schema = [] - for credential in provider_yaml.get("credentials_for_provider", {}).values(): - credentials_schema.append(credential) + for credential in provider_yaml.get("credentials_for_provider", {}): + credential_dict = provider_yaml.get("credentials_for_provider", {}).get(credential, {}) + credentials_schema.append(credential_dict) super().__init__( entity=ToolProviderEntity( diff --git a/api/fields/conversation_variable_fields.py b/api/fields/conversation_variable_fields.py index c6385efb5a..3aa3838def 100644 --- a/api/fields/conversation_variable_fields.py +++ b/api/fields/conversation_variable_fields.py @@ -19,3 +19,9 @@ paginated_conversation_variable_fields = { "has_more": fields.Boolean, "data": fields.List(fields.Nested(conversation_variable_fields), attribute="data"), } + +conversation_variable_infinite_scroll_pagination_fields = { + "limit": fields.Integer, + "has_more": fields.Boolean, + "data": fields.List(fields.Nested(conversation_variable_fields)), +} diff --git a/api/pyproject.toml b/api/pyproject.toml index 4992178423..6f526bc705 100644 --- a/api/pyproject.toml +++ b/api/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "dify-api" -version = "1.2.0" +version = "1.3.0" requires-python = ">=3.11,<3.13" dependencies = [ diff --git a/api/services/conversation_service.py b/api/services/conversation_service.py index 6485cbf37d..afdaa49465 100644 --- a/api/services/conversation_service.py +++ b/api/services/conversation_service.py @@ -9,9 +9,14 @@ from core.app.entities.app_invoke_entities import InvokeFrom from core.llm_generator.llm_generator import LLMGenerator from extensions.ext_database import db from libs.infinite_scroll_pagination import InfiniteScrollPagination +from models import ConversationVariable from models.account import Account from models.model import App, Conversation, EndUser, Message -from services.errors.conversation import ConversationNotExistsError, LastConversationNotExistsError +from services.errors.conversation import ( + ConversationNotExistsError, + ConversationVariableNotExistsError, + LastConversationNotExistsError, +) from services.errors.message import MessageNotExistsError @@ -166,3 +171,50 @@ class ConversationService: conversation.is_deleted = True conversation.updated_at = datetime.now(UTC).replace(tzinfo=None) db.session.commit() + + @classmethod + def get_conversational_variable( + cls, + app_model: App, + conversation_id: str, + user: Optional[Union[Account, EndUser]], + limit: int, + last_id: Optional[str], + ) -> InfiniteScrollPagination: + conversation = cls.get_conversation(app_model, conversation_id, user) + + stmt = ( + select(ConversationVariable) + .where(ConversationVariable.app_id == app_model.id) + .where(ConversationVariable.conversation_id == conversation.id) + .order_by(ConversationVariable.created_at) + ) + + with Session(db.engine) as session: + if last_id: + last_variable = session.scalar(stmt.where(ConversationVariable.id == last_id)) + if not last_variable: + raise ConversationVariableNotExistsError() + + # Filter for variables created after the last_id + stmt = stmt.where(ConversationVariable.created_at > last_variable.created_at) + + # Apply limit to query + query_stmt = stmt.limit(limit) # Get one extra to check if there are more + rows = session.scalars(query_stmt).all() + + has_more = False + if len(rows) > limit: + has_more = True + rows = rows[:limit] # Remove the extra item + + variables = [ + { + "created_at": row.created_at, + "updated_at": row.updated_at, + **row.to_variable().model_dump(), + } + for row in rows + ] + + return InfiniteScrollPagination(variables, limit, has_more) diff --git a/api/services/errors/conversation.py b/api/services/errors/conversation.py index 139dd9a70a..f8051e3417 100644 --- a/api/services/errors/conversation.py +++ b/api/services/errors/conversation.py @@ -11,3 +11,7 @@ class ConversationNotExistsError(BaseServiceError): class ConversationCompletedError(Exception): pass + + +class ConversationVariableNotExistsError(BaseServiceError): + pass diff --git a/api/tests/integration_tests/vdb/pyvastbase/__init__.py b/api/tests/integration_tests/vdb/pyvastbase/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/api/tests/integration_tests/vdb/pyvastbase/test_vastbase_vector.py b/api/tests/integration_tests/vdb/pyvastbase/test_vastbase_vector.py new file mode 100644 index 0000000000..3d7873442b --- /dev/null +++ b/api/tests/integration_tests/vdb/pyvastbase/test_vastbase_vector.py @@ -0,0 +1,27 @@ +from core.rag.datasource.vdb.pyvastbase.vastbase_vector import VastbaseVector, VastbaseVectorConfig +from tests.integration_tests.vdb.test_vector_store import ( + AbstractVectorTest, + get_example_text, + setup_mock_redis, +) + + +class VastbaseVectorTest(AbstractVectorTest): + def __init__(self): + super().__init__() + self.vector = VastbaseVector( + collection_name=self.collection_name, + config=VastbaseVectorConfig( + host="localhost", + port=5434, + user="dify", + password="Difyai123456", + database="dify", + min_connection=1, + max_connection=5, + ), + ) + + +def test_vastbase_vector(setup_mock_redis): + VastbaseVectorTest().run_all_tests() diff --git a/api/uv.lock b/api/uv.lock index 6c8699dd7c..0e2dd9c92a 100644 --- a/api/uv.lock +++ b/api/uv.lock @@ -1148,7 +1148,7 @@ wheels = [ [[package]] name = "dify-api" -version = "1.2.0" +version = "1.3.0" source = { virtual = "." } dependencies = [ { name = "authlib" }, diff --git a/docker/.env.example b/docker/.env.example index 0b80dccb37..83d975cec5 100644 --- a/docker/.env.example +++ b/docker/.env.example @@ -406,6 +406,7 @@ QDRANT_GRPC_PORT=6334 # Milvus configuration. Only available when VECTOR_STORE is `milvus`. # The milvus uri. MILVUS_URI=http://host.docker.internal:19530 +MILVUS_DATABASE= MILVUS_TOKEN= MILVUS_USER= MILVUS_PASSWORD= @@ -441,6 +442,15 @@ PGVECTOR_MAX_CONNECTION=5 PGVECTOR_PG_BIGM=false PGVECTOR_PG_BIGM_VERSION=1.2-20240606 +# vastbase configurations, only available when VECTOR_STORE is `vastbase` +VASTBASE_HOST=vastbase +VASTBASE_PORT=5432 +VASTBASE_USER=dify +VASTBASE_PASSWORD=Difyai123456 +VASTBASE_DATABASE=dify +VASTBASE_MIN_CONNECTION=1 +VASTBASE_MAX_CONNECTION=5 + # pgvecto-rs configurations, only available when VECTOR_STORE is `pgvecto-rs` PGVECTO_RS_HOST=pgvecto-rs PGVECTO_RS_PORT=5432 @@ -553,6 +563,7 @@ VIKINGDB_SOCKET_TIMEOUT=30 LINDORM_URL=http://lindorm:30070 LINDORM_USERNAME=lindorm LINDORM_PASSWORD=lindorm +LINDORM_QUERY_TIMEOUT=1 # OceanBase Vector configuration, only available when VECTOR_STORE is `oceanbase` OCEANBASE_VECTOR_HOST=oceanbase diff --git a/docker/docker-compose-template.yaml b/docker/docker-compose-template.yaml index 8c57a7c4c2..9ab1304492 100644 --- a/docker/docker-compose-template.yaml +++ b/docker/docker-compose-template.yaml @@ -142,7 +142,7 @@ services: # plugin daemon plugin_daemon: - image: langgenius/dify-plugin-daemon:0.0.8-local + image: langgenius/dify-plugin-daemon:0.0.9-local restart: always environment: # Use the shared environment variables. @@ -363,6 +363,30 @@ services: timeout: 3s retries: 30 + # get image from https://www.vastdata.com.cn/ + vastbase: + image: vastdata/vastbase-vector + profiles: + - vastbase + restart: always + environment: + - VB_DBCOMPATIBILITY=PG + - VB_DB=dify + - VB_USERNAME=dify + - VB_PASSWORD=Difyai123456 + ports: + - '5434:5432' + volumes: + - ./vastbase/lic:/home/vastbase/vastbase/lic + - ./vastbase/data:/home/vastbase/data + - ./vastbase/backup:/home/vastbase/backup + - ./vastbase/backup_log:/home/vastbase/backup_log + healthcheck: + test: [ 'CMD', 'pg_isready' ] + interval: 1s + timeout: 3s + retries: 30 + # pgvecto-rs vector store pgvecto-rs: image: tensorchord/pgvecto-rs:pg16-v0.3.0 diff --git a/docker/docker-compose.middleware.yaml b/docker/docker-compose.middleware.yaml index fc08edd264..01c7573a95 100644 --- a/docker/docker-compose.middleware.yaml +++ b/docker/docker-compose.middleware.yaml @@ -71,7 +71,7 @@ services: # plugin daemon plugin_daemon: - image: langgenius/dify-plugin-daemon:0.0.8-local + image: langgenius/dify-plugin-daemon:0.0.9-local restart: always env_file: - ./middleware.env diff --git a/docker/docker-compose.yaml b/docker/docker-compose.yaml index 3d3e3a901f..8edcd497c6 100644 --- a/docker/docker-compose.yaml +++ b/docker/docker-compose.yaml @@ -138,6 +138,7 @@ x-shared-env: &shared-api-worker-env QDRANT_GRPC_ENABLED: ${QDRANT_GRPC_ENABLED:-false} QDRANT_GRPC_PORT: ${QDRANT_GRPC_PORT:-6334} MILVUS_URI: ${MILVUS_URI:-http://host.docker.internal:19530} + MILVUS_DATABASE: ${MILVUS_DATABASE:-} MILVUS_TOKEN: ${MILVUS_TOKEN:-} MILVUS_USER: ${MILVUS_USER:-} MILVUS_PASSWORD: ${MILVUS_PASSWORD:-} @@ -163,6 +164,13 @@ x-shared-env: &shared-api-worker-env PGVECTOR_MAX_CONNECTION: ${PGVECTOR_MAX_CONNECTION:-5} PGVECTOR_PG_BIGM: ${PGVECTOR_PG_BIGM:-false} PGVECTOR_PG_BIGM_VERSION: ${PGVECTOR_PG_BIGM_VERSION:-1.2-20240606} + VASTBASE_HOST: ${VASTBASE_HOST:-vastbase} + VASTBASE_PORT: ${VASTBASE_PORT:-5432} + VASTBASE_USER: ${VASTBASE_USER:-dify} + VASTBASE_PASSWORD: ${VASTBASE_PASSWORD:-Difyai123456} + VASTBASE_DATABASE: ${VASTBASE_DATABASE:-dify} + VASTBASE_MIN_CONNECTION: ${VASTBASE_MIN_CONNECTION:-1} + VASTBASE_MAX_CONNECTION: ${VASTBASE_MAX_CONNECTION:-5} PGVECTO_RS_HOST: ${PGVECTO_RS_HOST:-pgvecto-rs} PGVECTO_RS_PORT: ${PGVECTO_RS_PORT:-5432} PGVECTO_RS_USER: ${PGVECTO_RS_USER:-postgres} @@ -250,6 +258,7 @@ x-shared-env: &shared-api-worker-env LINDORM_URL: ${LINDORM_URL:-http://lindorm:30070} LINDORM_USERNAME: ${LINDORM_USERNAME:-lindorm} LINDORM_PASSWORD: ${LINDORM_PASSWORD:-lindorm} + LINDORM_QUERY_TIMEOUT: ${LINDORM_QUERY_TIMEOUT:-1} OCEANBASE_VECTOR_HOST: ${OCEANBASE_VECTOR_HOST:-oceanbase} OCEANBASE_VECTOR_PORT: ${OCEANBASE_VECTOR_PORT:-2881} OCEANBASE_VECTOR_USER: ${OCEANBASE_VECTOR_USER:-root@test} @@ -619,7 +628,7 @@ services: # plugin daemon plugin_daemon: - image: langgenius/dify-plugin-daemon:0.0.8-local + image: langgenius/dify-plugin-daemon:0.0.9-local restart: always environment: # Use the shared environment variables. @@ -840,6 +849,30 @@ services: timeout: 3s retries: 30 + # get image from https://www.vastdata.com.cn/ + vastbase: + image: vastdata/vastbase-vector + profiles: + - vastbase + restart: always + environment: + - VB_DBCOMPATIBILITY=PG + - VB_DB=dify + - VB_USERNAME=dify + - VB_PASSWORD=Difyai123456 + ports: + - '5434:5432' + volumes: + - ./vastbase/lic:/home/vastbase/vastbase/lic + - ./vastbase/data:/home/vastbase/data + - ./vastbase/backup:/home/vastbase/backup + - ./vastbase/backup_log:/home/vastbase/backup_log + healthcheck: + test: [ 'CMD', 'pg_isready' ] + interval: 1s + timeout: 3s + retries: 30 + # pgvecto-rs vector store pgvecto-rs: image: tensorchord/pgvecto-rs:pg16-v0.3.0 diff --git a/web/app/components/app/configuration/dataset-config/select-dataset/index.tsx b/web/app/components/app/configuration/dataset-config/select-dataset/index.tsx index 64a7f6dc46..ffdb714f08 100644 --- a/web/app/components/app/configuration/dataset-config/select-dataset/index.tsx +++ b/web/app/components/app/configuration/dataset-config/select-dataset/index.tsx @@ -14,6 +14,7 @@ import Loading from '@/app/components/base/loading' import Badge from '@/app/components/base/badge' import { useKnowledge } from '@/hooks/use-knowledge' import cn from '@/utils/classnames' +import { basePath } from '@/utils/var' export type ISelectDataSetProps = { isShow: boolean @@ -111,7 +112,7 @@ const SelectDataSet: FC = ({ }} > {t('appDebug.feature.dataSet.noDataSet')} - {t('appDebug.feature.dataSet.toCreate')} + {t('appDebug.feature.dataSet.toCreate')} )} diff --git a/web/app/components/app/overview/embedded/index.tsx b/web/app/components/app/overview/embedded/index.tsx index 6ebd0fce69..691b727b8e 100644 --- a/web/app/components/app/overview/embedded/index.tsx +++ b/web/app/components/app/overview/embedded/index.tsx @@ -29,7 +29,7 @@ const OPTION_MAP = { iframe: { getContent: (url: string, token: string) => `