apify对接opengauss向量数据库

This commit is contained in:
Yongshun Wang 2025-09-23 16:18:29 +08:00
parent 3790d3f3c8
commit b1e92f55f7
46 changed files with 7666 additions and 0 deletions

View File

@ -0,0 +1,101 @@
# type: ignore
import os
from apify_client import ApifyClient
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_opengauss import OpenGauss, OpenGaussSettings
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
APIFY_API_TOKEN = os.getenv("APIFY_API_TOKEN") or "YOUR-APIFY-TOKEN"
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") or "YOUR-OPENAI-API-KEY"
OPENGAUSS_HOST = os.getenv("OPENGAUSS_HOST")
OPENGAUSS_PORT = os.getenv("OPENGAUSS_PORT")
OPENGAUSS_USER = os.getenv("OPENGAUSS_USER")
OPENGAUSS_PASSWORD = os.getenv("OPENGAUSS_PASSWORD")
OPENGAUSS_DBNAME = os.getenv("OPENGAUSS_DBNAME")
OPENGAUSS_TABLE_NAME = os.getenv("OPENGAUSS_TABLE_NAME")
client = ApifyClient(APIFY_API_TOKEN)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
print("Starting Apify's Website Content Crawler")
print("Crawling will take some time ... you can check the progress in the Apify console")
actor_call = client.actor(actor_id="apify/website-content-crawler").call(
run_input={"maxCrawlPages": 10, "startUrls": [{"url": "https://opengauss.org/"}]}
)
print("Actor website content crawler has finished")
print(actor_call)
opengauss_integration_inputs = {
"opengaussHost": OPENGAUSS_HOST,
"opengaussPort": OPENGAUSS_PORT,
"opengaussUser": OPENGAUSS_USER,
"opengaussPassword": OPENGAUSS_PASSWORD,
"opengaussDBname": OPENGAUSS_DBNAME,
"opengaussTableName": OPENGAUSS_TABLE_NAME,
"datasetFields": ["text"],
"datasetId": actor_call["defaultDatasetId"],
"deltaUpdatesPrimaryDatasetFields": ["url"],
"expiredObjectDeletionPeriodDays": 7,
"embeddingsApiKey": OPENAI_API_KEY,
"embeddingsConfig": {
"model": "text-embedding-3-small",
},
"embeddingsProvider": "OpenAI",
"performChunking": True,
"chunkSize": 2000,
"chunkOverlap": 200
}
print("Starting Apify's OpenGauss Integration")
actor_call = client.actor("wyswyz/opengauss-integration").call(run_input=opengauss_integration_inputs)
print("Apify's OpenGauss Integration has finished")
print(actor_call)
print("Question answering using OpenGauss database")
config = OpenGaussSettings(
host=OPENGAUSS_HOST,
port=OPENGAUSS_PORT,
user=OPENGAUSS_USER,
password=OPENGAUSS_PASSWORD,
database=OPENGAUSS_DBNAME,
table_name=OPENGAUSS_TABLE_NAME,
embedding_dimension=384,
index_type="HNSW",
distance_strategy="cosine",
)
vector_store = OpenGauss(embedding=embeddings, config=config)
prompt = PromptTemplate(
input_variables=["context", "question"],
template="Use the following pieces of retrieved context to answer the question. If you don't know the answer, "
"just say that you don't know. \nQuestion: {question} \nContext: {context} \nAnswer:",
)
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
rag_chain = (
{
"context": vector_store.as_retriever() | format_docs,
"question": RunnablePassthrough(),
}
| prompt
| ChatOpenAI(model="gpt-5-mini", temperature=1)
| StrOutputParser()
)
question = "什么是openGauss"
print("Question:", question)
print("Answer:", rag_chain.invoke(question))

View File

@ -0,0 +1,219 @@
# 使用 Apify 和 openGauss 构建企业级 RAG 问答系统:实战教程
本文将通过一个完整的实例,向您展示如何利用 Apify 强大的云端数据抓取与集成能力,结合 openGauss 数据库的向量存储功能,快速构建一个针对 openGauss 官方文档的 RAG 问答机器人。
**Apify 是一个云端 Web 抓取和自动化平台。** 它允许开发者轻松地从任何网站提取数据,或将任何网站的工作流程自动化。其核心是 **"Actors"**——可以执行任意任务的无服务器云程序。
Apify 平台的优势在于:
* **简化数据采集**您无需关心基础设施、IP 轮换、浏览器指纹等复杂问题Apify 会为您处理好一切。
* **丰富的 Actor 市场**Apify Store 中有大量预构建好的 Actors例如通用的网站内容爬虫Website Content Crawler、搜索引擎爬虫等可以开箱即用。
* **强大的集成能力**Apify 可以轻松地将采集到的数据推送到各种数据库、API 或云存储中,实现了从数据源到目的地的无缝衔接。
在本教程中,我们将使用两个关键的 Apify Actors
1. **`apify/website-content-crawler`**: 用于抓取 openGauss 官网的网页内容。
2. **`wyswyz/opengauss-integration`**: 用于将抓取到的内容进行处理(分块、生成向量),并自动存入 openGauss 数据库。
## **前置准备工作**
1. **Apify 账户**:注册一个 [Apify 账户](https://apify.com/) 并获取您的 API Token。
2. **OpenAI 账户**:获取您的 [OpenAI API Key](https://platform.openai.com/api-keys),用于内容向量化和问答生成。
3. **openGauss 数据库**:您需要一个的 openGauss 数据库实例,并且支持向量数据库功能
### 使用docker部署openGauss
**获取镜像**
```bash
$ docker pull opengauss/opengauss-server:latest
```
**查看镜像状态**
```bash
$ docker images
REPOSITORY TAG IMAGE ID CREATED SIZE
opengauss/opengauss-server latest 9763e8b26794 6 months ago 1.68GB
```
**运行容器**
```bash
$ docker run --name opengauss --privileged=true -d -e GS_PASSWORD=YourPassoword -p 8888:5432 opengauss/opengauss-server:latest
```
**验证容器运行状态**
```bash
$ docker ps
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
c5f0b44adf9a opengauss/opengauss-server:latest "entrypoint.sh gauss…" 7 weeks ago Up 7 weeks 0.0.0.0:8888->5432/tcp opengauss
```
至此已经成功用docker部署openGauss数据库
**配置环境变量**:为了安全和方便,建议将密钥和数据库连接信息配置为环境变量。
```bash
export APIFY_API_TOKEN="YOUR-APIFY-TOKEN"
export OPENAI_API_KEY="YOUR-OPENAI-API-KEY"
export OPENGAUSS_HOST="your-db-host"
export OPENGAUSS_PORT="your-db-port"
export OPENGAUSS_USER="your-db-user"
export OPENGAUSS_PASSWORD="your-db-password"
export OPENGAUSS_DBNAME="your-db-name"
export OPENGAUSS_TABLE_NAME="opengauss_docs"
```
**Python 环境**:安装必要的库。
```bash
pip install apify-client langchain-core langchain-opengauss langchain-openai
```
## **数据采集 - 使用 Apify 网站内容爬虫**
这是 RAG 流程的第一步:获取知识。我们使用 Apify 预构建的 `website-content-crawler` Actor 来抓取 openGauss 官网的内容。
```python
import os
from apify_client import ApifyClient
# ... [环境变量加载] ...
client = ApifyClient(APIFY_API_TOKEN)
print("Starting Apify's Website Content Crawler")
print("Crawling will take some time")
# 调用 Actor 开始抓取
actor_call = client.actor(actor_id="apify/website-content-crawler").call(
run_input={
"maxCrawlPages": 10,
"startUrls": [{"url": "https://opengauss.org/"}]
}
)
print("Actor website content crawler has finished")
print(actor_call)
```
website-content-crawler的爬取行为可以通过run_input进行定制我们选择爬取openGauss的主页。为了方便演示设置最大爬取页数为10。等待actor爬取完成。
```bash
2025-09-22T13:16:41.429Z INFO PlaywrightCrawler: Finished! Total 9 requests: 9 succeeded, 0 failed.
```
## **数据处理与入库 - 使用 Apify openGauss 集成 Actor**
现在我们有了原始的网页数据,下一步是将其处理并存入 openGauss 向量数据库。这正是 `opengauss-integration` Actor 的用武之地。它会自动完成**分块 (Chunking)、向量化 (Embedding) 和存储 (Storing)** 这一系列繁琐的工作。
首先配置 openGauss 数据库的参数,并设置我们需要的字段以及分块、向量化等参数,所有参数信息可以参考[actor详情页](https://apify.com/wyswyz/opengauss-integration)
```python
# 准备 openGauss 集成 Actor 的输入参数
opengauss_integration_inputs = {
# 数据库连接信息
"opengaussHost": OPENGAUSS_HOST,
"opengaussPort": OPENGAUSS_PORT,
"opengaussUser": OPENGAUSS_USER,
"opengaussPassword": OPENGAUSS_PASSWORD,
"opengaussDBname": OPENGAUSS_DBNAME,
"opengaussTableName": OPENGAUSS_TABLE_NAME,
# 数据源和处理配置
"datasetId": actor_call["defaultDatasetId"],
"datasetFields": ["text"],
"performChunking": True,
"chunkSize": 2000,
"chunkOverlap": 200,
# 向量化配置
"embeddingsProvider": "OpenAI",
"embeddingsApiKey": OPENAI_API_KEY,
"embeddingsConfig": {
"model": "text-embedding-3-small",
},
# 其他可选配置
"deltaUpdatesPrimaryDatasetFields": ["url"],
"expiredObjectDeletionPeriodDays": 7,
}
```
接下来我们调用`wyswyz/opengauss-integration`将数据存储到openGauss数据库中
```python
print("Starting Apify's OpenGauss Integration")
# 调用 openGauss 集成 Actor
actor_call = client.actor("wyswyz/opengauss-integration").call(run_input=opengauss_integration_inputs)
print("Apify's OpenGauss Integration has finished")
print(actor_call)
```
当这个 Actor 运行完毕后,您的 openGauss 数据库中指定的表(`opengauss_docs`)就已经包含了处理好的知识数据和对应的向量。
## **构建 RAG 应用 - 使用 LangChain 进行问答**
现在我们的知识库已经准备就绪,可以使用 LangChain 来构建一个问答应用了。
```python
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_opengauss import OpenGauss, OpenGaussSettings
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
# ... [接上文] ...
print("Question answering using OpenGauss database")
# 1. 配置 openGauss 连接
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
config = OpenGaussSettings(
host=OPENGAUSS_HOST,
port=OPENGAUSS_PORT,
user=OPENGAUSS_USER,
password=OPENGAUSS_PASSWORD,
database=OPENGAUSS_DBNAME,
table_name=OPENGAUSS_TABLE_NAME,
embedding_dimension=384,
index_type="HNSW",
distance_strategy="cosine",
)
# 初始化 LangChain 的 openGauss 向量存储
vector_store = OpenGauss(embedding=embeddings, config=config)
# 2. 定义 Prompt 模板
prompt = PromptTemplate(
input_variables=["context", "question"],
template="Use the following pieces of retrieved context to answer the question. If you don't know the answer, "
"just say that you don't know. \nQuestion: {question} \nContext: {context} \nAnswer:",
)
# 3. 定义文档格式化函数
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
# 4. 构建 RAG 链 (Chain)
rag_chain = (
{
"context": vector_store.as_retriever() | format_docs,
"question": RunnablePassthrough(),
}
| prompt
| ChatOpenAI(model="gpt-5-mini", temperature=1)
| StrOutputParser()
)
# 5. 提问与回答
question = "什么是openGauss"
print("Question:", question)
print("Answer:", rag_chain.invoke(question))
```
结果如下
```bash
Question: 什么是openGauss
Answer: openGauss 是一个开源数据库及其社区(即 openGauss 社区)。该项目面向企业级应用,倡导开源开放、社区协作,提供完善的技术文档和实践案例(如资源池化、告警机制、容灾集群、一主两备部署、线程池与 RDMA 指导、在 openEuler 上安装、与 OpenStack 集成等),并通过 meetup 等线下/线上活动推动生态繁荣。
```
# 总结
通过本教程,我们利用 Apify 和 openGauss 成功构建了一个从数据采集到智能问答的端到端 RAG 应用。
此集成支持增量更新,仅更新已更改的数据。这种方法减少了不必要的嵌入计算和存储操作,使其适用于搜索和检索增强生成。
更多有关 Apify-openGauss 集成的信息,请参考集成[README文件](https://apify.com/wyswyz/opengauss-integration)。

View File

@ -0,0 +1,18 @@
# Files and folders to ignore during Docker build
.env/
.env.example
# Git
.git
.gitignore
# Docker
.docker
# Python
.pytest_cache
.idea
.mypy_cache
# mono-repo
code/.venv

View File

@ -0,0 +1,207 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[codz]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py.cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# UV
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
#uv.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
#poetry.toml
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
# pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python.
# https://pdm-project.org/en/latest/usage/project/#working-with-version-control
#pdm.lock
#pdm.toml
.pdm-python
.pdm-build/
# pixi
# Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control.
#pixi.lock
# Pixi creates a virtual environment in the .pixi directory, just like venv module creates one
# in the .venv directory. It is recommended not to include this directory in version control.
.pixi
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.envrc
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
# Abstra
# Abstra is an AI-powered process automation framework.
# Ignore directories containing user credentials, local state, and settings.
# Learn more at https://abstra.io/docs
.abstra/
# Visual Studio Code
# Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore
# that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore
# and can be added to the global gitignore or merged into this file. However, if you prefer,
# you could uncomment the following to ignore the entire vscode folder
# .vscode/
# Ruff stuff:
.ruff_cache/
# PyPI configuration file
.pypirc
# Cursor
# Cursor is an AI-powered code editor. `.cursorignore` specifies files/directories to
# exclude from AI features like autocomplete and code analysis. Recommended for sensitive data
# refer to https://docs.cursor.com/context/ignore-files
.cursorignore
.cursorindexingignore
# Marimo
marimo/_static/
marimo/_lsp/
__marimo__/

View File

@ -0,0 +1,14 @@
repos:
- repo: local
hooks:
- id: lint
name: Lint codebase
entry: make lint
language: system
pass_filenames: false
- id: type-check
name: Type-check codebase
entry: make type-check
language: system
pass_filenames: false

View File

@ -0,0 +1,201 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [name of copyright owner]
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.

View File

@ -0,0 +1,37 @@
.PHONY: clean install-dev lint type-check check-code format
DIRS_WITH_CODE = code
DIRS_WITH_ACTORS = actors
clean:
rm -rf .venv .mypy_cache .pytest_cache .ruff_cache __pycache__
install-dev:
cd $(DIRS_WITH_CODE) && pip install --upgrade pip poetry && poetry install --with main,dev,opengauss && poetry run pre-commit install && cd ..
lint:
poetry run -C $(DIRS_WITH_CODE) ruff check
type-check:
poetry run -C $(DIRS_WITH_CODE) mypy
check-code: lint type-check
format:
poetry run -C $(DIRS_WITH_CODE) ruff check --fix
poetry run -C $(DIRS_WITH_CODE) ruff format
pydantic-model:
datamodel-codegen --input $(DIRS_WITH_ACTORS)/opengauss/.actor/input_schema.json --output $(DIRS_WITH_CODE)/src/models/opengauss_input_model.py --input-file-type jsonschema --field-constraints --enum-field-as-literal all
# Integration tests are marked with @pytest.mark.integration_test
# You will require all databased running to run these tests.
# Check docker-compose.yml for the list of databases.
test-integration:
poetry run -C $(DIRS_WITH_CODE) pytest --with-integration
test-unit:
poetry run -C $(DIRS_WITH_CODE) pytest
test: test-unit test-integration

View File

@ -0,0 +1,28 @@
# Apify Actor for OpenGauss Integrations
This project was inspired by and derived from the official `apify/actor-vector-database-integrations` repository. Special thanks to the Apify team for their foundational work.
| Actor | Actor badge |
|-----------------------------|---------------------|
| [OpenGauss](https://opengauss.org/) | [![Opengauss integration](https://apify.com/actor-badge?actor=wyswyz/opengauss-integration)](https://apify.com/wyswyz/opengauss-integration) |
#### Vector database integrations (Actors)
The Apify Vector Database Integrations facilitate the transfer of data from Apify Actors to openGauss vector database.
This process includes data processing, optional splitting into chunks, embedding computation, and data storage
These integrations support incremental updates, ensuring that only changed data is updated.
This reduces unnecessary embedding computation and storage operations, making it ideal for search and retrieval augmented generation (RAG) use cases.
## How does it work?
1. Retrieve a dataset as output from an Actor.
2. _[Optional]_ Split text data into chunks using [langchain](https://python.langchain.com).
3. _[Optional]_ Update only changed data.
4. Compute embeddings, e.g. using [OpenAI](https://platform.openai.com/docs/guides/embeddings) or [Cohere](https://cohere.com/embeddings).
5. Save data into the database.
## Supported Vector Embeddings
- [OpenAI](https://platform.openai.com/docs/guides/embeddings)
- [Cohere](https://cohere.com/embeddings)

View File

@ -0,0 +1,15 @@
{
"actorSpecification": 1,
"name": "opengauss-integration",
"title": "openGauss Integration",
"description": "Upload a dataset to openGauss",
"version": "0.0",
"input": "./input_schema.json",
"dockerfile": "../../../shared/Dockerfile",
"readme": "./README.md",
"changelog":"../../../shared/CHANGELOG.md",
"storages": {
"dataset": "../../../shared/dataset_schema.json"
},
"dockerContextDir": "../../.."
}

View File

@ -0,0 +1,180 @@
{
"title": "openGauss-integration",
"type": "object",
"schemaVersion": 1,
"properties": {
"opengaussHost": {
"title": "openGauss Host",
"type": "string",
"description": "The Host of openGauss",
"editor": "textfield"
},
"opengaussPort": {
"title": "openGauss Port",
"type": "string",
"description": "The Port of openGauss",
"editor": "textfield"
},
"opengaussUser": {
"title": "openGauss User",
"type": "string",
"description": "The User of openGauss",
"editor": "textfield"
},
"opengaussPassword": {
"title": "openGauss Password",
"type": "string",
"description": "The Password of openGauss",
"editor": "textfield"
},
"opengaussDBname": {
"title": "openGauss DBname",
"type": "string",
"description": "The DBname of openGauss",
"editor": "textfield"
},
"opengaussTableName": {
"title": "openGauss SQL table name",
"type": "string",
"description": "The name of the table to use",
"editor": "textfield"
},
"embeddingsProvider": {
"title": "Embeddings provider (as defined in the langchain API)",
"description": "Choose the embeddings provider to use for generating embeddings",
"type": "string",
"editor": "select",
"enum": ["OpenAI", "Cohere"],
"default": "OpenAI",
"sectionCaption": "Embeddings settings"
},
"embeddingsConfig": {
"title": "Configuration for embeddings provider",
"description": "Configure the parameters for the LangChain embedding class. Key points to consider:\n\n1. Typically, you only need to specify the model name. For example, for OpenAI, set the model name as {\"model\": \"text-embedding-3-small\"}.\n\n2. It's required to ensure that the vector size of your embeddings matches the size of embeddings in the database.\n\n3. Here are examples of embedding models:\n - [OpenAI](https://platform.openai.com/docs/guides/embeddings): `text-embedding-3-small`, `text-embedding-3-large`, etc.\n - [Cohere](https://docs.cohere.com/docs/cohere-embed): `embed-english-v3.0`, `embed-multilingual-light-v3.0`, etc.\n\n4. For more details about other parameters, refer to the [LangChain documentation](https://python.langchain.com/docs/integrations/text_embedding/).",
"type": "object",
"editor": "json"
},
"embeddingsApiKey": {
"title": "Embeddings API KEY (whenever applicable, depends on provider)",
"description": "Value of the API KEY for the embeddings provider (if required).\n\n For example for OpenAI it is OPENAI_API_KEY, for Cohere it is COHERE_API_KEY)",
"type": "string",
"editor": "textfield",
"isSecret": true
},
"datasetFields": {
"title": "Dataset fields to select from the dataset results and store in the database",
"type": "array",
"description": "This array specifies the dataset fields to be selected and stored in the vector store. Only the fields listed here will be included in the vector store.\n\nFor instance, when using the Website Content Crawler, you might choose to include fields such as `text`, `url`, and `metadata.title` in the vector store.",
"default": ["text"],
"prefill": ["text"],
"editor": "stringList",
"sectionCaption": "Dataset settings"
},
"metadataDatasetFields": {
"title": "Dataset fields to select from the dataset and store as metadata in the database",
"type": "object",
"description": "A list of dataset fields which should be selected from the dataset and stored as metadata in the vector stores.\n\nFor example, when using the Website Content Crawler, you might want to store `url` in metadata. In this case, use `metadataDatasetFields parameter as follows {\"url\": \"url\"}`",
"editor": "json"
},
"metadataObject": {
"title": "Custom object to be stored as metadata in the vector store database",
"type": "object",
"description": "This object allows you to store custom metadata for every item in the vector store.\n\nFor example, if you want to store the `domain` as metadata, use the `metadataObject` like this: {\"domain\": \"apify.com\"}.",
"editor": "json"
},
"datasetId": {
"title": "Dataset ID",
"type": "string",
"description": "Dataset ID (when running standalone without integration)",
"editor": "textfield"
},
"dataUpdatesStrategy": {
"title": "Update strategy (add, upsert, deltaUpdates (default))",
"description": "Choose the update strategy for the integration. The update strategy determines how the integration updates the data in the database.\n\nThe available options are:\n\n- **Add data** (`add`):\n - Always adds new records to the database.\n - No checks for existing records or updates are performed.\n - Useful when appending data without concern for duplicates.\n\n- **Upsert data** (`upsert`):\n - Updates existing records if they match a key or identifier.\n - Inserts new records into the database if they don't already exist.\n - Ideal for ensuring the database contains the most up-to-date data, avoiding duplicates.\n\n- **Update changed data based on deltas** (`deltaUpdates`):\n - Performs incremental updates by identifying differences (deltas) between the new dataset and the existing records.\n - Only adds new records and updates those that have changed.\n - Unchanged records are left untouched.\n - Maximizes efficiency by reducing unnecessary updates.\n\nSelect the strategy that best fits your use case.",
"type": "string",
"editor": "select",
"enum": ["add", "upsert", "deltaUpdates"],
"default": "deltaUpdates",
"prefill": "deltaUpdates",
"sectionCaption": "Data updates settings"
},
"dataUpdatesPrimaryDatasetFields": {
"title": "Dataset fields to uniquely identify dataset items (only relevant when dataUpdatesStrategy is `upsert` or `deltaUpdates`)",
"type": "array",
"description": "This array contains fields that are used to uniquely identify dataset items, which helps to handle content changes across different runs.\n\nFor instance, in a web content crawling scenario, the `url` field could serve as a unique identifier for each item.",
"editor": "stringList",
"default": [
"url"
],
"prefill": [
"url"
]
},
"enableDeltaUpdates": {
"title": "Enable incremental updates for objects based on deltas (deprecated)",
"type": "boolean",
"description": "When set to true, this setting enables incremental updates for objects in the database by comparing the changes (deltas) between the crawled dataset items and the existing objects, uniquely identified by the `datasetKeysToItemId` field.\n\n The integration will only add new objects and update those that have changed, reducing unnecessary updates. The `datasetFields`, `metadataDatasetFields`, and `metadataObject` fields are used to determine the changes.",
"default": true,
"editor": "hidden"
},
"deltaUpdatesPrimaryDatasetFields": {
"title": "Dataset fields to uniquely identify dataset items (only relevant when `enableDeltaUpdates` is enabled) (deprecated)",
"type": "array",
"description": "This array contains fields that are used to uniquely identify dataset items, which helps to handle content changes across different runs.\n\nFor instance, in a web content crawling scenario, the `url` field could serve as a unique identifier for each item.",
"editor": "hidden",
"default": [
"url"
],
"prefill": [
"url"
]
},
"deleteExpiredObjects": {
"title": "Delete expired objects from the database",
"type": "boolean",
"description": "When set to true, delete objects from the database that have not been crawled for a specified period.",
"default": true
},
"expiredObjectDeletionPeriodDays": {
"title": "Delete expired objects from the database after a specified number of days",
"type": "integer",
"description": "This setting allows the integration to manage the deletion of objects from the database that have not been crawled for a specified period. It is typically used in subsequent runs after the initial crawl.\n\nWhen the value is greater than 0, the integration checks if objects have been seen within the last X days (determined by the expiration period). If the objects are expired, they are deleted from the database. The specific value for `deletedExpiredObjectsDays` depends on your use case and how frequently you crawl data.\n\nFor example, if you crawl data daily, you can set `deletedExpiredObjectsDays` to 7 days. If you crawl data weekly, you can set `deletedExpiredObjectsDays` to 30 days.",
"default": 30,
"minimum": 0,
"unit": "days",
"editor": "number"
},
"performChunking": {
"title": "Enable text chunking",
"description": "When set to true, the text will be divided into smaller chunks based on the settings provided below. Proper chunking helps optimize retrieval and ensures accurate and efficient responses.",
"default": true,
"type": "boolean",
"sectionCaption": "Text chunking settings"
},
"chunkSize": {
"title": "Maximum chunk size",
"type": "integer",
"description": "Defines the maximum number of characters in each text chunk. Choosing the right size balances between detailed context and system performance. Optimal sizes ensure high relevancy and minimal response time.",
"default": 2000,
"minimum": 1
},
"chunkOverlap": {
"title": "Chunk overlap",
"type": "integer",
"description": "Specifies the number of overlapping characters between consecutive text chunks. Adjusting this helps maintain context across chunks, which is crucial for accuracy in retrieval-augmented generation systems.",
"default": 0,
"minimum": 0
}
},
"required": [
"opengaussHost",
"opengaussPort",
"opengaussUser",
"opengaussPassword",
"opengaussDBname",
"opengaussTableName",
"embeddingsProvider",
"embeddingsApiKey",
"datasetFields"
]
}

View File

@ -0,0 +1,248 @@
# openGauss integration
The Apify openGauss integration transfers selected data from Apify Actors to a [openGauss](https://opengauss.org/) database.
It processes the data, optionally splits it into chunks, computes embeddings, and saves them to openGauss.
This integration supports incremental updates, updating only the data that has changed.
This approach reduces unnecessary embedding computation and storage operations, making it suitable for search and retrieval augmented generation (RAG) use cases.
💡 **Note**: This Actor is meant to be used together with other Actors' integration sections.
For instance, if you are using the [Website Content Crawler](https://apify.com/apify/website-content-crawler), you can activate openGauss integration to save web data as vectors to openGauss.
## 📋 How does Apify-openGauss integration work?
Apify openGauss integration computes text embeddings and store them in openGauss.
It uses [LangChain](https://www.langchain.com/) to compute embeddings and interact with [openGauss](https://opengauss.org/).
1. Retrieve a dataset as output from an Actor
2. _[Optional]_ Split text data into chunks using `langchain`'s `RecursiveCharacterTextSplitter`
(enable/disable using `performChunking` and specify `chunkSize`, `chunkOverlap`)
3. _[Optional]_ Update only changed data (select `dataUpdatesStrategy`)
4. Compute embeddings, e.g. using `OpenAI` or `Cohere` (specify `embeddings` and `embeddingsConfig`)
5. Save data into the database
## ✅ Before you start
To utilize this integration, ensure you have:
- Created or existing `openGauss` database. You need to know `opengaussHost`, `opengaussPort`, `opengaussUser`, `opengaussPassword`, `opengaussDBName`and `opengaussTableName`.
- An account to compute embeddings using one of the providers, e.g., [OpenAI](https://platform.openai.com/docs/guides/embeddings) or [Cohere](https://docs.cohere.com/docs/cohere-embed).
## 👉 Examples
The configuration consists of three parts: openGauss, embeddings provider, and data.
Ensure that the vector size of your embeddings aligns with the configuration of your openGauss.
For instance, if you're using the `text-embedding-3-small` model from `OpenAI`, it generates vectors of size `1536`.
This means your openGauss vector should also be configured to accommodate vectors of the same size, `1536` in this case.
For detailed input information refer to the [Input page]().
#### Database: openGauss
```json
{
"opengaussHost": "YOUR-opengaussHost",
"opengaussPort": "YOUR-opengaussPort",
"opengaussUser": "YOUR-opengaussUser",
"opengaussPassword": "YOUR-opengaussPassword",
"opengaussDBname": "YOUR-opengaussDBname",
"opengaussTableName": "apif_collection"
}
```
#### Embeddings provider: OpenAI
```json
{
"embeddingsProvider": "OpenAIEmbeddings",
"embeddingsApiKey": "YOUR-OPENAI-API-KEY",
"embeddingsConfig": {"model": "text-embedding-3-large"}
}
```
### Save data from Website Content Crawler to openGauss
Data is transferred in the form of a dataset from [Website Content Crawler](https://apify.com/apify/website-content-crawler), which provides a dataset with the following output fields (truncated for brevity):
```json
{
"url": "https://www.apify.com",
"text": "Apify is a platform that enables developers to build, run, and share automation tasks.",
"metadata": {"title": "Apify"}
}
```
This dataset is then processed by the openGauss integration.
In the integration settings you need to specify which fields you want to save to openGauss, e.g., `["text"]` and which of them should be used as metadata, e.g., `{"title": "metadata.title"}`.
Without any other configuration, the data is saved to openGauss as is.
```json
{
"datasetFields": ["text"],
"metadataDatasetFields": {"title": "metadata.title"}
}
```
### Create chunks from Website Content Crawler data and save them to the database
Assume that the text data from the [Website Content Crawler](https://apify.com/apify/website-content-crawler) is too long to compute embeddings.
Therefore, we need to divide the data into smaller pieces called chunks.
We can leverage LangChain's `RecursiveCharacterTextSplitter` to split the text into chunks and save them into a database.
The parameters `chunkSize` and `chunkOverlap` are important.
The settings depend on your use case where a proper chunking helps optimize retrieval and ensures accurate responses.
```json
{
"datasetFields": ["text"],
"metadataDatasetFields": {"title": "metadata.title"},
"performChunking": true,
"chunkSize": 1000,
"chunkOverlap": 0
}
```
### Configure update strategy
To control how the integration updates data in the database, use the `dataUpdatesStrategy` parameter. This parameter allows you to choose between different update strategies based on your use case, such as adding new data, upserting records, or incrementally updating records based on changes (deltas). Below are the available strategies and explanations for when to use each:
- **Add data (`add`)**:
- Appends new data to the database without checking for duplicates or updating existing records.
- Suitable for cases where deduplication or updates are unnecessary, and the data simply needs to be added.
- For example, you might use this strategy to continually append data from independent crawls without regard for overlaps.
- **Upsert data (`upsert`)**:
- Delete existing records in the database if they match a key or identifier and inserts new records.
- Ideal when you want to maintain accurate and up-to-date data while avoiding duplication.
- For instance, this is useful in cases where unique items (such as user profiles or documents) need to be managed, ensuring the database reflects the latest changes.
- Check the `dataUpdatesPrimaryDatasetFields` parameter to specify which fields are used to uniquely identify each dataset item.
- **Delta updates (`deltaUpdates`)**:
- Incrementally updates records by identifying differences (deltas) between the new dataset and the existing database records.
- Ensures only new or modified records are processed, leaving unchanged records untouched. This minimizes unnecessary database operations and improves efficiency.
- This is the most efficient strategy when integrating data that evolves over time, such as website content or recurring crawls.
- Check the `dataUpdatesPrimaryDatasetFields` parameter to specify which fields are used to uniquely identify each dataset item.
### Incrementally update database from the Website Content Crawler
To incrementally update data from the [Website Content Crawler](https://apify.com/apify/website-content-crawler) to database, configure the integration to update only the changed or new data.
This is controlled by the `dataUpdatesStrategy` setting.
This way, the integration minimizes unnecessary updates and ensures that only new or modified data is processed.
A checksum is computed for each dataset item (together with all metadata) and stored in the database alongside the vectors.
When the data is re-crawled, the checksum is recomputed and compared with the stored checksum.
If the checksum is different, the old data (including vectors) is deleted and new data is saved.
Otherwise, only the `last_seen_at` metadata field is updated to indicate when the data was last seen.
#### Provide unique identifier for each dataset item
To incrementally update the data, you need to be able to uniquely identify each dataset item.
The variable `dataUpdatesPrimaryDatasetFields` specifies which fields are used to uniquely identify each dataset item and helps track content changes across different crawls.
For instance, when working with the Website Content Crawler, you can use the URL as a unique identifier.
```json
{
"dataUpdatesStrategy": "deltaUpdates",
"dataUpdatePrimaryDatasetFields": ["url"]
}
```
To fully maximize the potential of incremental data updates, it is recommended to start with an empty database.
While it is possible to use this feature with an existing database, records that were not originally saved using a prefix or metadata will not be updated.
### Delete outdated (expired) data
The integration can delete data from the database that hasn't been crawled for a specified period, which is useful when data becomes outdated, such as when a page is removed from a website.
The deletion feature can be enabled or disabled using the `deleteExpiredObjects` setting.
For each crawl, the `last_seen_at` metadata field is created or updated.
This field records the most recent time the data object was crawled.
The `expiredObjectDeletionPeriodDays` setting is used to control number of days since the last crawl, after which the data object is considered expired.
If a database object has not been seen for more than the `expiredObjectDeletionPeriodDays`, it will be deleted automatically.
The specific value of `expiredObjectDeletionPeriodDays` depends on your use case.
- If a website is crawled daily, `expiredObjectDeletionPeriodDays` can be set to 7.
- If you crawl weekly, it can be set to 30.
To disable this feature, set `deleteExpiredObjects` to `false`.
```json
{
"deleteExpiredObjects": true,
"expiredObjectDeletionPeriodDays": 30
}
```
💡 If you are using multiple Actors to update the same database, ensure that all Actors crawl the data at the same frequency.
Otherwise, data crawled by one Actor might expire due to inconsistent crawling schedules.
## 💾 Outputs
This integration will save the selected fields from your Actor to openGauss.
## 🔢 Example configuration
#### Full Input Example for Website Content Crawler Actor with openGauss integration
```json
{
"opengaussHost": "YOUR-opengaussHost",
"opengaussPort": "YOUR-opengaussPort",
"opengaussUser": "YOUR-opengaussUser",
"opengaussPassword": "YOUR-opengaussPassword",
"opengaussDBname": "YOUR-opengaussDBname",
"opengaussTableName": "apif_collection",
"embeddingsApiKey": "YOUR-OPENAI-API-KEY",
"embeddingsConfig": {
"model": "text-embedding-3-small"
},
"embeddingsProvider": "OpenAI",
"datasetFields": [
"text"
],
"dataUpdatesStrategy": "deltaUpdates",
"dataUpdatePrimaryDatasetFields": ["url"],
"expiredObjectDeletionPeriodDays": 7,
"performChunking": true,
"chunkSize": 2000,
"chunkOverlap": 200
}
```
#### openGauss
```json
{
"opengaussHost": "YOUR-opengaussHost",
"opengaussPort": "YOUR-opengaussPort",
"opengaussUser": "YOUR-opengaussUser",
"opengaussPassword": "YOUR-opengaussPassword",
"opengaussDBname": "YOUR-opengaussDBname",
"opengaussTableName": "apif_collection"
}
```
#### OpenAI embeddings
```json
{
"embeddingsApiKey": "YOUR-OPENAI-API-KEY",
"embeddings": "OpenAI",
"embeddingsConfig": {"model": "text-embedding-3-large"}
}
```
#### Cohere embeddings
```json
{
"embeddingsApiKey": "YOUR-COHERE-API-KEY",
"embeddings": "Cohere",
"embeddingsConfig": {"model": "embed-multilingual-v3.0"}
}
```

View File

@ -0,0 +1,18 @@
# IDE configurations
.idea
.vscode
# crawlee and apify storage folders
apify_storage
crawlee_storage
storage
# python related
__pycache__
.mypy_cache
.pytest_cache
.coverage
.venv
# git folder
.git

View File

@ -0,0 +1,15 @@
root = true
[*]
indent_style = space
indent_size = 4
charset = utf-8
trim_trailing_whitespace = true
insert_final_newline = true
end_of_line = lf
[Makefile]
indent_style = tab
[{*.yaml, *.yml, *.toml, *.ini, *.cfg}]
indent_size = 2

View File

@ -0,0 +1,9 @@
APIFY_API_TOKEN=
OPENAI_API_KEY=
# openGauss
OPENGAUSS_HOST=
OPENGAUSS_PORT=
OPENGAUSS_USER=
OPENGAUSS_PASSWORD=
OPENGAUSS_DBNAME=

View File

@ -0,0 +1,32 @@
# This file tells Git which files shouldn't be added to source control
.idea
.DS_Store
.vscode
apify_storage
storage
.venv/
.env/
__pypackages__
dist/
build/
*.egg-info/
*.egg
__pycache__
.mypy_cache
.dmypy.json
dmypy.json
.pytest_cache
.ruff_cache
.scrapy
*.log
# Added by Apify CLI
node_modules
.venv
.env

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,4 @@
cache-dir = "/tmp/.poetry-cache"
[virtualenvs]
in-project = true

View File

@ -0,0 +1,129 @@
[tool.poetry]
# These fields are not used by Apify, fill configs in .actor/ instead
authors = ["jiri.spilka@apify.com"]
description = ""
name = "store-vector-db"
readme = "README.md"
version = "0.1.5"
package-mode = false
[tool.poetry.dependencies]
apify = "^2.7.3"
apify-client = "^1.12.2"
backoff = "^2.2.1"
langchain-cohere = "^0.3.0"
langchain-community = "^0.3.0"
langchain-core = "0.3.70"
langchain-openai = "^0.2.0"
openai = "^1.17.0"
python = ">=3.11,<3.12"
python-dotenv = "^1.0.1"
langchain-apify = "^0.1.4"
[tool.poetry.group.dev.dependencies]
coverage = "^7.5.4"
datamodel-code-generator = "^0.25.5"
ipython = "^8.23.0"
mypy = "^1.9.0"
pandas = "^2.2.2"
pre-commit = "^3.7.0"
pytest = "^8.2.0"
pytest-asyncio = "^0.23.6"
pytest-integration-mark = "^0.2.0"
ruff = "^0.3.5"
[tool.poetry.group.opengauss]
optional = true
[tool.poetry.group.opengauss.dependencies]
langchain-opengauss = "^0.1.4"
[tool.ruff]
line-length = 150
exclude = ["src/models/**", "src/examples/**"]
[tool.ruff.lint]
select = ["ALL"]
ignore = [
"ANN101", # Missing type annotation for `{name}` in method
"ANN102", # Missing type annotation for `{name}` in classmethod
"ANN401", # Dynamically typed expressions (typing.Any) are disallowed in {filename}
"BLE001", # Do not catch blind exception
"C901", # `{name}` is too complex
"COM812", # This rule may cause conflicts when used with the formatter
"D100", # Missing docstring in public module
"D104", # Missing docstring in public package
"D107", # Missing docstring in `__init__`
"EM", # flake8-errmsg
"G004", # Logging statement uses f-string
"ISC001", # This rule may cause conflicts when used with the formatter
"FIX", # flake8-fixme
"PGH003", # Use specific rule codes when ignoring type issues
"PLR0911", # Too many return statements
"PLR0913", # Too many arguments in function definition
"PLR0915", # Too many statements
"PTH", # flake8-use-pathlib
"PYI034", # `__aenter__` methods in classes like `{name}` usually return `self` at runtime
"PYI036", # The second argument in `__aexit__` should be annotated with `object` or `BaseException | None`
"S102", # Use of `exec` detected
"S105", # Possible hardcoded password assigned to
"S106", # Possible hardcoded password assigned to argument: "{name}"
"S301", # `pickle` and modules that wrap it can be unsafe when used to deserialize untrusted data, possible security issue
"S303", # Use of insecure MD2, MD4, MD5, or SHA1 hash function
"S311", # Standard pseudo-random generators are not suitable for cryptographic purposes
"TD002", # Missing author in TODO; try: `# TODO(<author_name>): ...` or `# TODO @<author_name>: ...
"TID252", # Prefer absolute imports over relative imports from parent modules
"TRY003", # Avoid specifying long messages outside the exception class
#
"D",
]
[tool.ruff.format]
quote-style = "double"
indent-style = "space"
[tool.ruff.lint.per-file-ignores]
"**/__init__.py" = [
"F401", # Unused imports
]
"**/{tests}/*" = [
"D", # Everything from the pydocstyle
"INP001", # File {filename} is part of an implicit namespace package, add an __init__.py
"PLR2004", # Magic value used in comparison, consider replacing {value} with a constant variable
"S101", # Use of assert detected
"SLF001", # Private member accessed: `{name}`
"T20", # flake8-print
"TRY301", # Abstract `raise` to an inner function
]
[tool.ruff.lint.flake8-quotes]
docstring-quotes = "double"
inline-quotes = "double"
[tool.ruff.lint.pydocstyle]
convention = "google"
#[tool.ruff.lint.isort]
#known-first-party = ["apify"]
[tool.pytest.ini_options]
addopts = "-ra"
asyncio_mode = "auto"
timeout = 1200
[tool.mypy]
python_version = "3.11"
files = ["src", "tests"]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_redundant_casts = true
warn_return_any = true
warn_unreachable = true
warn_unused_ignores = true
[tool.mypy-sortedcollections]
ignore_missing_imports = true

View File

@ -0,0 +1,19 @@
import asyncio
import logging
from apify.log import ActorLogFormatter
from .entrypoint import main
handler = logging.StreamHandler()
handler.setFormatter(ActorLogFormatter())
apify_client_logger = logging.getLogger("apify_client")
apify_client_logger.setLevel(logging.INFO)
apify_client_logger.addHandler(handler)
apify_logger = logging.getLogger("apify")
apify_logger.setLevel(logging.DEBUG)
apify_logger.addHandler(handler)
asyncio.run(main())

View File

@ -0,0 +1,10 @@
from __future__ import annotations
from typing import TYPE_CHECKING, TypeAlias
if TYPE_CHECKING:
from .models import OpengaussIntegration
from .vector_stores import OpenGaussDatabase
ActorInputsDb: TypeAlias = OpengaussIntegration
VectorDb: TypeAlias = OpenGaussDatabase

View File

@ -0,0 +1,15 @@
import enum
VCR_HEADERS_EXCLUDE = ["Authorization", "Api-Key"]
DAY_IN_SECONDS = 24 * 3600
class SupportedVectorStores(str, enum.Enum):
opengauss = "opengauss"
class SupportedEmbeddings(str, enum.Enum):
openai = "OpenAI"
cohere = "Cohere"
fake = "Fake"

View File

@ -0,0 +1,37 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from apify import Actor
from .constants import SupportedEmbeddings
if TYPE_CHECKING:
from langchain_core.embeddings import Embeddings
async def get_embedding_provider(embeddings_name: str, api_key: str | None = None, config: dict | None = None) -> Embeddings:
"""Return the embeddings based on the user preference."""
if embeddings_name == SupportedEmbeddings.openai:
from langchain_openai.embeddings import OpenAIEmbeddings
config = config or {}
config["openai_api_key"] = api_key
return config and OpenAIEmbeddings(**config) or OpenAIEmbeddings()
if embeddings_name == SupportedEmbeddings.cohere:
from langchain_cohere import CohereEmbeddings
config = config or {}
config["cohere_api_key"] = api_key
return CohereEmbeddings(**config)
if embeddings_name == SupportedEmbeddings.fake:
from langchain_core.embeddings import FakeEmbeddings
config = config or {}
return FakeEmbeddings(**config)
await Actor.fail(status_message=f"Failed to get embeddings for embeddings: {embeddings_name} and config: {config}")
raise ValueError("Failed to get embeddings")

View File

@ -0,0 +1,64 @@
import os
from apify import Actor
from .constants import SupportedVectorStores
from .main import run_actor
from .models import OpengaussIntegration
async def main() -> None:
async with Actor:
Actor.log.info("Starting the Vector Store Actor")
if not (actor_input := await Actor.get_input() or {}):
await Actor.fail(status_message="No input provided", exit_code=1)
# Set the Apify API token if it is available in the environment variables.
if apify_api_token := os.getenv("APIFY_TOKEN"):
os.environ["APIFY_API_TOKEN"] = apify_api_token
if not (arg := os.getenv("ACTOR_PATH_IN_DOCKER_CONTEXT")):
if Actor.is_at_home():
await Actor.exit(
exit_code=100,
status_message="This Actor was built incorrectly; no environment variable specifies which Actor "
"to start. If you encounter this issue, please contact the Actor developer.",
)
arg = f"actors/{SupportedVectorStores.opengauss.value}"
Actor.log.warning(
f"The environment variable ACTOR_PATH_IN_DOCKER_CONTEXT was not specified. " f"Using default for local development: {arg}"
)
actor_type = arg.split("/")[-1]
Actor.log.info("Received start argument (vector database name): %s", actor_type)
actor_input_ensure_backward_compatibility(actor_input)
if actor_type == SupportedVectorStores.opengauss.value:
await run_actor(OpengaussIntegration(**actor_input), actor_input)
else:
await Actor.exit(
exit_code=10,
status_message=f"This Actor was built incorrectly; an unknown Actor was selected "
f"to start ({actor_type}). If you encounter this issue, please contact the Actor developer.",
)
def actor_input_ensure_backward_compatibility(actor_input: dict) -> None:
"""Ensure backward compatibility for the actor input."""
if not actor_input.get("dataUpdatesStrategy"):
# legacy update mechanism
if actor_input.get("enableDeltaUpdates") is False:
actor_input["dataUpdatesStrategy"] = "add"
else:
actor_input["dataUpdatesStrategy"] = "deltaUpdates"
else:
# for integrations that do not have updateStrategy implemented
actor_input["enableDeltaUpdates"] = actor_input["dataUpdatesStrategy"] == "deltaUpdates"
if not actor_input.get("dataUpdatesPrimaryDatasetFields"):
actor_input["dataUpdatesPrimaryDatasetFields"] = actor_input.get("deltaUpdatesPrimaryDatasetFields", [])
else:
# for integrations that do not have updateStrategy implemented
actor_input["deltaUpdatesPrimaryDatasetFields"] = actor_input.get("dataUpdatesPrimaryDatasetFields")

View File

@ -0,0 +1,129 @@
# type: ignore
import os
import time
from datetime import datetime, timezone
from dotenv import load_dotenv
from langchain_openai.embeddings import OpenAIEmbeddings
from ..models import OpengaussIntegration
from .data_examples_uuid import (
ID1, ID3, ID4A, ID4B, ID4C, ID5A, ID5B, ID5C, ID6,
crawl_1, crawl_2, expected_results,
)
from ..vcs import compare_crawled_data_with_db
from ..vector_stores.opengauss import OpenGaussDatabase
load_dotenv()
OPENGAUSS_TABLE_NAME = os.getenv("OPENGAUSS_TABLE_NAME", "apify")
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
DROP_AND_INSERT = True
db = OpenGaussDatabase(
actor_input=OpengaussIntegration(
opengaussHost=os.getenv("OPENGAUSS_HOST"),
opengaussPort=os.getenv("OPENGAUSS_PORT"),
opengaussUser=os.getenv("OPENGAUSS_USER"),
opengaussPassword=os.getenv("OPENGAUSS_PASSWORD"),
opengaussDBname=os.getenv("OPENGAUSS_DBNAME"),
opengaussTableName=OPENGAUSS_TABLE_NAME,
embeddingsProvider="OpenAI",
embeddingsApiKey=os.getenv("OPENAI_API_KEY"),
datasetFields=["text"],
),
embeddings=embeddings,
)
def wait_for_index(sec: float = 1.0):
time.sleep(sec)
if DROP_AND_INSERT:
db.delete_all()
r = db.similarity_search("text", k=100)
print("Initial results count:", len(r))
inserted = db.add_documents(documents=crawl_1, ids=[d.metadata["chunk_id"] for d in crawl_1])
print("Inserted ids:", inserted)
wait_for_index()
r = db.similarity_search("text", k=100)
print("Search results:", r)
print("Search results count:", len(r))
res = db.search_by_vector(db.dummy_vector, k=10)
print("Objects in the database:", len(res), res)
assert len(res) == 6, "Expected 6 objects in the database"
data_add, ids_update_last_seen, ids_del = compare_crawled_data_with_db(db, crawl_2)
print("Data to add", data_add)
print("Ids to update", ids_update_last_seen)
print("Ids to delete", ids_del)
assert len(data_add) == 4, "Expected 4 objects to add"
assert data_add[0].metadata["chunk_id"] == ID4C
assert data_add[1].metadata["chunk_id"] == ID5B
assert data_add[2].metadata["chunk_id"] == ID5C
assert data_add[3].metadata["chunk_id"] == ID6
assert len(ids_update_last_seen) == 1, "Expected 1 object to update"
assert ID3 in ids_update_last_seen, f"Expected {ID3} to be updated"
assert len(ids_del) == 3, "Expected 3 objects to delete"
assert ID4A in ids_del, f"Expected {ID4A} to be deleted"
assert ID4B in ids_del, f"Expected {ID4B} to be deleted"
assert ID5A in ids_del, f"Expected {ID5A} to be deleted"
# Delete data that were removed
db.delete(ids_del)
wait_for_index()
res = db.search_by_vector(db.dummy_vector, k=10)
print("Database objects after delete: ", len(res), res)
assert len(res) == 3, "Expected 3 objects in the database after deletion"
# Add new data
r = db.add_documents(data_add, ids=[d.metadata["chunk_id"] for d in data_add])
wait_for_index()
res = db.search_by_vector(db.dummy_vector, k=10)
print("Database objects after adding new", len(res), res)
ids = [r.metadata["chunk_id"] for r in res]
assert len(res) == 7, "Expected 7 objects in the database after addition"
assert ID4C in ids and ID5B in ids and ID5C in ids, "Expected new chunk_ids to be present"
# Update metadata (last_seen_at)
ts = int(datetime.now(timezone.utc).timestamp())
res = db.search_by_vector(db.dummy_vector, k=10)
# precondition in examples: ID3 initially has last_seen_at == 1
assert next(r for r in res if r.metadata["chunk_id"] == ID3).metadata["last_seen_at"] == 1
db.update_last_seen_at(ids_update_last_seen)
wait_for_index()
res = db.search_by_vector(db.dummy_vector, k=10)
assert len(res) == 7, "Expected 7 objects after metadata update"
assert next(r for r in res if r.metadata["chunk_id"] == ID3).metadata["last_seen_at"] >= ts, f"Expected {ID3} to be updated"
# delete expired objects
db.delete_expired(expired_ts=1)
wait_for_index()
res = db.search_by_vector(db.dummy_vector, k=10)
res = [r for r in res]
print("Database objects after all updates", len(res), res)
assert len(res) == 6, "Expected 6 objects after all updates"
assert next((r for r in res if r.metadata["chunk_id"] == ID1), None) is None, f"Expected {ID1} to be deleted"
# compare results with expected results
for r in expected_results:
d = db.get_by_id(r.metadata["chunk_id"])
assert d is not None, f"Expected document {r.metadata['chunk_id']} to exist"
metadata = d.metadata
assert metadata["item_id"] == r.metadata["item_id"], f"Expected item_id {r.metadata['item_id']}"
assert metadata["checksum"] == r.metadata["checksum"], f"Expected checksum {r.metadata['checksum']}"
print("DONE")

View File

@ -0,0 +1,19 @@
"""
Define crawled data for the database playground files
"""
from langchain_core.documents import Document
d1 = Document(page_content="Expired->del", metadata={"item_id": "id1", "id": "id1#1", "checksum": "1", "last_seen_at": 0})
d2 = Document(page_content="Old->not-del", metadata={"item_id": "id2", "id": "id2#2", "checksum": "2", "last_seen_at": 1})
d3a = Document(page_content="Unchanged->upt-meta", metadata={"item_id": "id3", "id": "id3#3", "checksum": "3", "last_seen_at": 1})
d3b = Document(page_content="Unchanged->upt-meta", metadata={"item_id": "id3", "id": "id3#3", "checksum": "3", "last_seen_at": 2})
d4a = Document(page_content="Changed->del", metadata={"item_id": "id4", "id": "id4#4a", "checksum": "4", "last_seen_at": 1})
d4b = Document(page_content="Changed->del", metadata={"item_id": "id4", "id": "id4#4b", "checksum": "4", "last_seen_at": 1})
d4c = Document(page_content="Changed->add-new", metadata={"item_id": "id4", "id": "id4#4c", "checksum": "0", "last_seen_at": 2})
d5 = Document(page_content="New->add", metadata={"item_id": "id5", "id": "id5#5", "checksum": "5", "last_seen_at": 2})
crawl_1 = [d1, d2, d3a, d4a, d4b]
crawl_2 = [d3b, d4c, d5]
expected_results = [d2, d3b, d4c, d5]

View File

@ -0,0 +1,33 @@
"""
Define crawled data for the database playground files
"""
from langchain_core.documents import Document
UUID = "00000000-0000-0000-0000-0000000000"
ID1 = f"{UUID}10"
ID2 = f"{UUID}20"
ID3 = f"{UUID}30"
ID4A, ID4B, ID4C = f"{UUID}4a", f"{UUID}4b", f"{UUID}4c"
ID5A, ID5B, ID5C = f"{UUID}5a", f"{UUID}5b", f"{UUID}5c"
ID6 = f"{UUID}60"
ITEM_ID1 = "id1"
ITEM_ID4 = "id4"
d1 = Document(page_content="Expired->del", metadata={"item_id": ITEM_ID1, "chunk_id": ID1, "checksum": "1", "last_seen_at": 0})
d2 = Document(page_content="Old->not-del", metadata={"item_id": "id2", "chunk_id": ID2, "checksum": "2", "last_seen_at": 1})
d3a = Document(page_content="Unchanged->upt-meta", metadata={"item_id": "id3", "chunk_id": ID3, "checksum": "3", "last_seen_at": 1})
d3b = Document(page_content="Unchanged->upt-meta", metadata={"item_id": "id3", "chunk_id": ID3, "checksum": "3", "last_seen_at": 2})
d4a = Document(page_content="Changed->del", metadata={"item_id": ITEM_ID4, "chunk_id": ID4A, "checksum": "4", "last_seen_at": 1})
d4b = Document(page_content="Changed->del", metadata={"item_id": ITEM_ID4, "chunk_id": ID4B, "checksum": "4", "last_seen_at": 1})
d4c = Document(page_content="Changed->add-new", metadata={"item_id": ITEM_ID4, "chunk_id": ID4C, "checksum": "4c", "last_seen_at": 2})
d5a = Document(page_content="Changed->del", metadata={"item_id": "id5", "chunk_id": ID5A, "checksum": "5", "last_seen_at": 1})
d5b = Document(page_content="Changed->add-new", metadata={"item_id": "id5", "chunk_id": ID5B, "checksum": "5bc", "last_seen_at": 2})
d5c = Document(page_content="Changed->add-new", metadata={"item_id": "id5", "chunk_id": ID5C, "checksum": "5bc", "last_seen_at": 2})
d6 = Document(page_content="New->add", metadata={"item_id": "id5", "chunk_id": ID6, "checksum": "6", "last_seen_at": 2})
crawl_1 = [d1, d2, d3a, d4a, d4b, d5a]
crawl_2 = [d3b, d4c, d5b, d5c, d6]
expected_results = [d2, d3b, d4c, d5b, d5c, d6]

View File

@ -0,0 +1,2 @@
class FailedToConnectToDatabaseError(Exception):
"""Failed to connect to a vector database."""

View File

@ -0,0 +1,154 @@
from __future__ import annotations
from datetime import datetime, timezone
from typing import TYPE_CHECKING
from apify import Actor
from langchain_text_splitters import RecursiveCharacterTextSplitter
from .constants import DAY_IN_SECONDS
from .emb import get_embedding_provider
from .utils import add_chunk_id, add_item_checksum, get_dataset_loader
from .vcs import delete_expired_objects, get_vector_database, update_db_with_crawled_data, upsert_db_with_crawled_data
if TYPE_CHECKING:
from langchain_core.documents import Document
from langchain_core.embeddings import Embeddings
from ._types import ActorInputsDb, VectorDb
async def run_actor(actor_input: ActorInputsDb, payload: dict) -> None:
"""Main function to run the actor.
It loads the dataset, chunks the documents if necessary and updates the vector store with the new documents while removing the old ones.
"""
payload = payload.get("payload", {})
resource = payload.get("resource", {})
if not (dataset_id := resource.get("defaultDatasetId") or actor_input.datasetId):
msg = (
"The `datasetId` is not provided. There are two ways to specify the datasetId:"
"1. Automatic Input: If this integration is used with other Actors, such as the Website Content Crawler, the datasetId should be "
"automatically passed in the 'payload'. Please check the `Input` payload to ensure the datasetId is included."
"2. Manual Input: If you are running this Actor independently, you need to manually specify the 'datasetId'. "
"You can do this by entering the dataset ID in the 'Dataset Settings' section of the Actor's input screen."
"Please verify that one of these options is correctly configured to provide the datasetId."
)
Actor.log.error(msg)
await Actor.fail(status_message=msg)
return
embeddings = await get_embeddings(actor_input)
documents = await load_dataset(actor_input, dataset_id)
documents = add_item_checksum(documents, actor_input.dataUpdatesPrimaryDatasetFields) # type: ignore[arg-type]
if actor_input.performChunking:
text_splitter = RecursiveCharacterTextSplitter(chunk_size=actor_input.chunkSize, chunk_overlap=actor_input.chunkOverlap)
documents = text_splitter.split_documents(documents)
Actor.log.info("Documents chunked to %s chunks", len(documents))
documents = add_chunk_id(documents)
try:
vcs_: VectorDb = await get_vector_database(actor_input, embeddings)
except Exception as e:
Actor.log.exception(e)
await Actor.fail(
status_message="Failed to connect/get database. Please ensure the following: "
"1. Database credentials are correct and the database is configure properly. "
"2. The vector dimension of your embedding model in the Actor input (Embedding settings -> model) matches the one set up in the database."
f" Database error message: {e}"
)
return
try:
data_update_strategy = hasattr(actor_input, "dataUpdatesStrategy") and actor_input.dataUpdatesStrategy
if data_update_strategy == "deltaUpdates":
Actor.log.info("Update database with crawled data. Delta updates enabled")
update_db_with_crawled_data(vcs_, documents)
elif data_update_strategy == "add":
vcs_.add_documents(documents)
Actor.log.info("Added %s new objects to the vector store", len(documents))
elif data_update_strategy == "upsert":
upsert_db_with_crawled_data(vcs_, documents)
else:
await Actor.fail(
status_message=f"Invalid dataUpdatesStrategy: {data_update_strategy}. "
f"Please ensure that the configuration in the Database Settings is correct."
)
if actor_input.deleteExpiredObjects:
expired_days = actor_input.expiredObjectDeletionPeriodDays or 0
ts_expired = expired_days and int(datetime.now(timezone.utc).timestamp() - expired_days * DAY_IN_SECONDS) or 0
Actor.log.info("Delete expired objects in the database: expired_days: %s", expired_days)
delete_expired_objects(vcs_, ts_expired)
await Actor.push_data([doc.dict() for doc in documents])
if hasattr(vcs_, "close"):
vcs_.close()
except Exception as e:
Actor.log.error(e)
# I had to create a msg variable to avoid a ruff lint error S608 (SQL Injection)
msg = (
"Failed to update database. Please ensure the following:"
"1. Database is configured properly."
"2. The vector dimension of your embedding model in the Actor input (Embedding settings -> model) matches the one set up in the database."
"Error message:"
)
await Actor.fail(status_message=f"{msg} {e}", exception=e)
async def get_embeddings(actor_input: ActorInputsDb) -> Embeddings: # type: ignore[return]
try:
embed_provider_name = str(actor_input.embeddingsProvider)
Actor.log.info("Get embeddings class: %s", embed_provider_name)
embeddings = await get_embedding_provider(
embed_provider_name,
actor_input.embeddingsApiKey,
actor_input.embeddingsConfig,
)
except Exception as e:
Actor.log.error(e)
await Actor.fail(status_message=f"Failed to get embeddings: {e}. Ensure that the configuration in the Embeddings Settings is correct.")
else:
return embeddings
async def load_dataset(actor_input: ActorInputsDb, dataset_id: str) -> list[Document]: # type: ignore[return]
"""Load dataset from the datasetId and extract fields from the dataset."""
# Add parameters related to chunking to every dataset item to be able to update DB when chunkSize, chunkOverlap or performChunking changes
meta_object = actor_input.metadataObject or {}
meta_object.update({"chunkSize": actor_input.chunkSize, "chunkOverlap": actor_input.chunkOverlap, "performChunking": actor_input.performChunking})
# Required for checksum calculation
# Update metadata fields with datasetFieldsToItemId for dataset loading
meta_fields = actor_input.metadataDatasetFields or {}
meta_fields.update({k: k for k in actor_input.dataUpdatesPrimaryDatasetFields or []})
Actor.log.info("Load Dataset ID %s and extract fields %s", dataset_id, actor_input.datasetFields)
try:
dataset_loader = get_dataset_loader(
str(dataset_id),
fields=actor_input.datasetFields,
meta_object=meta_object,
meta_fields=meta_fields,
)
documents = dataset_loader.load()
documents = [doc for doc in documents if doc.page_content]
Actor.log.info("Dataset loaded, number of documents: %s", len(documents))
except Exception as e:
Actor.log.error(e)
await Actor.fail(
status_message=f"Failed to load datasetId {dataset_id} due to error: {e}. Ensure the following: "
f"1. If running this Actor standalone, the dataset should exist. "
f"2. If this Actor is configured with another Actor (in the integration section), the `datasetId` should be correctly passed. "
f"3. If the problem persists, consider creating an issue."
)
else:
return documents

View File

@ -0,0 +1,2 @@
# __init__.py
from .opengauss_input_model import OpengaussIntegration

View File

@ -0,0 +1,116 @@
# generated by datamodel-codegen:
# filename: input_schema.json
# timestamp: 2025-09-19T03:51:44+00:00
from __future__ import annotations
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
from typing_extensions import Literal
class OpengaussIntegration(BaseModel):
opengaussHost: str = Field(
..., description='The Host of openGauss', title='openGauss Host'
)
opengaussPort: str = Field(
..., description='The Port of openGauss', title='openGauss Port'
)
opengaussUser: str = Field(
..., description='The User of openGauss', title='openGauss User'
)
opengaussPassword: str = Field(
..., description='The Password of openGauss', title='openGauss Password'
)
opengaussDBname: str = Field(
..., description='The DBname of openGauss', title='openGauss DBname'
)
opengaussTableName: str = Field(
...,
description='The name of the table to use',
title='openGauss SQL table name',
)
embeddingsProvider: Literal['OpenAI', 'Cohere'] = Field(
...,
description='Choose the embeddings provider to use for generating embeddings',
title='Embeddings provider (as defined in the langchain API)',
)
embeddingsConfig: Optional[Dict[str, Any]] = Field(
None,
description='Configure the parameters for the LangChain embedding class. Key points to consider:\n\n1. Typically, you only need to specify the model name. For example, for OpenAI, set the model name as {"model": "text-embedding-3-small"}.\n\n2. It\'s required to ensure that the vector size of your embeddings matches the size of embeddings in the database.\n\n3. Here are examples of embedding models:\n - [OpenAI](https://platform.openai.com/docs/guides/embeddings): `text-embedding-3-small`, `text-embedding-3-large`, etc.\n - [Cohere](https://docs.cohere.com/docs/cohere-embed): `embed-english-v3.0`, `embed-multilingual-light-v3.0`, etc.\n\n4. For more details about other parameters, refer to the [LangChain documentation](https://python.langchain.com/docs/integrations/text_embedding/).',
title='Configuration for embeddings provider',
)
embeddingsApiKey: str = Field(
...,
description='Value of the API KEY for the embeddings provider (if required).\n\n For example for OpenAI it is OPENAI_API_KEY, for Cohere it is COHERE_API_KEY)',
title='Embeddings API KEY (whenever applicable, depends on provider)',
)
datasetFields: List = Field(
...,
description='This array specifies the dataset fields to be selected and stored in the vector store. Only the fields listed here will be included in the vector store.\n\nFor instance, when using the Website Content Crawler, you might choose to include fields such as `text`, `url`, and `metadata.title` in the vector store.',
title='Dataset fields to select from the dataset results and store in the database',
)
metadataDatasetFields: Optional[Dict[str, Any]] = Field(
None,
description='A list of dataset fields which should be selected from the dataset and stored as metadata in the vector stores.\n\nFor example, when using the Website Content Crawler, you might want to store `url` in metadata. In this case, use `metadataDatasetFields parameter as follows {"url": "url"}`',
title='Dataset fields to select from the dataset and store as metadata in the database',
)
metadataObject: Optional[Dict[str, Any]] = Field(
None,
description='This object allows you to store custom metadata for every item in the vector store.\n\nFor example, if you want to store the `domain` as metadata, use the `metadataObject` like this: {"domain": "apify.com"}.',
title='Custom object to be stored as metadata in the vector store database',
)
datasetId: Optional[str] = Field(
None,
description='Dataset ID (when running standalone without integration)',
title='Dataset ID',
)
dataUpdatesStrategy: Optional[Literal['add', 'upsert', 'deltaUpdates']] = Field(
'deltaUpdates',
description="Choose the update strategy for the integration. The update strategy determines how the integration updates the data in the database.\n\nThe available options are:\n\n- **Add data** (`add`):\n - Always adds new records to the database.\n - No checks for existing records or updates are performed.\n - Useful when appending data without concern for duplicates.\n\n- **Upsert data** (`upsert`):\n - Updates existing records if they match a key or identifier.\n - Inserts new records into the database if they don't already exist.\n - Ideal for ensuring the database contains the most up-to-date data, avoiding duplicates.\n\n- **Update changed data based on deltas** (`deltaUpdates`):\n - Performs incremental updates by identifying differences (deltas) between the new dataset and the existing records.\n - Only adds new records and updates those that have changed.\n - Unchanged records are left untouched.\n - Maximizes efficiency by reducing unnecessary updates.\n\nSelect the strategy that best fits your use case.",
title='Update strategy (add, upsert, deltaUpdates (default))',
)
dataUpdatesPrimaryDatasetFields: Optional[List] = Field(
['url'],
description='This array contains fields that are used to uniquely identify dataset items, which helps to handle content changes across different runs.\n\nFor instance, in a web content crawling scenario, the `url` field could serve as a unique identifier for each item.',
title='Dataset fields to uniquely identify dataset items (only relevant when dataUpdatesStrategy is `upsert` or `deltaUpdates`)',
)
enableDeltaUpdates: Optional[bool] = Field(
True,
description='When set to true, this setting enables incremental updates for objects in the database by comparing the changes (deltas) between the crawled dataset items and the existing objects, uniquely identified by the `datasetKeysToItemId` field.\n\n The integration will only add new objects and update those that have changed, reducing unnecessary updates. The `datasetFields`, `metadataDatasetFields`, and `metadataObject` fields are used to determine the changes.',
title='Enable incremental updates for objects based on deltas (deprecated)',
)
deltaUpdatesPrimaryDatasetFields: Optional[List] = Field(
['url'],
description='This array contains fields that are used to uniquely identify dataset items, which helps to handle content changes across different runs.\n\nFor instance, in a web content crawling scenario, the `url` field could serve as a unique identifier for each item.',
title='Dataset fields to uniquely identify dataset items (only relevant when `enableDeltaUpdates` is enabled) (deprecated)',
)
deleteExpiredObjects: Optional[bool] = Field(
True,
description='When set to true, delete objects from the database that have not been crawled for a specified period.',
title='Delete expired objects from the database',
)
expiredObjectDeletionPeriodDays: Optional[int] = Field(
30,
description='This setting allows the integration to manage the deletion of objects from the database that have not been crawled for a specified period. It is typically used in subsequent runs after the initial crawl.\n\nWhen the value is greater than 0, the integration checks if objects have been seen within the last X days (determined by the expiration period). If the objects are expired, they are deleted from the database. The specific value for `deletedExpiredObjectsDays` depends on your use case and how frequently you crawl data.\n\nFor example, if you crawl data daily, you can set `deletedExpiredObjectsDays` to 7 days. If you crawl data weekly, you can set `deletedExpiredObjectsDays` to 30 days.',
ge=0,
title='Delete expired objects from the database after a specified number of days',
)
performChunking: Optional[bool] = Field(
True,
description='When set to true, the text will be divided into smaller chunks based on the settings provided below. Proper chunking helps optimize retrieval and ensures accurate and efficient responses.',
title='Enable text chunking',
)
chunkSize: Optional[int] = Field(
2000,
description='Defines the maximum number of characters in each text chunk. Choosing the right size balances between detailed context and system performance. Optimal sizes ensure high relevancy and minimal response time.',
ge=1,
title='Maximum chunk size',
)
chunkOverlap: Optional[int] = Field(
0,
description='Specifies the number of overlapping characters between consecutive text chunks. Adjusting this helps maintain context across chunks, which is crucial for accuracy in retrieval-augmented generation systems.',
ge=0,
title='Chunk overlap',
)

View File

@ -0,0 +1,159 @@
from __future__ import annotations
import copy
import hashlib
import logging
from collections import defaultdict
from datetime import datetime, timezone
from typing import Any
from uuid import uuid4
from langchain_apify import ApifyDatasetLoader
from langchain_core.documents import Document
EXCLUDE_KEYS_FROM_CHECKSUM = {"metadata": {"chunk_id", "id", "checksum", "last_seen_at", "item_id"}}
DAY_IN_SECONDS = 24 * 3600
logger = logging.getLogger("apify")
def get_nested_value(d: dict, keys: str) -> Any:
"""
Extract nested value from dict.
Example:
>>> get_nested_value({"a": "v1", "c1": {"c2": "v2"}}, "c1.c2")
'v2'
"""
d = copy.deepcopy(d)
for key in keys.split("."):
if d and isinstance(d, dict) and d.get(key):
d = d[key]
else:
return ""
return d
def stringify_dict(d: dict, keys: list[str]) -> str:
"""Stringify all values in a dictionary.
Example:
>>> d_ = {"a": {"text": "Apify is cool"}, "description": "Apify platform"}
>>> stringify_dict(d_, ["a.text", "description"])
'a.text: Apify is cool\\ndescription: Apify platform'
"""
return "\n".join([f"{key}: {value}" for key in keys if (value := get_nested_value(d, key))])
def get_dataset_loader(dataset_id: str, fields: list[str], meta_object: dict, meta_fields: dict) -> ApifyDatasetLoader:
"""Load dataset by dataset_id using ApifyDatasetLoader.
The dataset_mapping_function is used to map the dataset item to a Document object.
Stringify dict using the fields.
"""
return ApifyDatasetLoader(
dataset_id,
dataset_mapping_function=lambda dataset_item: Document(
page_content=stringify_dict(dataset_item, fields) or "",
metadata={
**meta_object,
**{key: get_nested_value(dataset_item, value) for key, value in meta_fields.items()},
},
),
)
def compute_hash(text: str) -> str:
"""Compute hash of the text."""
return hashlib.sha256(text.encode()).hexdigest()
def get_chunks_to_delete(chunks_prev: list[Document], chunks_current: list[Document], expired_days: float) -> tuple[list[Document], list[Document]]:
"""
Identifies chunks to be deleted based on their last seen timestamp and presence in the current run.
Compare the chunks from the previous and current runs and identify chunks that are not present
in the current run and have not been updated within the specified 'expired_days'. These chunks are marked for deletion.
"""
ids_current = {d.metadata["item_id"] for d in chunks_current}
ts_expired = int(datetime.now(timezone.utc).timestamp() - expired_days * DAY_IN_SECONDS)
chunks_expired_delete, chunks_old_keep = [], []
# chunks that have been crawled in the current run and are older than ts_expired => to delete
for d in chunks_prev:
if d.metadata["item_id"] not in ids_current:
if d.metadata["last_seen_at"] < ts_expired:
chunks_expired_delete.append(d)
else:
chunks_old_keep.append(d)
return chunks_expired_delete, chunks_old_keep
def get_chunks_to_update(chunks_prev: list[Document], chunks_current: list[Document]) -> tuple[list[Document], list[Document]]:
"""
Identifies chunks that need to be updated or added based on their unique identifiers and checksums.
Compare the chunks from the previous and current runs and identify chunks that are new or have
undergone content changes by comparing their checksums. These chunks are marked for addition. chunks that are
present in both runs but have not undergone content changes are marked for metadata update.
"""
prev_id_checksum = defaultdict(list)
for chunk in chunks_prev:
prev_id_checksum[chunk.metadata["item_id"]].append(chunk.metadata["checksum"])
chunks_add = []
chunks_update_metadata = []
for chunk in chunks_current:
if chunk.metadata["item_id"] in prev_id_checksum:
if chunk.metadata["checksum"] in prev_id_checksum[chunk.metadata["item_id"]]:
chunks_update_metadata.append(chunk)
else:
chunks_add.append(chunk)
else:
chunks_add.append(chunk)
return chunks_add, chunks_update_metadata
def add_item_last_seen_at(items: list[Document]) -> list[Document]:
"""Add last_seen_at timestamp to the metadata of each dataset item."""
for item in items:
item.metadata["last_seen_at"] = int(datetime.now(timezone.utc).timestamp())
return items
def add_item_checksum(items: list[Document], dataset_fields_to_item_id: list[str]) -> list[Document]:
"""
Adds a checksum and unique item_id to the metadata of each dataset item.
This function computes a checksum for each item based on its content and metadata, excluding certain keys.
The checksum is then added to the document's metadata. Additionally, a unique item ID is generated based on
specified keys in the document's metadata and added to the metadata as well.
"""
for item in items:
item.metadata["checksum"] = compute_hash(item.json(exclude=EXCLUDE_KEYS_FROM_CHECKSUM))
hash_str = "".join([str(item.metadata[key]) for key in dataset_fields_to_item_id])
item.metadata["item_id"] = compute_hash(hash_str)
if not hash_str:
logger.warning(
"Item_id %s was generated with an empty hash. This typically means that `dataUpdatesPrimaryDatasetFields` "
"are empty or non-existent.",
item.metadata["item_id"],
)
return add_item_last_seen_at(items)
def add_chunk_id(chunks: list[Document]) -> list[Document]:
"""For every chunk (document stored in vector db) add chunk_id to metadata.
The chunk_id is a unique identifier for each chunk and is not required, but it is better to keep it in metadata.
"""
for d in chunks:
d.metadata["chunk_id"] = d.metadata.get("chunk_id", str(uuid4()))
return chunks

View File

@ -0,0 +1,177 @@
from __future__ import annotations
import concurrent.futures
import datetime
from collections import defaultdict
from typing import TYPE_CHECKING
from apify import Actor
from langchain_core.documents import Document
from langchain_core.vectorstores import VectorStore
from .models import OpengaussIntegration
from .utils import get_chunks_to_delete, get_chunks_to_update
if TYPE_CHECKING:
from langchain.vectorstores import VectorStore
from langchain_core.embeddings import Embeddings
from ._types import ActorInputsDb, VectorDb
async def get_vector_database(actor_input: ActorInputsDb | None, embeddings: Embeddings) -> VectorDb:
"""Get database based on the integration type."""
if isinstance(actor_input, OpengaussIntegration):
from .vector_stores.opengauss import OpenGaussDatabase
return OpenGaussDatabase(actor_input, embeddings)
raise ValueError("Unknown integration type")
def update_db_with_crawled_data(vector_store: VectorDb, documents: list[Document]) -> None:
"""Update the database with new crawled data."""
Actor.log.info("Comparing crawled data with the database ...")
data_add, ids_update_last_seen, ids_del = compare_crawled_data_with_db(vector_store, documents)
Actor.log.info("Objects: to add: %s, to update last_seen_at: %s, to delete: %s", len(data_add), len(ids_update_last_seen), len(ids_del))
# Delete data that were updated
if ids_del:
vector_store.delete(ids_del)
Actor.log.info("Deleted %s objects from the vector store where the content has changed since the last update", len(ids_del))
# Add new data
if data_add:
Actor.log.info("Adding %s new objects to the vector store", len(data_add))
vector_store.add_documents(data_add, ids=[d.metadata["chunk_id"] for d in data_add])
Actor.log.info("Added %s new objects to the vector store", len(data_add))
# Update metadata data
if ids_update_last_seen:
vector_store.update_last_seen_at(ids_update_last_seen)
Actor.log.info("Updated last_seen_at metadata for %s objects", len(ids_update_last_seen))
def upsert_db_with_crawled_data(vector_store: VectorDb, documents: list[Document]) -> None:
"""Upsert crawled data into the database by first deleting all documents and then adding all the documents."""
Actor.log.info("Upsert crawled data into database")
Actor.log.info("Delete documents by item_id. This might take a while as documents are deleted one by one.")
for d in documents:
vector_store.delete_by_item_id(d.metadata["item_id"])
Actor.log.info("Delete documents by item_id. Done")
Actor.log.info("Add documents")
vector_store.add_documents(documents, ids=[d.metadata["chunk_id"] for d in documents])
Actor.log.info("Added %s new objects to the vector store", len(documents))
def delete_expired_objects(vector_store: VectorDb, timestamp_expired: int) -> None:
"""Delete expired objects from the database."""
if timestamp_expired:
dt = datetime.datetime.fromtimestamp(timestamp_expired, tz=datetime.timezone.utc)
Actor.log.info("About to delete objects from the database that were not seen since %s (timestamp: %s)", dt, timestamp_expired)
vector_store.delete_expired(timestamp_expired)
def get_items_ids_from_db(vector_store: VectorDb, data: list[Document]) -> dict[str, list[Document]]:
"""Get documents from the database by item_id."""
items_ids = {d.metadata["item_id"] for d in data}
def _get_item_id(item_id: str) -> tuple[str, list[Document]]:
return item_id, vector_store.get_by_item_id(item_id)
crawled_db = defaultdict(list)
with concurrent.futures.ThreadPoolExecutor() as executor:
future_to_item_id = {executor.submit(_get_item_id, item_id): item_id for item_id in items_ids}
for k, future in enumerate(concurrent.futures.as_completed(future_to_item_id)):
item_id = future_to_item_id[future]
if k % 1000 == 0:
Actor.log.info("Processing item_id %s (%d/%d) to compare crawled data with the database", item_id, k, len(items_ids))
try:
item_id, documents = future.result()
crawled_db[item_id].extend(documents)
except Exception as e:
Actor.log.error("Item_id %s generated an error", item_id, e)
return dict(crawled_db)
def compare_crawled_data_with_db(vector_store: VectorDb, data: list[Document]) -> tuple[list[Document], list[str], list[str]]:
"""Compare current crawled data with the data in the database. Return data to add, delete and update.
New data is added
Data that was not changed -> update metadata last_seen_at
Data that was changed -> delete and add new
"""
data_add = []
ids_delete: set[str] = set()
ids_update_last_seen: set[str] = set()
if hasattr(vector_store, "count") and vector_store.count() == 0:
return data, [], []
crawled_db = get_items_ids_from_db(vector_store, data)
for d in data:
if res := crawled_db.get(d.metadata["item_id"]):
if d.metadata["checksum"] in {r.metadata["checksum"] for r in res}:
# Because of weaviate database, we need to use chunk_id instead of id
ids_update_last_seen.update({r.metadata.get("id") or r.metadata.get("chunk_id", ""): r for r in res})
else:
ids_delete.update({r.metadata.get("id") or r.metadata.get("chunk_id", ""): r for r in res})
data_add.append(d)
else:
data_add.append(d)
return data_add, list(ids_update_last_seen), list(ids_delete)
async def update_db_with_crawled_data_using_internal_cache(
vector_store: VectorStore, documents: list[Document], cache_key_name: str, cache_kv_store_name: str, expired_days: float
) -> None:
"""
Updates the vector store with new documents and removes outdated ones.
This function uses Apify's key-value store to handle documents. Each document, along with its metadata,
is hashed and stored as a key in the key-value store.
The function performs a comparison between the current set of documents and the set from the previous runs.
It identifies new documents and those no longer present. New documents are added to the vector store, while
documents older than the specified 'expired_days' are removed.
"""
Actor.log.info("Load previous cache %s from the key-value store: %s", cache_key_name, cache_kv_store_name)
kv_store = await Actor.open_key_value_store(name=cache_kv_store_name)
previous_runs = await kv_store.get_value(cache_key_name) or {}
previous_runs = [Document.parse_obj(doc) for doc in previous_runs]
Actor.log.info("Previous runs contains: %s records", len(previous_runs))
chunks_to_add, chunks_to_update = get_chunks_to_update(previous_runs, documents)
chunks_to_delete, chunks_old_keep = get_chunks_to_delete(previous_runs, documents, expired_days=expired_days)
Actor.log.info("Chunks to add: %s, chunks to update last_seen metadata: %s", len(chunks_to_add), len(chunks_to_update))
Actor.log.info("Chunks to delete: %s", len(chunks_to_delete))
if chunks_to_delete:
vector_store.delete(ids=[x.metadata["id"] for x in chunks_to_delete if x.metadata["id"]])
Actor.log.info("Deleted %s from database", len(chunks_to_delete))
if chunks_to_add:
inserted = vector_store.add_documents(chunks_to_add, ids=[x.metadata["id"] for x in chunks_to_add if x.metadata["id"]])
Actor.log.info("Added %s documents to the vector store", len(inserted))
else:
Actor.log.info("No new documents to add")
# update cache
if current_cache := chunks_to_add + chunks_to_update + chunks_old_keep:
await kv_store.set_value(cache_key_name, current_cache)
Actor.log.info("Updated cache: %s in the key-value store with %s entries", cache_key_name, len(current_cache))
Actor.log.info("Push chunked data to the unnamed output dataset")
await Actor.push_data([doc.dict() for doc in current_cache])

View File

@ -0,0 +1,4 @@
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .opengauss import OpenGaussDatabase

View File

@ -0,0 +1,43 @@
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from langchain_core.documents import Document
BACKOFF_MAX_TIME_SECONDS = 900
BACKOFF_MAX_TIME_DELETE_SECONDS = 900 # 15 minutes (if many objects were added it takes time to search in the database)
class VectorDbBase(ABC):
# only for testing purposes (to wait for the index to be updated, e.g. in Pinecone)
unit_test_wait_for_index = 0
@abstractmethod
def get_by_item_id(self, item_id: str) -> list[Document]:
"""Get documents by item_id."""
@abstractmethod
def update_last_seen_at(self, ids: list[str], last_seen_at: int | None = None) -> None:
"""Update last_seen_at field in the database."""
@abstractmethod
def delete_by_item_id(self, item_id: str) -> None:
"""Delete documents by item_id."""
@abstractmethod
def delete_expired(self, expired_ts: int) -> None:
"""Delete documents that are older than the ts_expired timestamp."""
@abstractmethod
def delete_all(self) -> None:
"""Delete all documents from the database (internal function for testing purposes)."""
@abstractmethod
async def is_connected(self) -> bool:
"""Check if the database is connected."""
@abstractmethod
def search_by_vector(self, vector: list[float], k: int, filter_: dict | None = None) -> list[Document]:
"""Search for documents by vector. Return a list of documents."""

View File

@ -0,0 +1,165 @@
from __future__ import annotations
import json
from datetime import datetime, timezone
from typing import TYPE_CHECKING
from langchain_core.documents import Document
from langchain_opengauss import OpenGauss
from langchain_opengauss.config import OpenGaussSettings
from psycopg2 import sql
from .base import VectorDbBase
if TYPE_CHECKING:
from langchain_core.embeddings import Embeddings
from ..models import OpengaussIntegration
class OpenGaussDatabase(OpenGauss, VectorDbBase):
def __init__(self, actor_input: OpengaussIntegration, embeddings: Embeddings) -> None:
try:
db_host = actor_input.opengaussHost
db_port = actor_input.opengaussPort
db_user = actor_input.opengaussUser
db_password = actor_input.opengaussPassword
db_name = actor_input.opengaussDBname or "postgres"
except (ValueError, IndexError) as e:
raise ValueError(
"Could not construct openGauss connection parameters from actor_input. "
"Ensure fields follow opengauss_input_model.py."
) from e
dummy_vector = embeddings.embed_query("get dimension")
embedding_dim = len(dummy_vector)
settings = OpenGaussSettings(
host=db_host,
port=db_port,
user=db_user,
password=db_password,
database=db_name,
table_name=actor_input.opengaussTableName,
embedding_dimension=embedding_dim,
)
super().__init__(embedding=embeddings, config=settings)
self._dummy_vector: list[float] = dummy_vector
@property
def dummy_vector(self) -> list[float]:
if not self._dummy_vector and self.embeddings:
self._dummy_vector = self.embeddings.embed_query("dummy")
return self._dummy_vector
async def is_connected(self) -> bool:
try:
with self._get_cursor() as cur:
cur.execute("SELECT 1")
return cur.fetchone()[0] == 1
except Exception:
return False
def get_by_item_id(self, item_id: str) -> list[Document]:
"""
Get all document chunks associated with a specific item_id from the metadata.
"""
if not item_id:
return []
query = sql.SQL("""
SELECT id, metadata FROM {table}
WHERE metadata ->> 'item_id' = %s
""").format(table=sql.Identifier(self.config.table_name))
docs = []
with self._get_cursor() as cur:
cur.execute(query, (item_id,))
for row in cur.fetchall():
doc_id, metadata = row
metadata["chunk_id"] = doc_id
docs.append(Document(page_content="", metadata=metadata))
return docs
def update_last_seen_at(self, ids: list[str], last_seen_at: int | None = None) -> None:
"""Update last_seen_at field in the database."""
if not ids:
return
last_seen_at = last_seen_at or int(datetime.now(timezone.utc).timestamp())
new_value_jsonb = json.dumps(last_seen_at)
update_sql = sql.SQL("""
UPDATE {table}
SET metadata = jsonb_set(
metadata,
'{{last_seen_at}}',
%s::jsonb,
true
)
WHERE id = ANY(%s)
""").format(table=sql.Identifier(self.config.table_name))
with self._get_cursor() as cur:
cur.execute(update_sql, (new_value_jsonb, ids))
def delete_by_item_id(self, item_id: str) -> None:
"""Delete object by item_id."""
if not item_id:
return
query = sql.SQL("""
DELETE FROM {table}
WHERE metadata ->> 'item_id' = %s
""").format(table=sql.Identifier(self.config.table_name))
with self._get_cursor() as cur:
cur.execute(query, (item_id,))
def delete_expired(self, expired_ts: int) -> None:
"""Delete objects from the index that are expired."""
query = sql.SQL("""
DELETE FROM {table}
WHERE (metadata ->> 'last_seen_at')::bigint < %s
""").format(table=sql.Identifier(self.config.table_name))
with self._get_cursor() as cur:
cur.execute(query, (expired_ts,))
def get_by_id(self, id_: str) -> Document | None:
"""Get a document by id from the database.
Used only for testing purposes.
"""
results = self.get_by_ids([id_])
return results[0] if results else None
def get_all_ids(self) -> list[str]:
"""Get all document ids from the database.
Used only for testing purposes.
"""
query = sql.SQL("SELECT id FROM {table}").format(
table=sql.Identifier(self.config.table_name)
)
with self._get_cursor() as cur:
cur.execute(query)
return [row[0] for row in cur.fetchall()]
def delete_all(self) -> None:
"""Delete all documents from the database.
Used only for testing purposes.
"""
self.delete(ids=None)
def search_by_vector(
self,
vector: list[float],
k: int = 4,
filter_: dict | None = None
) -> list[Document]:
"""Search by vector and return the results."""
return self.similarity_search_by_vector(embedding=vector, k=k, filter=filter_)

View File

@ -0,0 +1,5 @@
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))

View File

@ -0,0 +1,95 @@
from __future__ import annotations
import os
import time
import pytest
from dotenv import load_dotenv
from langchain_core.documents import Document
from langchain_openai.embeddings import OpenAIEmbeddings
from models import OpengaussIntegration # type: ignore
from utils import add_item_checksum # type: ignore[import-not-found]
from vector_stores.opengauss import OpenGaussDatabase # type: ignore[import-not-found]
load_dotenv()
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
INDEX_NAME = "apifyunittest"
# Database fixtures to test. Fill here the name of the fixtures you want to test
DATABASE_FIXTURES = [
"db_opengauss",
]
UUID = "00000000-0000-0000-0000-0000000000"
ID1 = f"{UUID}10"
ID2 = f"{UUID}20"
ID3 = f"{UUID}30"
ID4A, ID4B, ID4C = f"{UUID}4a", f"{UUID}4b", f"{UUID}4c"
ID5A, ID5B, ID5C = f"{UUID}5a", f"{UUID}5b", f"{UUID}5c"
ID6 = f"{UUID}60"
ITEM_ID1 = "id1"
ITEM_ID4 = "id4"
d1 = Document(page_content="Expired->del", metadata={"item_id": ITEM_ID1, "chunk_id": ID1, "checksum": "1", "last_seen_at": 0})
d2 = Document(page_content="Old->not-del", metadata={"item_id": "id2", "chunk_id": ID2, "checksum": "2", "last_seen_at": 1})
d3a = Document(page_content="Unchanged->upt-meta", metadata={"item_id": "id3", "chunk_id": ID3, "checksum": "3", "last_seen_at": 1})
d3b = Document(page_content="Unchanged->upt-meta", metadata={"item_id": "id3", "chunk_id": ID3, "checksum": "3", "last_seen_at": 2})
d4a = Document(page_content="Changed->del", metadata={"item_id": ITEM_ID4, "chunk_id": ID4A, "checksum": "4", "last_seen_at": 1})
d4b = Document(page_content="Changed->del", metadata={"item_id": ITEM_ID4, "chunk_id": ID4B, "checksum": "4", "last_seen_at": 1})
d4c = Document(page_content="Changed->add-new", metadata={"item_id": ITEM_ID4, "chunk_id": ID4C, "checksum": "4c", "last_seen_at": 2})
d5a = Document(page_content="Changed->del", metadata={"item_id": "id5", "chunk_id": ID5A, "checksum": "5", "last_seen_at": 1})
d5b = Document(page_content="Changed->add-new", metadata={"item_id": "id5", "chunk_id": ID5B, "checksum": "5bc", "last_seen_at": 2})
d5c = Document(page_content="Changed->add-new", metadata={"item_id": "id5", "chunk_id": ID5C, "checksum": "5bc", "last_seen_at": 2})
d6 = Document(page_content="New->add", metadata={"item_id": "id5", "chunk_id": ID6, "checksum": "6", "last_seen_at": 2})
@pytest.fixture()
def crawl_1() -> list[Document]:
return [d1, d2, d3a, d4a, d4b, d5a]
@pytest.fixture()
def crawl_2() -> list[Document]:
return [d3b, d4c, d5b, d5c, d6]
@pytest.fixture()
def expected_results() -> list[Document]:
return [d2, d3b, d4c, d5b, d5c, d6]
@pytest.fixture()
def documents() -> list[Document]:
d = Document(page_content="Content", metadata={"url": "https://url1.com"})
return add_item_checksum([d], ["url"]) # type: ignore[no-any-return]
@pytest.fixture()
def db_opengauss(crawl_1: list[Document]) -> OpenGaussDatabase:
db = OpenGaussDatabase(
actor_input=OpengaussIntegration(
opengaussHost=os.getenv("OPENGAUSS_HOST"),
opengaussPort=os.getenv("OPENGAUSS_PORT"),
opengaussUser=os.getenv("OPENGAUSS_USER"),
opengaussPassword=os.getenv("OPENGAUSS_PASSWORD"),
opengaussDBname=os.getenv("OPENGAUSS_DBNAME"),
opengaussTableName=INDEX_NAME,
embeddingsProvider="OpenAI",
embeddingsApiKey=os.getenv("OPENAI_API_KEY"),
datasetFields=["text"],
),
embeddings=embeddings,
)
db.unit_test_wait_for_index = 0
db.delete_all()
# Insert initially crawled objects
db.add_documents(documents=crawl_1, ids=[d.metadata["chunk_id"] for d in crawl_1])
time.sleep(db.unit_test_wait_for_index)
yield db
db.delete_all()

View File

@ -0,0 +1,166 @@
from __future__ import annotations
import copy
from langchain_core.documents import Document
from src.utils import (
add_item_checksum,
compute_hash,
get_chunks_to_delete,
get_chunks_to_update,
get_dataset_loader,
get_nested_value,
stringify_dict,
)
def test_get_nested_value_with_nested_keys() -> None:
d = {"a": {"b": {"c": "value"}}}
assert get_nested_value(d, "a.b.c") == "value"
def test_get_nested_value_with_top_level_key() -> None:
d = {"a": "value"}
assert get_nested_value(d, "a") == "value"
def test_get_nested_value_with_nonexistent_key() -> None:
d = {"a": "value"}
assert get_nested_value(d, "b") == ""
def test_get_nested_value_with_empty_dict() -> None:
assert get_nested_value({}, "a") == ""
def test_stringify_dict_with_multiple_keys() -> None:
d = {"a": "value1", "b": "value2"}
keys = ["a", "b"]
assert stringify_dict(d, keys) == "a: value1\nb: value2"
def test_stringify_dict_with_nested_keys() -> None:
d = {"a": {"b": "value"}}
keys = ["a.b"]
assert stringify_dict(d, keys) == "a.b: value"
def test_stringify_dict_with_nonexistent_keys() -> None:
d = {"a": "value"}
keys = ["b"]
assert stringify_dict(d, keys) == ""
def test_stringify_dict_with_empty_dict() -> None:
assert stringify_dict({}, ["a"]) == ""
def test_load_page_content() -> None:
dataset_items = [{"text": "This is a test"}]
loader = get_dataset_loader("1234", ["text"], {}, {})
result = list(map(loader.dataset_mapping_function, dataset_items))
assert result == [Document(page_content="text: This is a test")]
def test_load_page_content_with_metadata() -> None:
dataset_items = [
{"text": "This is a test", "url": "https://example.com", "metadata": {"title": "Test Title"}},
{"text": "Another test", "url": "https://example2.com", "metadata": {"title": "Test Title 2"}},
]
meta_values = {"source": "test source"}
meta_fields = {"page_url": "url", "page_title": "metadata.title"}
loader = get_dataset_loader("1234", ["text", "url"], meta_values, meta_fields)
result = list(map(loader.dataset_mapping_function, dataset_items))
expected_result = [
Document(
page_content="text: This is a test\nurl: https://example.com",
metadata={"source": "test source", "page_url": "https://example.com", "page_title": "Test Title"},
),
Document(
page_content="text: Another test\nurl: https://example2.com",
metadata={"source": "test source", "page_url": "https://example2.com", "page_title": "Test Title 2"},
),
]
assert result == expected_result
def test_compute_hash() -> None:
text = "test"
assert compute_hash(text) == "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08"
def test_get_chunks_empty() -> None:
add_, update_ = get_chunks_to_update([], [])
assert len(add_) == 0
assert len(update_) == 0
def test_get_chunks_previous_run_empty(documents: list[Document]) -> None:
add_, update_ = get_chunks_to_update([], documents)
assert len(add_) == 1
assert len(update_) == 0
assert add_[0].metadata["item_id"] == documents[0].metadata["item_id"]
def test_get_chunks_current_run_empty(documents: list[Document]) -> None:
add_, update_ = get_chunks_to_update(documents, [])
assert len(add_) == 0
assert len(update_) == 0
def test_get_chunks_update_metadata(documents: list[Document]) -> None:
chunks = add_item_checksum(documents, ["url"])
add_, update_ = get_chunks_to_update(chunks, chunks)
assert len(add_) == 0
assert len(update_) == 1
assert update_[0].metadata["checksum"] == "0feb3e25afe9430e2d23d726cbe2cecccef2afff29cdecf7a747264433605fa4"
assert update_[0].metadata["item_id"] == "f2881510b05f8c3567c1d63a3212d3ebb8bbfc5510241db1f39da8f66df1defd"
def test_get_chunks_to_update_with_content_changes(documents: list[Document]) -> None:
chunks_prev = add_item_checksum(documents, ["url"])
chunks_curr = copy.deepcopy(chunks_prev)
chunks_curr[0].page_content = "Content has changed between runs"
chunks_curr = add_item_checksum(chunks_curr, ["url"])
assert chunks_prev[0].metadata["item_id"] == chunks_curr[0].metadata["item_id"]
assert chunks_prev[0].metadata["checksum"] != chunks_curr[0].metadata["checksum"]
add_, update_ = get_chunks_to_update(chunks_prev, chunks_curr)
assert len(add_) == 1
assert len(update_) == 0
assert add_[0] == chunks_curr[0]
def test_get_chunks_to_delete_empty() -> None:
chunks_prev = add_item_checksum([], ["url"])
delete_, old_keep_ = get_chunks_to_delete(chunks_prev, chunks_prev, 1)
assert len(delete_) == 0
assert len(old_keep_) == 0
def test_get_chunks_to_delete_no_delete(documents: list[Document]) -> None:
chunks_prev = add_item_checksum(documents, ["url"])
delete_, old_keep_ = get_chunks_to_delete(chunks_prev, chunks_prev, 1)
assert len(delete_) == 0
assert len(old_keep_) == 0
def test_get_chunks_to_delete_delete_expired(documents: list[Document]) -> None:
chunks_prev = add_item_checksum(documents, ["url"])
chunks_prev[0].metadata["last_seen_at"] = 1
delete_, old_keep = get_chunks_to_delete(chunks_prev, [], 1)
assert len(delete_) == 1
assert len(old_keep) == 0
assert delete_[0] == chunks_prev[0]

View File

@ -0,0 +1,251 @@
from __future__ import annotations
import time
from datetime import datetime, timezone
from typing import TYPE_CHECKING
import pytest
from src.vcs import compare_crawled_data_with_db, delete_expired_objects, update_db_with_crawled_data
from .conftest import DATABASE_FIXTURES, ID1, ID3, ID4A, ID4B, ID4C, ID5A, ID5B, ID5C, ID6, ITEM_ID1, ITEM_ID4
if TYPE_CHECKING:
from _pytest.fixtures import FixtureRequest
from langchain_core.documents import Document
from src._types import VectorDb
def wait_for_db(sec: int = 3) -> None:
# Wait for the database to update (Pinecone)
# Data freshness - Pinecone is eventually consistent, so there can be a slight delay before new or changed records are visible to queries.
time.sleep(sec)
# Helper to compute the expected stored chunk_id
def get_expected_id(db: VectorDb, item_id: str, chunk_id: str) -> str:
if hasattr(db, "use_id_prefix") and getattr(db, "use_id_prefix", True):
return f"{item_id}#{chunk_id}"
return chunk_id
@pytest.mark.integration()
@pytest.mark.parametrize("input_db", DATABASE_FIXTURES)
def test_add_newly_crawled_data(input_db: str, crawl_2: list[Document], request: FixtureRequest) -> None:
db: VectorDb = request.getfixturevalue(input_db)
res = db.search_by_vector(db.dummy_vector, k=10)
assert len(res) == 6, "Expected 6 initial objects in the database"
data_add, _, _ = compare_crawled_data_with_db(db, crawl_2)
assert len(data_add) == 4, "Expected 4 objects to add"
assert data_add[0].metadata["chunk_id"] == ID4C
assert data_add[1].metadata["chunk_id"] == ID5B
assert data_add[2].metadata["chunk_id"] == ID5C
assert data_add[3].metadata["chunk_id"] == ID6
# Add new data
db.add_documents(data_add, ids=[d.metadata["chunk_id"] for d in data_add])
wait_for_db(db.unit_test_wait_for_index)
id4c = get_expected_id(db, "id4", ID4C)
id5b = get_expected_id(db, "id5", ID5B)
id5c = get_expected_id(db, "id5", ID5C)
res = db.search_by_vector(db.dummy_vector, k=10)
ids = [r.metadata["chunk_id"] for r in res]
assert len(res) == 10, "Expected 10 objects in the database after addition"
assert id4c in ids, f"Expected {id4c} to be added"
assert id5b in ids, f"Expected {id5b} to be added"
assert id5c in ids, f"Expected {id5c} to be added"
@pytest.mark.integration()
@pytest.mark.parametrize("input_db", DATABASE_FIXTURES)
def test_get_by_item_id(input_db: str, request: FixtureRequest) -> None:
db: VectorDb = request.getfixturevalue(input_db)
res = db.get_by_item_id(ITEM_ID1)
id1 = get_expected_id(db, "id1", ID1)
id4a = get_expected_id(db, "id4", ID4A)
id4b = get_expected_id(db, "id4", ID4B)
assert len(res) == 1, "Expected 1 object to be returned"
assert res[0].metadata["item_id"] == ITEM_ID1, f"Expected {ITEM_ID1} to be returned"
assert res[0].metadata["chunk_id"] == id1, f"Expected {id1} to be returned"
res = db.get_by_item_id(ITEM_ID4)
assert len(res) == 2, "Expected 2 objects to be returned"
ids = [r.metadata["chunk_id"] for r in res]
assert id4a in ids, f"Expected {id4a} to be returned"
assert id4b in ids, f"Expected {id4b} to be returned"
res = db.get_by_item_id("idX")
assert not res, "Expected [] to be returned"
res = db.get_by_item_id("")
assert not res, "Expected [] to be returned"
@pytest.mark.integration()
@pytest.mark.parametrize("input_db", DATABASE_FIXTURES)
def test_update_metadata_last_seen_at(input_db: str, crawl_2: list[Document], request: FixtureRequest) -> None:
# to test whether the object was updated
ts_init = int(datetime.now(timezone.utc).timestamp())
db: VectorDb = request.getfixturevalue(input_db)
id3 = get_expected_id(db, "id3", ID3)
res = db.search_by_vector(db.dummy_vector, k=10)
assert len(res) == 6, "Expected 6 initial objects in the database"
_, ids_update_last_seen, _ = compare_crawled_data_with_db(db, crawl_2)
assert len(ids_update_last_seen) == 1, "Expected 1 object to update"
ids_orig = ids_update_last_seen
if hasattr(db, "get_by_id") and (v := db.get_by_id(ids_update_last_seen[0])):
ids_orig = v.metadata["chunk_id"]
assert id3 in ids_orig, f"Expected {id3} to be updated"
res = db.search_by_vector(db.dummy_vector, k=10)
assert next(r for r in res if r.metadata["chunk_id"] == id3).metadata["last_seen_at"] == 1
# Update metadata data
db.update_last_seen_at(ids_update_last_seen)
wait_for_db(db.unit_test_wait_for_index)
res = db.search_by_vector(db.dummy_vector, k=10)
assert len(res) == 6, "Expected 6 objects in the database after last_seen update"
assert next(r for r in res if r.metadata["chunk_id"] == id3).metadata["last_seen_at"] >= ts_init, f"Expected {id3} to be updated"
@pytest.mark.integration()
@pytest.mark.parametrize("input_db", DATABASE_FIXTURES)
def test_delete_updated_data(input_db: str, crawl_2: list[Document], request: FixtureRequest) -> None:
db: VectorDb = request.getfixturevalue(input_db)
id4a = get_expected_id(db, "id4", ID4A)
id4b = get_expected_id(db, "id4", ID4B)
id5a = get_expected_id(db, "id5", ID5A)
res = db.search_by_vector(db.dummy_vector, k=10)
assert len(res) == 6, "Expected 6 initial objects in the database"
_, _, ids_del = compare_crawled_data_with_db(db, crawl_2)
# OpenSearch serverless does not support to create a document with ID, so we cannot check the ID directly
# Therefore, we need to get the ID from the database
ids_orig = ids_del
if hasattr(db, "get_by_id") and (data := [db.get_by_id(id_) for id_ in ids_del]):
ids_orig = [d.metadata["chunk_id"] for d in data if d]
assert len(ids_del) == 3, "Expected 1 object to delete"
assert id4a in ids_orig, f"Expected {id4a} to be deleted"
assert id4b in ids_orig, f"Expected {id4b} to be deleted"
assert id5a in ids_orig, f"Expected {id5a} to be deleted"
db.delete(ids=ids_del)
wait_for_db(db.unit_test_wait_for_index)
res = db.search_by_vector(db.dummy_vector, k=10)
ids = [r.metadata["chunk_id"] for r in res]
assert len(ids) == 3, "Expected 3 objects in the database after deletion"
assert id4a not in ids, f"Expected {id4a} to be deleted"
assert id4b not in ids, f"Expected {id4b} to be deleted"
assert id5a not in ids, f"Expected {id5a} to be deleted"
@pytest.mark.integration()
@pytest.mark.parametrize("input_db", DATABASE_FIXTURES)
def test_deleted_expired_data(input_db: str, request: FixtureRequest) -> None:
db: VectorDb = request.getfixturevalue(input_db)
id1 = get_expected_id(db, ITEM_ID1, ID1)
res = db.search_by_vector(db.dummy_vector, k=10)
assert len(res) == 6, "Expected 6 initial objects in the database"
# Delete expired objects
db.delete_expired(expired_ts=1)
wait_for_db(db.unit_test_wait_for_index)
res = db.search_by_vector(db.dummy_vector, k=10)
assert len(res) == 5, "Expected 5 objects in the database after deletion"
assert id1 not in [r.metadata["chunk_id"] for r in res], f"Expected {id1} to be deleted"
@pytest.mark.integration()
@pytest.mark.parametrize("input_db", DATABASE_FIXTURES)
def test_update_db_with_crawled_data_all(input_db: str, crawl_2: list[Document], expected_results: list[Document], request: FixtureRequest) -> None:
db: VectorDb = request.getfixturevalue(input_db)
res = db.search_by_vector(db.dummy_vector, k=10)
assert len(res) == 6, "Expected 6 initial objects in the database"
update_db_with_crawled_data(db, crawl_2)
wait_for_db(db.unit_test_wait_for_index)
delete_expired_objects(db, 1)
wait_for_db(db.unit_test_wait_for_index)
res = db.search_by_vector(db.dummy_vector, k=10)
assert len(res) == 6, "Expected 6 objects in the database after all updates"
# Compare results with expected results
for expected in expected_results:
_id = get_expected_id(db, expected.metadata["item_id"], expected.metadata["chunk_id"])
d = next(r for r in res if _id == r.metadata["chunk_id"])
assert d.metadata["item_id"] == expected.metadata["item_id"], f"Expected item_id {expected.metadata['item_id']}"
assert d.metadata["checksum"] == expected.metadata["checksum"], f"Expected checksum {expected.metadata['checksum']}"
@pytest.mark.integration()
@pytest.mark.parametrize("input_db", DATABASE_FIXTURES)
def test_get_delete_all(input_db: str, request: FixtureRequest) -> None:
"""Test that all items have benn deleted (delete_all is an internal function)."""
db: VectorDb = request.getfixturevalue(input_db)
res = db.search_by_vector(db.dummy_vector, k=10)
assert res
db.delete_all()
wait_for_db(db.unit_test_wait_for_index)
res = db.search_by_vector(db.dummy_vector, k=10)
assert not res
@pytest.mark.integration()
@pytest.mark.parametrize("input_db", DATABASE_FIXTURES)
def test_delete_by_item_id(input_db: str, request: FixtureRequest) -> None:
db: VectorDb = request.getfixturevalue(input_db)
res = db.search_by_vector(db.dummy_vector, k=10)
assert len(res) == 6, "Expected 6 initial objects in the database"
id4a = get_expected_id(db, "id4", ID4A)
id4b = get_expected_id(db, "id4", ID4B)
res = db.get_by_item_id(ITEM_ID4)
assert len(res) == 2, "Expected 2 objects to be returned"
ids = [r.metadata["chunk_id"] for r in res]
assert id4a in ids, f"Expected {id4a} to be returned"
assert id4b in ids, f"Expected {id4b} to be returned"
db.delete_by_item_id(ITEM_ID4)
wait_for_db(db.unit_test_wait_for_index)
res = db.get_by_item_id(ITEM_ID4)
assert not res, "Expected None to be returned"
db.delete_by_item_id("idX")
wait_for_db(db.unit_test_wait_for_index)
res = db.get_by_item_id("idX")
assert not res, "Expected None to be returned"

View File

@ -0,0 +1,20 @@
services:
opengauss:
    image: opengauss/opengauss-server:latest
    environment:
      GS_USERNAME: GSDBUSER
      GS_PASSWORD: Apify@123
      GS_DB: gs_apify
    volumes:
      - ./volumes/opengauss/data:/var/lib/opengauss/data
    healthcheck:
      test: [ "CMD-SHELL", "netstat -lntp | grep tcp6 > /dev/null 2>&1" ]
      interval: 10s
      timeout: 10s
      retries: 10
    ports:
       - "8888:5432"
networks:
default:
name: milvus

View File

@ -0,0 +1,4 @@
#!/usr/bin/env bash
echo "Building the Docker image"
docker build --tag vs_db --file shared/Dockerfile --build-arg ACTOR_PATH_IN_DOCKER_CONTEXT=actors/opensearch .

View File

@ -0,0 +1,11 @@
#!/usr/bin/env bash
echo "Stopping and removing the container"
docker stop vs_db 2> /dev/null || true
docker rm vs_db 2> /dev/null || true
echo "Running the container"
docker run \
--name vs_db \
-it \
vs_db

View File

@ -0,0 +1,4 @@
# Change Log
## 0.0.0 (2025-09-23)
- Initial import

View File

@ -0,0 +1,26 @@
FROM apify/actor-python:3.11
WORKDIR /usr/src/app
COPY code/pyproject.toml code/poetry.lock code/pyproject.toml ./
ARG ACTOR_PATH_IN_DOCKER_CONTEXT
ENV ACTOR_PATH_IN_DOCKER_CONTEXT="${ACTOR_PATH_IN_DOCKER_CONTEXT}"
RUN echo "Python version:" \
&& python --version \
&& echo "Pip version:" \
&& pip --version \
&& echo "Installing Poetry:" \
&& pip install --no-cache-dir poetry~=1.8 \
&& echo "Installing dependencies:" \
&& poetry config virtualenvs.create false \
&& poetry install --only "main,${ACTOR_PATH_IN_DOCKER_CONTEXT#actors/}" --no-interaction --no-ansi \
&& rm -rf /tmp/.poetry-cache \
&& echo "All installed Python packages:" \
&& pip freeze
COPY code ./
CMD ["python3", "-m", "src"]

View File

@ -0,0 +1,27 @@
{
"actorSpecification": 1,
"views": {
"overview": {
"title": "Overview",
"transformation": {
"fields": [
"page_content",
"metadata"
]
},
"display": {
"component": "table",
"properties": {
"page_content": {
"label": "Page content (chunk)",
"format": "text"
},
"metadata": {
"label": "Metadata (if any)",
"format": "object"
}
}
}
}
}
}