!96 merge master into master

OSPP: whyhow_opengauss

Created-by: paradox
Commit-by: paradox
Merged-by: opengauss_bot
Description: 【标题】openGauss向量数据库对接知识工程最佳实践

【实现内容】: WhyHow 与openGauss向量数据库适配,包括简单知识文档导入、对接大模型完成RAG端到端验证。

See merge request: opengauss/examples!96
This commit is contained in:
opengauss_bot 2025-10-28 11:00:03 +08:00
commit 96ceea0ac7
No known key found for this signature in database
GPG Key ID: 6685280F6FDE8141
139 changed files with 11017 additions and 0 deletions

75
whyhow_ospp/.env.sample Normal file
View File

@ -0,0 +1,75 @@
# ----------------------- #
# Minimum Environment Variables
# ----------------------- #
WHYHOW__EMBEDDING__OPENAI__API_KEY=<你的openai api key>
WHYHOW__GENERATIVE__OPENAI__API_KEY=<你的openai api key>
WHYHOW__OPENGAUSS__HOST=<数据库的host>
WHYHOW__OPENGAUSS__PORT=<数据库docker映射出来的端口>5432
WHYHOW__OPENGAUSS__DATABASE=<数据库名称>
WHYHOW__OPENGAUSS__USER=<数据库用户名>
WHYHOW__OPENGAUSS__PASSWORD=<数据库密码>
WHYHOW__OPENGAUSS__ECHO_SQL=<是否打印 SQL 语句>
# ----------------------- #
# Auth0
# ----------------------- #
# WHYHOW__API__AUTH0__DOMAIN
# WHYHOW__API__AUTH0__AUDIENCE
# WHYHOW__API__AUTH0__ALGORITHM
# ----------------------- #
# AWS
# ----------------------- #
# S3
# WHYHOW__AWS__S3__BUCKET
# ----------------------- #
# EMBEDDING
# ----------------------- #
# WHYHOW__EMBEDDING__PROVIDER # possible values: openai
# OPENAI
# WHYHOW__EMBEDDING__OPENAI__API_KEY
# WHYHOW__EMBEDDING__OPENAI__MODEL
# ----------------------- #
# DEV
# ----------------------- #
# WHYHOW__DEV__LOG_LEVEL # possible values: DEBUG, INFO, WARNING, ERROR, CRITICAL
# WHYHOW__DEV__OPENAPI_URL
# ----------------------- #
# GENERATIVE
# ----------------------- #
# WHYHOW__GENERATIVE__PROVIDER # possible values: openai, fake
# ----------------------- #
# LOGFIRE
# ----------------------- #
# WHYHOW__LOGFIRE__TOKEN
# OPENAI
# WHYHOW__GENERATIVE__OPENAI__API_KEY
# WHYHOW__GENERATIVE__OPENAI__MODEL
# WHYHOW__GENERATIVE__OPENAI__TEMPERATURE
# WHYHOW__GENERATIVE__OPENAI__MAX_TOKENS
# ----------------------- #
# STORE
# ----------------------- #
# MONGODB
# WHYHOW__MONGODB__USERNAME
# WHYHOW__MONGODB__PASSWORD
# WHYHOW__MONGODB__DATABASE_NAME
# WHYHOW__MONGODB__HOST
# WHYHOW__MONGODB__PUBLIC_KEY
# WHYHOW__MONGODB__PRIVATE_KEY
# WHYHOW__MONGODB__GROUP_ID
# WHYHOW__MONGODB__CLUSTER_NAME
# WHYHOW__MONGODB__CHUNK_COLLECTION_NAME
# WHYHOW__MONGODB__VECTOR_SEARCH_EMBEDDING_SIZE

149
whyhow_ospp/README.md Normal file
View File

@ -0,0 +1,149 @@
## openGauss向量数据库对接知识工程最佳实践
WhyHow 与openGauss向量数据库适配包括简单知识文档导入、对接大模型完成RAG端到端验证。
本项目旨在实现:
- 灵活的数据模型: 将 WhyHow 的知识图谱数据模型对接 openGauss充分利用其关系型和图数据库的双重能力。
- 端到端 RAG: 结合主流的嵌入模型和大型语言模型LLMs实现从文档摄取、知识抽取、图谱构建到智能问答的完整 RAG 工作流。
### 下载与部署
#### 1. 所需环境
- 2vCPUs | 4GiB | s7.large.2 CentOS 7.6 64bit
- Docker / Docker Compose
- Python 3.11强烈建议用conda创建虚拟环境
- openGauss 3.xdocker部署
- OpenAI API Key用于 Embedding/LLM
#### 2. 下载代码
```shell
git clone https://gitcode.com/paradox/whyhow_opengauss.git
cd knowledge-graph-studio
pip install -r requirements.txt
```
#### 3. openGauss 部署
1. 拉取docker镜像
```shell
docker pull enmotech/opengauss:latest
```
Tips建议服务器上面挂个代理或者本地拉取我本地有代理导入服务器
2. 创建opengauss容器并启动
```shell
docker run -d --name opengauss \
-e GS_PASSWORD='Enmo@123' \
-e GAUSSHOME=/usr/local/opengauss \
-e LD_LIBRARY_PATH=/usr/local/opengauss/lib \
-e PATH=/usr/local/opengauss/bin:$PATH \
-p 5432:5432 enmotech/opengauss:3.1.0
```
后续开机只需要启动就好
```shell
docker start opengauss
```
3. 创建表这里用的gsql
```shell
export GAUSSHOME=/usr/local/opengauss
export LD_LIBRARY_PATH=$GAUSSHOME/lib:$LD_LIBRARY_PATH
export PATH=$GAUSSHOME/bin:$PATH
gsql -d postgres -U gaussdb -W Enmo@123
```
#### 4. 配置环境变量
```shell
cp .env.sample .env
WHYHOW__EMBEDDING__OPENAI__API_KEY=<你的openai api key>
WHYHOW__GENERATIVE__OPENAI__API_KEY=<你的openai api key>
WHYHOW__OPENGAUSS__HOST=<数据库的host>
WHYHOW__OPENGAUSS__PORT=<数据库docker映射出来的端口>5432
WHYHOW__OPENGAUSS__DATABASE=<数据库名称>
WHYHOW__OPENGAUSS__USER=<数据库用户名>
WHYHOW__OPENGAUSS__PASSWORD=<数据库密码>
WHYHOW__OPENGAUSS__ECHO_SQL=<是否打印 SQL 语句>
# e.g.
# WHYHOW__OPENGAUSS__HOST=127.0.0.1
# WHYHOW__OPENGAUSS__PORT=5432
# WHYHOW__OPENGAUSS__DATABASE=postgres
# WHYHOW__OPENGAUSS__USER=gaussdb
# WHYHOW__OPENGAUSS__PASSWORD=Enmo@123
# WHYHOW__OPENGAUSS__ECHO_SQL=true
```
#### 5. 创建数据表(见适配文档)
#### 6. 启动API
当所有的配置完成就可以启动API服务器开始服务
```shell
uvicorn whyhow_api.main:app --host 0.0.0.0 --port 8000 --reload
```
### Quickstart
#### 1. 健康检查
```shell
curl -s -H "x-api-key: $KEY" "http://127.0.0.1:8000/db" | jq
```
#### 2. 创建工作区
```shell
cat <<JSON | curl -s -X POST \
-H "x-api-key: $KEY" -H "Content-Type: application/json" \
--data-binary @- "http://127.0.0.1:8000/workspaces" | jq
{
"name": "PG-demo",
"description": "demo ws"
}
JSON
export WS_ID=xxx
```
#### 3. 创建chunks
```shell
cat >/tmp/chunks_more.json <<'JSON'
{
"chunks_in": [
{
"content": "openGauss 是企业级开源数据库,兼容 PostgreSQL具备高可用与高性能。",
"tags": ["db","opengauss"],
"user_metadata": { "lang": "zh" }
},
{
"content": "WhyHow 支持文档分块、图谱抽取与 RAG 检索问答,便于企业知识应用。",
"tags": ["whyhow","rag"],
"user_metadata": { "lang": "zh" }
}
]
}
JSON
curl -s -X POST \
-H "x-api-key: $KEY" \
-H "Content-Type: application/json" \
--data-binary @/tmp/chunks_more.json \
"http://127.0.0.1:8000/chunks?workspace_id=$WS_ID" | jq
```
#### 4. RAG查询
```shell
curl -s -G -H "x-api-key: $KEY" \
--data-urlencode "workspace_id=$WS_ID" \
--data-urlencode "text=openGauss 的优势是什么?" \
--data-urlencode "top_k=5" \
"http://127.0.0.1:8000/queries/rag" \
| jq '{answer, retrieved:(.top_chunks // [])}'
```

233
whyhow_ospp/config.py Normal file
View File

@ -0,0 +1,233 @@
"""Configuration."""
import logging
from typing import Literal, Optional
from pydantic import BaseModel, SecretStr, field_validator, model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
from urllib.parse import quote_plus
logger = logging.getLogger(__name__)
OPENAI_TOKEN_COSTS = {
"gpt-4o": {"input": 5 / 1000000, "output": 15 / 1000000},
"gpt-3.5-turbo": {
"input": 0.5 / 1000000,
"output": 1.5 / 1000000,
},
"text-embedding-3-large": {"input": 0.002 / 1000000},
"text-embedding-3-small": {"input": 0.13 / 1000000},
}
OPENAI_RATE_LIMITS = {
1: {
"gpt-4o": {"rpm": 500, "tpm": 30000},
"gpt-3.5-turbo": {"rpm": 3500, "tpm": 200000},
"text-embedding-3-large": {"rpm": 300, "tpm": 1000000},
"text-embedding-3-small": {"rpm": 300, "tpm": 1000000},
},
2: {
"gpt-4o": {"rpm": 5000, "tpm": 450000},
"gpt-3.5-turbo": {"rpm": 3500, "tpm": 2000000},
"text-embedding-3-large": {"rpm": 5000, "tpm": 1000000},
"text-embedding-3-small": {"rpm": 5000, "tpm": 1000000},
},
3: {
"gpt-4o": {"rpm": 5000, "tpm": 800000},
"gpt-3.5-turbo": {"rpm": 3500, "tpm": 4000000},
"text-embedding-3-large": {"rpm": 5000, "tpm": 5000000},
"text-embedding-3-small": {"rpm": 5000, "tpm": 5000000},
},
4: {
"gpt-4o": {"rpm": 10000, "tpm": 2000000},
"gpt-3.5-turbo": {"rpm": 10000, "tpm": 10000000},
"text-embedding-3-large": {"rpm": 10000, "tpm": 5000000},
"text-embedding-3-small": {"rpm": 10000, "tpm": 5000000},
},
5: {
"gpt-4o": {"rpm": 10000, "tpm": 3000000},
"gpt-3.5-turbo": {"rpm": 10000, "tpm": 20000000},
"text-embedding-3-large": {"rpm": 10000, "tpm": 10000000},
"text-embedding-3-small": {"rpm": 10000, "tpm": 10000000},
},
}
OPENAI_TIERS = Literal[1, 2, 3, 4, 5]
class SettingsDev(BaseModel):
"""Developer / runtime switches."""
log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO"
openapi_url: Optional[str] = "/openapi.json"
model_config = dict(frozen=True)
@field_validator("openapi_url", mode="before")
@classmethod
def empty_to_none(cls, v: Optional[str]) -> Optional[str]:
return None if v == "" else v
class SettingsAPI(BaseModel):
"""API settings."""
# auth0: SettingsAuth0 = SettingsAuth0()
limit_frequency_value: int = 30 # tokens added per second
bucket_capacity: int = 45 # max tokens in bucket
excluded_paths: list[str] = [
"/",
"/openapi.json",
"/docs",
] # These paths are excluded from rate limiting
public_paths: list[str] = [
"/graphs/public",
"/graphs/public/triples",
"/graphs/public/nodes",
"/graphs/public/chunks",
"/graphs/public/rules",
] # These paths are rate limited but not authenticated
max_chars_per_chunk: int = 1024
max_patterns: int = 64
max_chunk_pattern_product: int = 512
max_chunk_per_batch: int = 1
query_sim_triple_limit: int = (
64 # max number of triples in a similarity search query
)
query_sim_triple_candidates: int = (
64 # max number of candidates to consider (default mongodb)
)
restrict_structured_chunk_retrieval: bool = False
model_config = SettingsConfigDict(frozen=True)
class SettingsGenerativeOpenAI(BaseModel):
"""OpenAI settings."""
api_key: SecretStr | None = None
model: str = "gpt-4o"
temperature: float = 0.0
max_tokens: int = 3000
tier: OPENAI_TIERS = 2
@property
def rpm_limit(self) -> int:
"""Get the RPM limit."""
return OPENAI_RATE_LIMITS[self.tier][self.model]["rpm"]
@property
def tpm_limit(self) -> int:
"""Get the TPM limit."""
return OPENAI_RATE_LIMITS[self.tier][self.model]["tpm"]
@property
def input_token_cost(self) -> float:
"""Get the input token cost."""
return OPENAI_TOKEN_COSTS[self.model]["input"]
@property
def output_token_cost(self) -> float:
"""Get the output token cost."""
return OPENAI_TOKEN_COSTS[self.model]["output"]
model_config = SettingsConfigDict(frozen=True)
class SettingsGenerative(BaseModel):
"""Generative settings."""
provider: Literal["openai"] = "openai"
openai: SettingsGenerativeOpenAI = SettingsGenerativeOpenAI()
model_config = SettingsConfigDict(frozen=True)
class SettingsEmbeddingOpenAI(BaseModel):
"""OpenAI settings."""
api_key: SecretStr | None = None
model: str = "text-embedding-3-large"
model_config = SettingsConfigDict(frozen=True)
class SettingsEmbedding(BaseModel):
"""Embedding settings."""
provider: Literal["openai"] = "openai"
openai: SettingsEmbeddingOpenAI = SettingsEmbeddingOpenAI()
model_config = SettingsConfigDict(frozen=True)
class SettingsS3(BaseModel):
"""S3 settings."""
bucket: str = ""
presigned_post_expiration: int = 360 # in seconds
presigned_download_expiration: int = 360 # in seconds
presigned_post_max_bytes: int = 50 * int(1e6)
model_config = SettingsConfigDict(frozen=True)
class SettingsAWS(BaseModel):
"""AWS settings."""
s3: SettingsS3 = SettingsS3()
model_config = SettingsConfigDict(frozen=True)
class SettingsLogfire(BaseModel):
"""Logfire settings."""
token: SecretStr | None = None
# Create opengauss settings
class SettingsOpenGauss(BaseModel):
host: str = "127.0.0.1"
port: int = 5432
database: str = "whyhow"
user: str = "whyhow"
password: SecretStr = SecretStr("secret")
sslmode: str | None = None
echo_sql: bool = False
model_config = SettingsConfigDict(frozen=True)
@property
def dsn(self) -> str:
pw_raw = self.password.get_secret_value()
user_enc = quote_plus(self.user)
pw_enc = quote_plus(pw_raw)
qs = f"?sslmode={self.sslmode}" if self.sslmode else ""
return (
f"postgresql+asyncpg://{user_enc}:{pw_enc}"
f"@{self.host}:{self.port}/{self.database}{qs}"
)
class Settings(BaseSettings):
"""All settings."""
api: SettingsAPI = SettingsAPI()
aws: SettingsAWS = SettingsAWS()
dev: SettingsDev = SettingsDev()
embedding: SettingsEmbedding = SettingsEmbedding()
generative: SettingsGenerative = SettingsGenerative()
opengauss: SettingsOpenGauss = SettingsOpenGauss()
logfire: SettingsLogfire = SettingsLogfire()
model_config = SettingsConfigDict(
env_file=".env",
env_prefix="WHYHOW__",
env_nested_delimiter="__",
frozen=True,
case_sensitive=False,
)
@model_validator(mode="after")
def check(self) -> "Settings":
"""Check everything correct."""
return self

View File

@ -0,0 +1,84 @@
"""Logging configuration for the project."""
from logging.config import dictConfig
from typing import Literal
LOG_LEVEL_TYPE = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
def configure_logging(project_log_level: LOG_LEVEL_TYPE = "INFO") -> None:
"""Configure logging for the project."""
dictConfig(
{
"version": 1,
"disable_existing_loggers": False,
"filters": {
"correlation_id": {
"()": "asgi_correlation_id.CorrelationIdFilter",
"uuid_length": 8,
"default_value": "-",
},
},
"formatters": {
"console": {
"class": "logging.Formatter",
"datefmt": "%Y-%m-%d %H:%M:%S",
"format": "%(levelname)s: %(asctime)s %(name)s:%(lineno)d [%(correlation_id)s] %(message)s",
},
# Copy uvicorn's formatters
"access": {
"()": "uvicorn.logging.AccessFormatter",
"fmt": "%(levelprefix)s %(client_addr)s - %(request_line)s [%(correlation_id)s] %(status_code)s",
},
"default": {
"()": "uvicorn.logging.DefaultFormatter",
"fmt": "%(levelprefix)s %(message)s",
"use_colors": None,
},
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"filters": ["correlation_id"],
"formatter": "console",
"stream": "ext://sys.stdout",
},
# Copy uvicorn's handlers
"access": {
"class": "logging.StreamHandler",
"filters": ["correlation_id"],
"formatter": "access",
"stream": "ext://sys.stdout",
},
"default": {
"class": "logging.StreamHandler",
"formatter": "default",
"stream": "ext://sys.stderr",
},
},
"loggers": {
# root logger
"": {"handlers": ["console"], "level": "WARNING"},
# project logger
"whyhow_api": {
"handlers": ["console"],
"level": project_log_level,
"propagate": False,
},
# uvicorn loggers
"uvicorn": {
"handlers": ["default"],
"level": "INFO",
"propagate": False,
},
"uvicorn.error": {
"level": "INFO",
},
"uvicorn.access": {
"handlers": ["access"],
"level": "INFO",
"propagate": False,
},
},
}
)

66
whyhow_ospp/database.py Normal file
View File

@ -0,0 +1,66 @@
"""Database connection and session management (PG-only)."""
import logging
import re
from contextlib import asynccontextmanager
from typing import AsyncGenerator
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql.base import PGDialect
from sqlalchemy.ext.asyncio import (
AsyncEngine,
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from whyhow_api.config import Settings
logger = logging.getLogger(__name__)
pg_engine: AsyncEngine | None = None
pg_sessionmaker: async_sessionmaker[AsyncSession] | None = None
# --- openGauss server_version ---
def _og_get_server_version_info(self, connection):
v = connection.exec_driver_sql("select version()").scalar()
if not isinstance(v, str):
return (13, 0)
m = re.search(r"openGauss\s+(\d+)\.(\d+)\.(\d+)", v, re.IGNORECASE)
if m:
return tuple(int(x) for x in m.groups())
m = re.search(r"(\d+)\.(\d+)(?:\.(\d+))?", v)
if m:
return tuple(int(x) for x in m.groups() if x is not None)
return (13, 0)
PGDialect._get_server_version_info = _og_get_server_version_info
async def connect_to_pg(settings: Settings) -> None:
"""初始化 openGauss/Postgres 引擎与会话工厂。"""
global pg_engine, pg_sessionmaker
if pg_engine is None:
pg_engine = create_async_engine(
settings.opengauss.dsn, # postgresql+asyncpg://user:pass@host:port/db
echo=settings.opengauss.echo_sql,
pool_pre_ping=True,
)
pg_sessionmaker = async_sessionmaker(pg_engine, expire_on_commit=False)
logger.info("Connected to openGauss/Postgres.")
async def close_pg() -> None:
"""关闭引擎。"""
global pg_engine, pg_sessionmaker
if pg_engine is not None:
await pg_engine.dispose()
pg_engine = None
pg_sessionmaker = None
logger.info("openGauss/Postgres connection closed.")
@asynccontextmanager
async def get_pg_session() -> AsyncGenerator[AsyncSession, None]:
"""获取 AsyncSession事务由调用方决定是否显式使用"""
if pg_sessionmaker is None:
raise RuntimeError("Postgres has not been initialised. Call connect_to_pg() first.")
async with pg_sessionmaker() as session:
yield session

181
whyhow_ospp/dependencies.py Normal file
View File

@ -0,0 +1,181 @@
"""Dependencies for FastAPI (PG-only)."""
import logging
from functools import cache
from typing import Any, AsyncGenerator, Dict, List
from uuid import UUID
from fastapi import Depends, HTTPException, Request, status
from fastapi.security import APIKeyHeader
import sqlalchemy as sa
from sqlalchemy.ext.asyncio import AsyncSession
from openai import AsyncAzureOpenAI, AsyncOpenAI
from pydantic import ValidationError
from whyhow_api.config import Settings
from whyhow_api.database import get_pg_session
from whyhow_api.models.common import LLMClient
from whyhow_api.schemas.chunks import ChunkDocumentModel
from whyhow_api.schemas.documents import DocumentOutWithWorkspaceDetails
from whyhow_api.schemas.graphs import CreateGraphBody, DetailedGraphDocumentModel, GraphDocumentModel
from whyhow_api.schemas.nodes import NodeDocumentModel
from whyhow_api.schemas.queries import QueryDocumentModel
from whyhow_api.schemas.schemas import SchemaDocumentModel, SchemaOutWithWorkspaceDetails
from whyhow_api.schemas.triples import TripleDocumentModel
from whyhow_api.schemas.users import BYOAzureOpenAIMetadata, BYOOpenAIMetadata, ProviderConfig
from whyhow_api.schemas.workspaces import WorkspaceDocumentModel
from whyhow_api.services.crud.user_pg import get_user_by_api_key
from whyhow_api.services.crud.workspace_pg import get_workspace
from whyhow_api.services.crud.schema_pg import get_schema_with_workspace
from whyhow_api.services.crud.document_pg import get_document_with_workspace
from whyhow_api.services.crud.chunks_pg import get_chunk_basic
from whyhow_api.services.crud.node_pg import get_node
from whyhow_api.services.crud.triple_pg import get_triple
from whyhow_api.services.crud.graph_pg import get_graph as get_graph_pg
from whyhow_api.services.crud.queries_pg import get_query
logger = logging.getLogger(__name__)
api_key_header = APIKeyHeader(name="x-api-key", auto_error=False)
@cache
def get_settings() -> Settings:
return Settings()
# ---------- PG 会话 ----------
async def get_pg() -> AsyncGenerator[AsyncSession, None]:
async with get_pg_session() as session:
yield session
# ---------- 用户鉴权 ----------
async def get_user_pg(
request: Request,
api_key: str | None = Depends(api_key_header),
session: AsyncSession = Depends(get_pg),
) -> UUID:
if not api_key:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing x-api-key")
row = await get_user_by_api_key(session, api_key)
if not row:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key")
return row["id"] # UUID
# ---------- LLM 客户端 ----------
async def get_llm_client(
user_id: UUID = Depends(get_user_pg),
session: AsyncSession = Depends(get_pg),
settings: Settings = Depends(get_settings),
) -> LLMClient:
"""
优先 users.providers(JSONB) 里的 BYO 配置
退化使用全局 Settingsopenai配置
"""
row = (await session.execute(
sa.text("SELECT providers FROM users WHERE id = :uid"),
{"uid": str(user_id)}
)).mappings().first()
providers_json = row["providers"] if row and "providers" in row else None
if providers_json:
try:
provider_config = ProviderConfig.model_validate({"providers": providers_json})
llm_providers = [p for p in provider_config.providers if p.type == "llm"]
if llm_providers:
lp = llm_providers[0]
if lp.value == "byo-azure-openai":
meta = BYOAzureOpenAIMetadata.model_validate(lp.metadata["byo-azure-openai"])
if not lp.api_key or not meta.api_version or not meta.azure_endpoint or not meta.language_model_name or not meta.embedding_name:
raise HTTPException(status_code=401, detail="Invalid BYO Azure OpenAI config")
client = AsyncAzureOpenAI(api_key=lp.api_key, api_version=meta.api_version, azure_endpoint=meta.azure_endpoint)
return LLMClient(client, meta)
elif lp.value == "byo-openai":
meta = BYOOpenAIMetadata.model_validate(lp.metadata["byo-openai"])
if not lp.api_key:
raise HTTPException(status_code=401, detail="Missing BYO OpenAI key")
client = AsyncOpenAI(api_key=lp.api_key)
return LLMClient(client, meta)
except ValidationError as e:
logger.error(f"Invalid provider config in PG: {e}")
if settings.generative.openai.api_key is None:
raise HTTPException(status_code=401, detail="No LLM provider configured")
client = AsyncOpenAI(api_key=settings.generative.openai.api_key.get_secret_value())
meta = BYOOpenAIMetadata(
language_model_name=settings.generative.openai.model,
embedding_name=settings.embedding.openai.model,
temperature=settings.generative.openai.temperature,
max_tokens=settings.generative.openai.max_tokens,
)
return LLMClient(client, meta)
# ---------- 资源校验PG 版) ----------
async def valid_workspace_id(workspace_id: UUID, user_id: UUID = Depends(get_user_pg), session: AsyncSession = Depends(get_pg)) -> WorkspaceDocumentModel:
ws = await get_workspace(session, workspace_id, user_id)
if ws is None:
raise HTTPException(status_code=404, detail="Workspace not found")
return WorkspaceDocumentModel.model_validate(ws)
async def valid_schema_id(schema_id: UUID, user_id: UUID = Depends(get_user_pg), session: AsyncSession = Depends(get_pg)) -> SchemaOutWithWorkspaceDetails:
sc = await get_schema_with_workspace(session, schema_id, user_id)
if sc is None:
raise HTTPException(status_code=404, detail="Schema not found")
return SchemaOutWithWorkspaceDetails.model_validate(sc)
async def valid_chunk_id(chunk_id: UUID, user_id: UUID = Depends(get_user_pg), session: AsyncSession = Depends(get_pg)) -> ChunkDocumentModel:
ck = await get_chunk_basic(session, chunk_id, user_id)
if ck is None:
raise HTTPException(status_code=404, detail="Chunk not found")
return ChunkDocumentModel.model_validate(ck)
async def valid_node_id(node_id: UUID, user_id: UUID = Depends(get_user_pg), session: AsyncSession = Depends(get_pg)) -> NodeDocumentModel:
nd = await get_node(session, node_id, user_id)
if nd is None:
raise HTTPException(status_code=404, detail="Node not found")
return NodeDocumentModel.model_validate(nd)
async def valid_triple_id(triple_id: UUID, user_id: UUID = Depends(get_user_pg), session: AsyncSession = Depends(get_pg)) -> TripleDocumentModel:
tp = await get_triple(session, triple_id, user_id)
if tp is None:
raise HTTPException(status_code=404, detail="Triple not found")
return TripleDocumentModel.model_validate(tp)
async def valid_graph_id(graph_id: UUID, user_id: UUID = Depends(get_user_pg), session: AsyncSession = Depends(get_pg)) -> DetailedGraphDocumentModel:
gp = await get_graph_pg(session, graph_id, user_id, include_details=True)
if gp is None:
raise HTTPException(status_code=404, detail="Graph not found")
return DetailedGraphDocumentModel.model_validate(gp)
async def valid_public_graph_id(graph_id: UUID, session: AsyncSession = Depends(get_pg)) -> DetailedGraphDocumentModel:
gp = await get_graph_pg(session, graph_id, user_id=None, public=True, include_details=True)
if gp is None:
raise HTTPException(status_code=404, detail="Graph not found")
return DetailedGraphDocumentModel.model_validate(gp)
async def valid_query_id(query_id: UUID, user_id: UUID = Depends(get_user_pg), session: AsyncSession = Depends(get_pg)) -> QueryDocumentModel:
q = await get_query(session, query_id, user_id)
if q is None:
raise HTTPException(status_code=404, detail="Query not found")
return QueryDocumentModel.model_validate(q)
async def valid_document_id(document_id: UUID, user_id: UUID = Depends(get_user_pg), session: AsyncSession = Depends(get_pg)) -> DocumentOutWithWorkspaceDetails:
doc = await get_document_with_workspace(session, document_id, user_id)
if doc is None:
raise HTTPException(status_code=404, detail="Document not found")
return DocumentOutWithWorkspaceDetails.model_validate(doc)
async def valid_create_graph(body: CreateGraphBody, user_id: UUID = Depends(get_user_pg), session: AsyncSession = Depends(get_pg)) -> bool:
ws = await get_workspace(session, body.workspace, user_id)
if ws is None:
raise HTTPException(status_code=404, detail="Workspace not found.")
if body.schema_ is not None:
sc = await get_schema_with_workspace(session, body.schema_, user_id)
if sc is None:
raise HTTPException(status_code=404, detail="Schema not found.")
filters: Dict[str, Any] = {"name": body.name, "workspace": body.workspace}
if body.schema_:
filters["schema"] = body.schema_
exist = await get_graph_pg(session, None, user_id, filters=filters)
if exist:
raise HTTPException(status_code=409, detail="Graph already exists or is being created.")
return True

View File

@ -0,0 +1,711 @@
## API 使用与测试
### **/users**:查看/轮换 API Key
```mysql
INSERT INTO users (id, email, username, firstname, lastname, api_key, providers, active);
VALUES ('123e4567-e89b-12d3-a456-426614174000', 'test@example.com', 'testuser', 'Test', 'User', 'cOjLLn804m2nEG09qqtaShNbT732LOPq42q8PB6i', '[]', TRUE);
```
1. **读取 API Key若尚未生成会为空**
```shell
curl -s -H "x-api-key:cOjLLn804m2nEGO9qqtaShNbT732LOPq42q8PB6i" http://127.0.0.1:8000/users/api_key | jq
```
测试结果:![image-20250830190633902](image/users1.png)
2. **生成/轮换 API Key**
```shell
curl -s -X POST -H "x-api-key:cOjLLn804m2nEGO9qqtaShNbT732LOPq42q8PB6i" http://127.0.0.1:8000/users/rotate_api_key | jq
export KEY="uJWVnYlz8VAxv2XN6u2iQm7EIwJ4uTCJuUBhDA8Q"
export USER_UUID="12345678-1234-1234-1234-1234567890ab"
```
测试结果:![](image/users2.png)
### **/workspaces**:增删改查工作区
```mysql
INSERT INTO workspaces (id, name, created_by)
VALUES ('00000000-0000-0000-0000-000000000001', 'PG-demo', '12345678-1234-1234-1234-1234567890ab');
```
1. 列出workspace
```shell
# 列表
curl -s -H "x-api-key: $KEY" http://127.0.0.1:8000/workspaces | jq
```
![image-20250831163450868](image/workspaces1.png)
2. 新建一个workspace
```shell
curl -s -X POST -H "x-api-key: $KEY" -H "Content-Type: application/json" \
-d '{"name":"PG-demo","description":"demo ws","user_id":"'"$USER_UUID"'"}' \
http://127.0.0.1:8000/workspaces | jq
```
![image-20250831164705416](image/workspaces2.png)
3. 按照id查询workspace
```shell
export WS_ID="e39d8628-7d82-4894-9df9-89d475b4d93a"
curl -s -H "x-api-key: $KEY" http://127.0.0.1:8000/workspaces/$WS_ID | jq
```
![image-20250831164954767](image/workspaces3.png)
4. 更新workspace
```shell
curl -s -X PATCH -H "x-api-key: $KEY" -H "Content-Type: application/json" \
-d '{"name":"PG-demo-2"}' http://127.0.0.1:8000/workspaces/$WS_ID | jq
```
![image-20250831165611712](image/workspaces4.png)
5. 删除workspace
```shell
curl -s -X DELETE -H "x-api-key: $KEY" http://127.0.0.1:8000/workspaces/$WS_ID
```
![image-20250831171054984](image/workspaces5.png)
### **/schemas**:定义实体/关系抽取规则
1. 列出Schemas
```bash
curl -s -H "x-api-key: $KEY" "http://127.0.0.1:8000/schemas?workspace_id=$WS_ID" | jq
```
![image-20250831172155432](image/schemas1.png)
2. 新建Schemas
```shell
curl -s -X POST -H "x-api-key: $KEY" -H "Content-Type: application/json" \
"http://127.0.0.1:8000/schemas?workspace_id=$WS_ID&name=Person" \
-d '{
"entities":[{"type":"Person","properties":["name","age"]}],
"relations":[{"from":"Person","name":"knows","to":"Person"}]
}' | jq
```
![image-20250831193952222](image/schemas2.png)
3. 读取Schemas
```shell
export SCHEMA_ID="fdc3e9fc-11ee-4240-b518-6506f4fa1692"
curl -s -H "x-api-key: $KEY" "http://127.0.0.1:8000/schemas/$SCHEMA_ID" | jq
```
![image-20250831194208781](image/schemas3.png)
4. 更新Schemas名称
```shell
curl -s -X PUT -H "x-api-key: $KEY" -H "Content-Type: application/json" \
"http://127.0.0.1:8000/schemas/$SCHEMA_ID?name=PersonV2" -d '{}' | jq
```
![image-20250831194233212](image/schemas4.png)
5. 更新Schemas body
```shell
curl -s -X PUT -H "x-api-key: $KEY" -H "Content-Type: application/json" \
"http://127.0.0.1:8000/schemas/$SCHEMA_ID" \
-d '{"relations":[{"from":"Person","name":"works_at","to":"Company"}]}' | jq
```
![image-20250831194305834](image/schemas5.png)
6. 删除Schemas
```shell
curl -s -X DELETE -H "x-api-key: $KEY" "http://127.0.0.1:8000/schemas/$SCHEMA_ID" | jq
```
![image-20250831194324068](image/schemas6.png)
7. LLM 生成
```shell
curl -s -X POST -H "Content-Type: application/json" \
"http://127.0.0.1:8000/schemas/generate" \
-d '{"questions":["请为公司-员工关系设计一个图谱schema","人和人之间如何表示社交关系?"]}' | jq
```
![image-20250902161320095](image/schemas7.png)
![image-20250902161335502](image/schemas8.png)
### **/documents**:文档处理
1. 列出documents
```shell
curl -s -H "x-api-key: $KEY" "http://127.0.0.1:8000/documents?limit=10&order=-1" | jq
```
![image-20250902170737596](image/documents1.png)
2. 创建documents
```shell
curl -s -X POST -H "x-api-key: $KEY" -H "Content-Type: application/json" \
-d '{"workspace_id":"'$WS_ID'","title":"doc-1","source":"inline","meta":{"desc":"demo"}}' \
http://127.0.0.1:8000/documents | jq
```
![image-20250902170836448](image/documents2.png)
3. 更新documents
```shell
export DOC_ID="7580d3f1-2711-4cef-b60e-c25dfdb9424c"
curl -s -X POST -H "x-api-key: $KEY" -H "Content-Type: application/json" \
-d '{"status_value":"processed","errors":[]}' \
"http://127.0.0.1:8000/documents/$DOC_ID/state" | jq
```
![image-20250906212755586](image/documents3.png)
4. 绑定/解绑documents
```shell
# 需要注意的是由于前面测试了删除workspace所以前面设置的WS-ID已经不存在了
export WS_ID="00000000-0000-0000-0000-000000000001"
curl -s -X POST -H "x-api-key: $KEY" -H "Content-Type: application/json" \
"http://127.0.0.1:8000/documents/assign?workspace_id=$WS_ID" \
-d '{"document_ids":["'$DOC_ID'"]}' | jq
curl -s -X POST -H "x-api-key: $KEY" -H "Content-Type: application/json" \
"http://127.0.0.1:8000/documents/unassign?workspace_id=$WS_ID" \
-d '{"document_ids":["'$DOC_ID'"]}' | jq
```
![image-20250902172747595](image/documents4.png)
### **/chunks**:切块、嵌入
1. 列出chunks
```shell
# 全部
curl -s -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/chunks?skip=0&limit=10&populate=true" | jq
# 按 document 过滤
curl -s -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/chunks?document_id=$DOC_ID" | jq
```
![image-20250902173909800](image/chunks1.png)
![image-20250902173923213](image/chunks2.png)
2. 创建chunks
```shell
curl -s -X POST -H "x-api-key: $KEY" -H "Content-Type: application/json" \
-d '{
"chunks_in":[
{"content":"this is a text chunk","tags":["a","b"],"user_metadata":{"score":0.9}},
{"content":{"k1":"v1","k2":2},"tags":["x"],"user_metadata":{"note":"obj"}}
]
}' \
"http://127.0.0.1:8000/chunks?workspace_id=$WS_ID" | jq
```
![image-20250906222040127](image/chunks3.png)
![image-20250906222054692](image/chunks4.png)
![image-20250906222132188](image/chunks5.png)
3. 上传文件自动切块csv/json/pdf/txt
```shell
# txt
printf 'Hello\nThis is a txt chunk.\n' |
curl -s -X POST -H "x-api-key: $KEY" \
-F "file=@-;filename=sample.txt" \
"http://127.0.0.1:8000/chunks/upload?workspace_id=$WS_ID&document_id=$DOC_ID&extension=txt" | jq
```
![image-20250906213354449](image/chunks6.png)
```shell
# CSV
cat <<'CSV' |
id,name,score
1,Alice,0.9
2,Bob,0.8
CSV
curl -s -X POST -H "x-api-key: $KEY" \
-F "file=@/tmp/sample.csv;type=text/csv" \
"http://127.0.0.1:8000/chunks/upload?workspace_id=$WS_ID&document_id=$DOC_ID&extension=csv" | jq
```
![image-20250906222221525](image/chunks7.png)
![image-20250906222244618](image/chunks8.png)
```shell
# JSON
cat > /tmp/data.json <<'JSON'
[
{"id": 1, "name": "alpha", "score": 0.91},
{"id": 2, "name": "beta", "score": 0.83}
]
JSON
curl -s -X POST -H "x-api-key: $KEY" \
-F "file=@/tmp/data.json;type=application/json" \
"http://127.0.0.1:8000/chunks/upload?workspace_id=$WS_ID&document_id=$DOC_ID&extension=json" | jq
```
![image-20250906222353112](image/chunks9.png)
![image-20250906222416791](image/chunks10.png)
```shell
# PDF
base64 -d <<'B64' |
JVBERi0xLjQKMSAwIG9iago8PC9UeXBlL1BhZ2UvUGFyZW50IDIgMCBSL1Jlc291cmNlczw8Pj4vTWVkaWFCb3hbMCAw
IDU5NSA4NDJdPj4KZW5kb2JqCjIgMCBvYmoKPDwvVHlwZS9QYWdlcy9LaWRzWzEgMCBSXT4+CmVuZG9iagozIDAgb2Jq
Cjw8L1R5cGUvQ2F0YWxvZy9QYWdlcyAyIDAgUj4+CmVuZG9iagp4cmVmCjAgNAowMDAwMDAwMDAwIDY1NTM1IGYgCjAw
MDAwMDAxMDAgMDAwMDAgbiAKMDAwMDAwMDA1MCAwMDAwMCBuIAowMDAwMDAwMTUwIDAwMDAwIG4gCnRyYWlsZXIKPDwv
Um9vdCAzIDAgUi9TaXplIDQ+PgpzdGFydHhyZWYKMjgwCiUlRU9G
B64
curl -s -X POST -H "x-api-key: $KEY" \
-F "file=@-;filename=mini.pdf" \
"http://127.0.0.1:8000/chunks/upload?workspace_id=$WS_ID&document_id=$DOC_ID&extension=pdf" | jq
```
![image-20250906222457595](image/chunks11.png)
4. 绑定/解绑workspace
```shell
# 绑定
curl -s -X POST -H "x-api-key: $KEY" -H "Content-Type: application/json" \
-d '{"chunk_ids":["'"$CHUNK_ID"'"]}' \
"http://127.0.0.1:8000/chunks/assign?workspace_id=$WS_ID" | jq
# 解绑
curl -s -X POST -H "x-api-key: $KEY" -H "Content-Type: application/json" \
-d '{"chunk_ids":["'"$CHUNK_ID"'"]}' \
"http://127.0.0.1:8000/chunks/unassign?workspace_id=$WS_ID" | jq
```
![image-20250906224838400](image/chunks12.png)
![image-20250906233332717](image/chunks13.png)
5. 更新 tags / user_metadata
```shell
curl -s -X PATCH \
-H "x-api-key: $KEY" \
-H "Content-Type: application/json" \
-d '{"tags": ["a","b","c"], "user_metadata": {"score": 0.95, "note": "updated by API"}}' \
"http://127.0.0.1:8000/chunks/$CHUNK_ID?workspace_id=$WS_ID" | jq
```
![image-20250906234504910](image/chunks14.png)
![image-20250906234516506](image/chunks15.png)
6. 删除
```shell
curl -s -X DELETE -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/chunks/$CHUNK_ID" | jq
```
![image-20250906234548336](image/chunks16.png)
![image-20250906234559210](image/chunks18.png)
### **/graphs**:构建图谱
1. 用三元组创建图
```shell
export GRAPH_ID=$(uuidgen | tr 'A-Z' 'a-z')
echo $GRAPH_ID
cat <<'JSON' | \
curl -s -X POST \
-H "x-api-key: $KEY" \
-H "Content-Type: application/json" \
-d @- \
"http://127.0.0.1:8000/graphs/from_triples?graph_id=$GRAPH_ID" | jq
{
"triples": [
{
"head": "Acme Inc",
"head_type": "company",
"relation": "employs",
"tail": "Alice",
"tail_type": "employee",
"relation_properties": {"source": "demo-1"}
},
{
"head": "Alice",
"head_type": "employee",
"relation": "works_in",
"tail": "R&D",
"tail_type": "department",
"relation_properties": {"source": "demo-2"}
}
]
}
JSON
```
![image-20250907111015974](image/graphs1.png)
![image-20250907111032853](image/graphs2.png)
2. 创建图
```shell
curl -s -X POST -H "x-api-key: $KEY" -H "Content-Type: application/json" \
-d '{"name":"demo-graph-1","workspace":"'"$WS_ID"'", "schema_id": null}' \
http://127.0.0.1:8000/graphs | jq
```
![image-20250907120347721](image/graphs3.png)
2. 列出用户的图
```shell
curl -s -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/graphs?limit=20&order=-1" | jq
```
![image-20250907124429246](image/graphs4.png)
3. 读取单个图
```shell
curl -s -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/graphs/$GRAPH_ID" | jq
```
![image-20250907124507766](image/graphs5.png)
4. 列出图的节点
```shell
curl -s -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/graphs/$GRAPH_ID/nodes?created_by=$USER_UUID&skip=0&limit=100&order=-1" | jq
```
![image-20250907130958351](image/graphs6.png)
5. 列出图的关系
```shell
curl -s -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/graphs/$GRAPH_ID/relations?created_by=$USER_UUID&skip=0&limit=100&order=-1" | jq
```
![image-20250907130126206](image/graphs8.png)
6. 删除图
```shell
curl -s -X DELETE -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/graphs/$GRAPH_ID?created_by=$USER_UUID" | jq
```
![image-20250907131439647](image/graphs9.png)
### /nodes构建节点
1. 创建节点
```shell
cat <<JSON | curl -s -X POST \
-H "x-api-key: $KEY" \
-H "Content-Type: application/json" \
--data-binary @- \
"http://127.0.0.1:8000/nodes" | tee /tmp/node_create.json | jq
{
"graph": "$GRAPH_ID",
"name": "Alice",
"type": "employee",
"properties": { "email": "alice@example.com", "department": "R&D" },
"chunks": ["$CHUNK_ID"]
}
JSON
```
![image-20250909170145969](image/nodes1.png)
2. 列表
```shell
export NODE_ID="10df2f58-ec49-4ccc-8ae6-564a4faf297d"
curl -s -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/nodes?graph_id=$GRAPH_ID&skip=0&limit=50&order=-1" | jq
```
![image-20250909170757183](image/nodes2.png)
3. 读取单个节点
```shell
curl -s -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/nodes/$NODE_ID?graph_id=$GRAPH_ID" | jq
```
![image-20250909170843832](image/nodes3.png)
4. 更换节点
```shell
cat <<JSON | curl -s -X PUT \
-H "x-api-key: $KEY" \
-H "Content-Type: application/json" \
--data-binary @- \
"http://127.0.0.1:8000/nodes/$NODE_ID" | jq
{
"name": "Alice Chen",
"type": "person",
"properties": { "email": "alice.chen@example.com", "department": "R&D", "level": "Senior" },
"chunks": ["$CHUNK_ID"]
}
JSON
```
![image-20250909170944873](image/nodes4.png)
5. 删除节点
```shell
curl -s -X DELETE -H "x-api-key: $KEY" "http://127.0.0.1:8000/nodes/$NODE_ID" | jq
```
![image-20250909171603427](image/nodes5.png)
### /triples构建三元组
1. 创建三元组
```shell
cat <<JSON | curl -s -X POST \
-H "x-api-key: $KEY" \
-H "Content-Type: application/json" \
--data-binary @- \
"http://127.0.0.1:8000/triples" | tee /tmp/triple_create_1.json | jq
{
"graph_id": "$GRAPH_ID",
"subject": "Alice",
"predicate": "works_in",
"object": "R&D"
}
JSON
```
![image-20250909195418170](image/triples1.png)
2. 列出当前图的三元组
```shell
curl -s -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/triples?graph_id=$GRAPH_ID&skip=0&limit=50" \
| tee /tmp/triple_list.json | jq
```
![image-20250909195502133](image/triples2.png)
3. 更新三元组
```shell
export TRIPLE_ID="0f04fda8-6f0f-439e-b756-68d888c4a6c9"
```
- 修改谓词
```shell
curl -s -X PUT \
-H "x-api-key: $KEY" \
-H "Content-Type: application/json" \
-d '{"predicate":"reports_to"}' \
"http://127.0.0.1:8000/triples/$TRIPLE_ID" | jq
```
![image-20250909200952071](image/triples4.png)
- 合并 properties并更新 subject
```shell
curl -s -X PUT \
-H "x-api-key: $KEY" \
-H "Content-Type: application/json" \
-d '{"subject":"Alice","properties":{"reason":"org change"}}' \
"http://127.0.0.1:8000/triples/$TRIPLE_ID" | jq
```
![image-20250909201043929](image/triples5.png)
- 覆盖 properties
```shell
curl -s -X PUT \
-H "x-api-key: $KEY" \
-H "Content-Type: application/json" \
-d '{"properties":{"source":"manual-sync"}, "merge_properties": false}' \
"http://127.0.0.1:8000/triples/$TRIPLE_ID" | jq
```
![image-20250909201125814](image/triples6.png)
- 更新 head/tail 节点及 chunks
```shell
curl -s -X PUT \
-H "x-api-key: $KEY" \
-H "Content-Type: application/json" \
-d '{"head_node_id":"'"$NODE_A_ID"'","tail_node_id":"'"$NODE_B_ID"'","chunks":["'"$CHUNK_ID"'"]}' \
"http://127.0.0.1:8000/triples/$TRIPLE_ID" | jq
```
![image-20250909201206802](image/triples7.png)
4. 删除三元组
```shell
curl -i -X DELETE -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/triples/$TRIPLE_ID"
```
![image-20250909202146645](image/triples8.png)
### **/rules**:抽取规则维护
1. 创建规则workspace
```bash
cat <<'JSON' | curl -s -X POST \
-H "x-api-key: $KEY" -H "Content-Type: application/json" \
--data-binary @- \
"http://127.0.0.1:8000/rules?workspace_id=$WS_ID" | jq
{
"name": "mask-employee-email",
"if": { "field": "label", "eq": "employee" },
"then": { "action": "mask", "fields": ["email"] },
"tags": ["demo", "workspace"]
}
JSON
```
![image-20250907150555463](image/rules1.png)
2. 查询规则workspace
```shell
curl -s -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/rules?workspace_id=$WS_ID&limit=50&order=-1" | jq
```
![image-20250907150652500](image/rules2.png)
3. 创建规则graphs
```shell
# 前面测试的时候graph被删掉了重新生成了一个
export GRAPH_ID="1cbb9f63-1f48-46ac-a18f-888fb2b364c5"
cat <<'JSON' | \
curl -s -X POST \
-H "x-api-key: $KEY" \
-H "Content-Type: application/json" \
--data-binary @- \
"http://127.0.0.1:8000/rules?graph_id=$GRAPH_ID" | jq
{
"name": "deny-secret-dept",
"if": { "field": "department", "eq": "R&D" },
"then": { "action": "deny" },
"tags": ["demo", "graph"]
}
JSON
```
![image-20250907151815165](image/rules3.png)
4. 查询规则graphs
```shell
curl -s -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/rules?graph_id=$GRAPH_ID&skip=0&limit=50&order=-1" | jq
```
![image-20250907151831375](image/rules4.png)
5. 删除规则
```shell
curl -s -X DELETE -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/rules/$RULE_ID" | jq
```
![image-20250907151930745](image/rules5.png)
### **/tasks**:异步任务/执行记录
1. 创建任务
```bash
curl -s -X POST -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/tasks?created_by=$USER_UUID&title=t1&description=d1" | jq
```
![image-20250907130336357](image/tasks1.png)
2. 查询任务
```shell
export TASK_ID="19089051-05d4-465b-918d-cc8c2a06a3c0"
curl -s -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/tasks/$TASK_ID?created_by=$USER_UUID" | jq
```
![image-20250907130502491](image/task2.png)

View File

@ -0,0 +1,158 @@
## RAG实践端到端
思路:以 **workspace 过滤****/chunks 语义检索OpenAI Embedding** → **拼接上下文****调用 LLM** 为主线
### 0 健康检查
```shell
curl -s -H "x-api-key: $KEY" "http://127.0.0.1:8000/db" | jq
```
![image-20250909203206372](image/rag-健康检查.png)
### 1 创建工作区
```bash
cat <<JSON | curl -s -X POST \
-H "x-api-key: $KEY" -H "Content-Type: application/json" \
--data-binary @- "http://127.0.0.1:8000/workspaces" | jq
{
"name": "PG-demo",
"description": "demo ws"
}
JSON
```
![image-20250909203223052](image/rag-创建工作区.png)
```shell
export WS_ID="83a48b27-7ed9-4111-ad19-ca910e477653"
```
### 2 创建chunks
```bash
# 1) 组织 payload
cat >/tmp/chunks_payload.json <<'JSON'
{
"chunks_in": [
{
"content": "this is a text chunk",
"tags": ["a", "b"],
"user_metadata": { "score": 0.9 }
},
{
"content": { "k1": "v1", "k2": 2 },
"tags": ["x"],
"user_metadata": { "note": "obj" }
}
]
}
JSON
# 2) 创建(按 workspace 归属;只需 x-api-key不用传 created_by
curl -s -X POST \
-H "x-api-key: $KEY" \
-H "Content-Type: application/json" \
--data-binary @/tmp/chunks_payload.json \
"http://127.0.0.1:8000/chunks?workspace_id=$WS_ID" \
| tee /tmp/chunks_create.json | jq
export CHUNK_A=$(jq -r '.chunks[0].id // .chunks_created[0].id' /tmp/chunks_create.json)
export CHUNK_B=$(jq -r '.chunks[1].id // .chunks_created[1].id' /tmp/chunks_create.json)
echo "CHUNK_A=$CHUNK_A"
echo "CHUNK_B=$CHUNK_B"
# 3)增加中文chunk
cat >/tmp/chunks_more.json <<'JSON'
{
"chunks_in": [
{
"content": "openGauss 是企业级开源数据库,兼容 PostgreSQL具备高可用与高性能。",
"tags": ["db","opengauss"],
"user_metadata": { "lang": "zh" }
},
{
"content": "WhyHow 支持文档分块、图谱抽取与 RAG 检索问答,便于企业知识应用。",
"tags": ["whyhow","rag"],
"user_metadata": { "lang": "zh" }
}
]
}
JSON
curl -s -X POST \
-H "x-api-key: $KEY" \
-H "Content-Type: application/json" \
--data-binary @/tmp/chunks_more.json \
"http://127.0.0.1:8000/chunks?workspace_id=$WS_ID" | jq
```
![image-20250909205358200](image/rag-创建chunks.png)
检查embedding
```shell
curl -s -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/chunks?workspace_id=$WS_ID&limit=50&populate=true&include_embeddings=true" \
| jq '.chunks[] | {id, type: .data_type, dim: (.embedding|length? // 0), preview: ((.content|tostring)[0:50])}'
```
![image-20250910110446457](image/rag-创建chunks2.png)
```shell
curl -s -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/chunks?workspace_id=$WS_ID&limit=50&populate=true" \
| jq '.chunks[] | {id, has_emb: (has("embedding") and .embedding!=null), type: .data_type, preview: ((.content|tostring)[0:50])}'
```
![image-20250910110502061](image/rag-创建chunks1.png)
### 3 RAG查询
1. GET
```shell
curl -s -G -H "x-api-key: $KEY" \
--data-urlencode "workspace_id=$WS_ID" \
--data-urlencode "text=openGauss 的优势是什么?" \
--data-urlencode "top_k=5" \
"http://127.0.0.1:8000/queries/rag" \
| jq '{answer, retrieved:(.top_chunks // [])}'
```
![image-20250910114837111](image/rag-查询.png)
2. POST
```shell
jq -n --arg ws "$WS_ID" --arg txt "openGauss 的优势是什么?" --argjson top 5 \
'{workspace_id:$ws, text:$txt, top_k:$top}' > /tmp/rag_req.json
curl -s -X POST \
-H "x-api-key: $KEY" -H "Content-Type: application/json" \
--data-binary @/tmp/rag_req.json \
"http://127.0.0.1:8000/queries/rag" \
| tee /tmp/rag_resp.json \
| jq '{answer, retrieved: (.retrieval // [] | map({id, score, preview: ((.content|tostring)[0:80])}))}'
```
![image-20250910114304437](image/rag-post.png)
3. 查列表
```shell
curl -s -H "x-api-key: $KEY" \
"http://127.0.0.1:8000/queries?limit=5" | jq
```
![image-20250910115049527](image/rag-查列表1.png)
![image-20250910115100784](image/rag-查列表2.png)

Binary file not shown.

After

Width:  |  Height:  |  Size: 65 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 843 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 136 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 66 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 68 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 316 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 204 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 130 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 122 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 61 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 139 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 116 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 107 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 52 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 47 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 848 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 47 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 68 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 226 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 86 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 134 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 328 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 119 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 129 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 147 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 147 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 113 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 57 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 36 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 57 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 62 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 39 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 59 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 72 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 107 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 7.2 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 97 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 208 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 205 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 42 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 204 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 235 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 105 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 62 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 85 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 117 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 123 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 156 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 156 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 43 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 40 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 143 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 131 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 93 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 118 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 36 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 183 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 118 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 79 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 83 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 57 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 44 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 87 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 97 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 97 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 107 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 64 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 151 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 152 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 84 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 84 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 71 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 83 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 30 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 254 KiB

View File

@ -0,0 +1,236 @@
## 部署文档
### 1 软硬件依赖
- 2vCPUs | 4GiB | s7.large.2 CentOS 7.6 64bit
- Docker / Docker Compose
- Python 3.11强烈建议用conda创建虚拟环境
- openGauss 3.x容器
- OpenAI API Key用于 Embedding/LLM
### 2 openGauss 部署
1. 拉取docker镜像
```shell
docker pull enmotech/opengauss:latest
```
Tips建议服务器上面挂个代理或者本地拉取我本地有代理导入服务器
2. 创建opengauss容器并启动
```shell
docker run -d --name opengauss \
-e GS_PASSWORD='Enmo@123' \
-e GAUSSHOME=/usr/local/opengauss \
-e LD_LIBRARY_PATH=/usr/local/opengauss/lib \
-e PATH=/usr/local/opengauss/bin:$PATH \
-p 5432:5432 enmotech/opengauss:3.1.0
```
后续开机只需要启动就好
```shell
docker start opengauss
```
3. 创建表我这里用的gsql
```shell
export GAUSSHOME=/usr/local/opengauss
export LD_LIBRARY_PATH=$GAUSSHOME/lib:$LD_LIBRARY_PATH
export PATH=$GAUSSHOME/bin:$PATH
gsql -d postgres -U gaussdb -W Enmo@123
```
### 3 WhyHow部署
1. 下载安装程序
```shell
git clone https://gitcode.com/paradox/whyhow_opengauss.git
cd knowledge-graph-studio
pip install -r requirements.txt
```
2. 配置环境变量
```shell
cp .env.sample .env
WHYHOW__EMBEDDING__OPENAI__API_KEY=<你的openai api key>
WHYHOW__GENERATIVE__OPENAI__API_KEY=<你的openai api key>
WHYHOW__OPENGAUSS__HOST=<数据库的host>
WHYHOW__OPENGAUSS__PORT=<数据库docker映射出来的端口>5432
WHYHOW__OPENGAUSS__DATABASE=<数据库名称>
WHYHOW__OPENGAUSS__USER=<数据库用户名>
WHYHOW__OPENGAUSS__PASSWORD=<数据库密码>
WHYHOW__OPENGAUSS__ECHO_SQL=<是否打印 SQL 语句>
# e.g.
# WHYHOW__OPENGAUSS__HOST=127.0.0.1
# WHYHOW__OPENGAUSS__PORT=5432
# WHYHOW__OPENGAUSS__DATABASE=postgres
# WHYHOW__OPENGAUSS__USER=gaussdb
# WHYHOW__OPENGAUSS__PASSWORD=Enmo@123
# WHYHOW__OPENGAUSS__ECHO_SQL=true
```
3. 运行API服务器
```shell
uvicorn whyhow_api.main:app --host 0.0.0.0 --port 8000 --reload
```
Tips如果服务器断连记着先杀死进程再重新运行
```shell
pkill -f "uvicorn .*whyhow_api.main:app" || true
```
#### 4 初始化数据库,创建表
```mysql
-- 基础表(统一 UUID 主键)
CREATE TABLE IF NOT EXISTS users (
id UUID PRIMARY KEY,
email VARCHAR(255) UNIQUE,
username VARCHAR(255),
firstname VARCHAR(255),
lastname VARCHAR(255),
api_key VARCHAR(64) UNIQUE NOT NULL,
providers JSON,
active BOOLEAN NOT NULL DEFAULT TRUE,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE TABLE IF NOT EXISTS workspaces (
id UUID PRIMARY KEY,
name VARCHAR(128) UNIQUE NOT NULL,
description TEXT,
user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE TABLE IF NOT EXISTS graphs (
id uuid PRIMARY KEY,
schema_id uuid NULL,
workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
created_by uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE,
public boolean NOT NULL DEFAULT false,
name text NOT NULL,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS idx_graphs_created_by ON graphs(created_by);
CREATE INDEX IF NOT EXISTS idx_graphs_workspace_id ON graphs(workspace_id);
CREATE INDEX IF NOT EXISTS idx_graphs_created_at ON graphs(created_at DESC);
CREATE TABLE IF NOT EXISTS rules (
id uuid PRIMARY KEY,
workspace_id uuid NULL REFERENCES workspaces(id) ON DELETE CASCADE,
created_by uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE,
name text,
body json,
created_at timestamptz NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS schemas (
id uuid PRIMARY KEY,
workspace_id uuid NOT NULL,
created_by uuid NOT NULL,
name text,
body jsonb,
created_at timestamptz DEFAULT now()
);
CREATE TABLE IF NOT EXISTS documents (
id uuid PRIMARY KEY,
created_by uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE,
status varchar(32) NOT NULL DEFAULT 'uploaded',
metadata jsonb,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS document_workspaces (
document_id uuid NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
created_at timestamptz NOT NULL DEFAULT now(),
PRIMARY KEY (document_id, workspace_id)
);
CREATE TABLE IF NOT EXISTS chunks (
id uuid PRIMARY KEY,
document_id uuid NULL REFERENCES documents(id) ON DELETE CASCADE,
workspaces uuid[] NOT NULL, -- 参与的 workspace 列表
data_type text NOT NULL, -- 'string' / 'object'
content text,
content_obj json,
embedding json,
tags json NOT NULL DEFAULT '{}'::json,
user_metadata json NOT NULL DEFAULT '{}'::json,
metadata json NOT NULL DEFAULT '{}'::json,
created_by uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS nodes (
id uuid PRIMARY KEY,
graph_id uuid NOT NULL,
name text NOT NULL,
label text NOT NULL,
properties json NOT NULL DEFAULT '{}'::json,
chunks uuid[] NOT NULL DEFAULT '{}', -- 关键:默认空数组
created_by uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS triples (
id uuid PRIMARY KEY,
graph_id uuid NOT NULL,
head_node_id uuid,
tail_node_id uuid,
relation_name text NOT NULL,
properties json NOT NULL DEFAULT '{}'::json,
chunks uuid[] NOT NULL DEFAULT '{}', -- 关键:默认空数组
created_by uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE,
embedding json,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS tasks (
id CHAR(36) NOT NULL,
user_id UUID NOT NULL,
title VARCHAR(255) NOT NULL,
description TEXT,
status VARCHAR(50) NOT NULL DEFAULT 'pending',
created_at TIMESTAMPTZ DEFAULT now(),
updated_at TIMESTAMPTZ DEFAULT now(),
CONSTRAINT tasks_pkey PRIMARY KEY (id),
CONSTRAINT tasks_user_id_fkey
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
)
WITH (orientation = row, compression = no);
CREATE TABLE IF NOT EXISTS queries (
id uuid PRIMARY KEY,
user_id uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE,
graph_id uuid,
status varchar(32) NOT NULL DEFAULT 'pending',
name text,
payload json,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
```

View File

@ -0,0 +1,87 @@
## 项目结构介绍
### 整体架构
**WhyHow (openGauss)** 是一个面向企业知识检索与问答的 **RAG 服务**。它把非结构化文本切分为chunks对其向量化后存入openGauss在查询阶段按工作空间范围进行语义检索将命中的片段拼接成上下文并调用 LLM 生成答案。系统采用“**入口 → 中间件 → 路由 → 业务服务 → 基础设施/数据库**”的分层架构职责单一、依赖清晰便于扩展到图谱graphs/nodes/triples和规则rules等高级能力。
**openGauss** 作为持久层统一保存业务数据与向量,以事务一致性和多租户隔离为核心;向量相似度在应用层计算。
<img src="image/框架图.png" alt="框架图" style="zoom: 33%;" />
#### 1) 应用入口FastAPI
- **职责**:启动与挂载、统一配置加载、路由注册与生命周期管理。
- **输入/输出**:接收 HTTP 请求,输出统一的 JSON 响应与 OpenAPI 文档。
#### 2) 中间件和依赖
- **内容**:认证(`x-api-key`)、请求/响应日志、异常收敛、CORS、限流、跨请求上下文依赖注入。
- **特点**:请求进入路由前即生效,确保所有下游层拿到一致的上下文与安全基线。
#### 3) RoutersHTTP 接口)
- **职责**:把 REST 入口映射到 Service只做**轻校验与参数解析**,不落具体业务。
- **成员**`users/workspaces/chunks/documents/graphs/nodes/triples/queries/tasks/rules/schemas`。
- **约定**:统一分页/排序参数、统一鉴权头(`x-api-key`)、统一错误格式。
#### 4) SchemasPydantic
- **职责**:请求/响应 DTO、字段校验、公共枚举与错误模型、分页与过滤器定义为 OpenAPI 生成提供强类型描述。
- **价值**:将“接口契约”前置,解耦路由/服务;避免隐式 JSON 结构造成的兼容风险。
#### 5) Services / CRUD业务编排
- **职责**
- 事务管理、过滤/分页/排序封装,统一从依赖注入获取 `db/session/config/user`
- **RAG 主线**`chunks` 写入与向量化(创建时或批量重建)→ 检索 Top-K → 拼接上下文 → 调用 LLM → 记录 `queries`
- 多租户隔离;读写审计。
#### 6) Models / Common
- **职责**:通用数据结构、类型别名、常量与工具函数;供 Schemas/Services 复用。
- **定位**:放领域无关的共享部件,避免循环依赖与重复实现。
#### 7) 基础设施和工具
- **内容**:配置管理、日志与追踪、数据库连接池/会话、构建器(切分/解析/预处理、校验器、导出工具、CLI、静态模板。
- **作用**:给上层提供统一的可复用能力。
#### 8) openGaussPostgreSQL 方言兼容)
- **职责**单一事实来源持久化业务数据与嵌入向量JSON/数组),保障事务一致;以 `workspace` 维度实现多租户隔离。
- **检索策略**:当前在应用层做余弦相似度与 Top-K。
- **表面向对象**`users/workspaces/documents/chunks/queries`;图谱 `graphs/nodes/triples` 为可选增强。
### 数据模型与表结构
#### 实体介绍
1. **User**:调用方/租户的身份与配额。
2. **Workspace**:数据隔离与分享的基本单元。
3. **Document**:原始载体(文件、网页、表格等),用于聚合 chunks 的来源。
4. **Chunk****RAG 的原子单元**。包含文本/对象内容、轻量标签与embedding 向量。一个 chunk 可被多个 workspace 共享。
5. **Query**:一次 RAG 请求的完整记录:查询文本、使用的 workspace、命中片段概要、答案、耗时等。
6. **Graph / Node / Triple**结构化知识扩展。Chunk 可在 `metadata` 中挂接 Node/Triple 的引用,实现“语义 + 结构化”的混合检索。
7. **Task / Rule**:离线重建、清洗、批处理与规则检查等辅助能力。
#### RAG 主流程时序图
1. /chunks 接口的请求处理时序图
客户端向 `/chunks` 提交内容与 `workspace_id`,路由做鉴权与参数校验后把请求交给服务层。服务层先在 openGauss 插入一条 **pending** 的 chunk随后调用嵌入服务获得向量回写到该记录并将状态标记为 **ready**,最后把 `chunk_id` 返回给客户端201 Created。这一流程确保写入具备事务一致性与多租户隔离。![image-20250916203830642](image/rag1.png)
2. RAG查询时序图
客户端向 `/queries/rag` 提交 `workspace_id + text + top_k`,路由校验后由服务层执行检索。服务层先对问题做向量化,随后在 openGauss筛出候选在应用层计算余弦相似度取 **Top-K**,将命中片段拼成上下文并调用 LLM 生成答案。全过程(请求、命中、答案、耗时)会写入 `queries` 便于审计,最终返回 `{answer, top_chunks}`200 OK
![image-20250916204031346](image/rag2.png)

View File

@ -0,0 +1,9 @@
"""Custom exceptions for the WhyHow API."""
class NotFoundException(Exception):
"""Exception raised when an item is not found."""
def __init__(self, message: str):
"""Initialise the NotFoundException."""
super().__init__(message)

117
whyhow_ospp/main.py Normal file
View File

@ -0,0 +1,117 @@
"""Main entrypoint (PG-only)."""
import logging
from contextlib import asynccontextmanager
from logging import basicConfig
from pathlib import Path
from typing import Annotated, Any
import logfire
from asgi_correlation_id import CorrelationIdMiddleware
from asgi_correlation_id.context import correlation_id
from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.exception_handlers import http_exception_handler
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import Response
from fastapi.staticfiles import StaticFiles
import sqlalchemy as sa
from sqlalchemy.ext.asyncio import AsyncSession
from whyhow_api import __version__
from whyhow_api.config import Settings
from whyhow_api.custom_logging import configure_logging
from whyhow_api.middleware import RateLimiter
from whyhow_api.routers import (
chunks,
documents,
graphs,
nodes,
queries,
rules,
schemas,
tasks,
triples,
users,
workspaces,
)
from whyhow_api.database import connect_to_pg, close_pg
from whyhow_api.dependencies import get_settings, get_pg
logger = logging.getLogger("whyhow_api.main")
@asynccontextmanager
async def lifespan(app: FastAPI): # type: ignore
settings = app.dependency_overrides.get(get_settings, get_settings)()
configure_logging(project_log_level=settings.dev.log_level)
basicConfig(handlers=[logfire.LogfireLoggingHandler()])
await connect_to_pg(settings)
logger.info("Connected to openGauss/Postgres.")
try:
yield
finally:
await close_pg()
settings_ = get_settings()
logfire_token = (
None if settings_.logfire.token is None else settings_.logfire.token.get_secret_value()
)
logfire.configure(token=logfire_token, send_to_logfire="if-token-present", console=False)
app = FastAPI(
title="WhyHow API",
summary="RAG with knowledge graphs",
version=__version__,
lifespan=lifespan,
openapi_url=get_settings().dev.openapi_url,
)
logfire.instrument_fastapi(app)
@app.exception_handler(Exception)
async def unhandled_exception_handler(request: Request, exc: Exception) -> Response:
return await http_exception_handler(
request,
HTTPException(
500,
"Internal Server Error",
headers={"X-Request-ID": correlation_id.get() or ""},
),
)
app.add_middleware(RateLimiter)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], allow_credentials=True,
allow_methods=["*"], allow_headers=["*"],
expose_headers=["X-Request-ID"],
)
app.add_middleware(CorrelationIdMiddleware)
app.mount("/static", StaticFiles(directory=Path(__file__).resolve().parent / "static"), name="static")
app.include_router(workspaces.router)
app.include_router(schemas.router)
app.include_router(graphs.router)
app.include_router(triples.router)
app.include_router(nodes.router)
app.include_router(documents.router)
app.include_router(chunks.router)
app.include_router(users.router)
app.include_router(queries.router)
app.include_router(rules.router)
app.include_router(tasks.router)
@app.get("/")
def root() -> str:
return f"Welcome to version {__version__} of the WhyHow API."
@app.get("/db")
async def database(session: AsyncSession = Depends(get_pg)) -> str:
await session.execute(sa.text("SELECT 1"))
return "Connected to openGauss (pg)."
@app.get("/settings")
def settings(settings: Annotated[Settings, Depends(get_settings)]) -> Any:
return settings

164
whyhow_ospp/middleware.py Normal file
View File

@ -0,0 +1,164 @@
"""Middleware for FastAPI."""
import collections
import logging
from datetime import datetime, timezone
###########################
# Auth0 used for UI #######
###########################
# import jwt
from fastapi import HTTPException, Request, status
from fastapi.responses import JSONResponse
# from fastapi.security import OAuth2AuthorizationCodeBearer
from starlette.middleware.base import (
BaseHTTPMiddleware,
RequestResponseEndpoint,
)
from starlette.responses import Response
from whyhow_api.config import Settings
from whyhow_api.dependencies import get_settings
from whyhow_api.utilities.routers import clean_url
logger = logging.getLogger(__name__)
# Initialize token_buckets with a lambda that sets initial tokens based on settings
def create_initial_bucket() -> dict[str, float]:
"""Create an initial token bucket."""
settings: Settings = get_settings()
return {
"last_check": datetime.now(timezone.utc).timestamp(),
"tokens": settings.api.bucket_capacity, # Set initial tokens to the bucket capacity
}
# This will store the token buckets for each user
token_buckets: dict[str, dict[str, float]] = collections.defaultdict(
create_initial_bucket
)
class RateLimiter(BaseHTTPMiddleware):
"""Token bucket rate limiter middleware for FastAPI."""
async def dispatch(
self, request: Request, call_next: RequestResponseEndpoint
) -> Response:
"""Dispatch the request."""
settings: Settings = get_settings()
path_pattern = clean_url(request.url.path)
excluded_path = (
path_pattern
in settings.api.excluded_paths + settings.api.public_paths
) # Exclude paths from rate limiting (TODO: public paths should be rate limited on IP.)
if excluded_path:
return await call_next(request)
user_key = await self.get_rate_limit_key(request, settings)
rate = settings.api.limit_frequency_value # Tokens added per second
capacity = settings.api.bucket_capacity
now = datetime.now(timezone.utc).timestamp()
bucket = token_buckets[user_key]
time_passed = max(
0, now - bucket["last_check"]
) # Ensure non-negative time passed
bucket["last_check"] = now
# Add tokens to the bucket based on elapsed time
tokens_to_add = time_passed * rate # Tokens are added per second
bucket["tokens"] = min(capacity, bucket["tokens"] + tokens_to_add)
# Logging state before processing the request
# logger.info(
# f"Processing request from {user_key}. Available tokens before request: {bucket['tokens']}. Tokens to add: {tokens_to_add}. Time passed: {time_passed}s."
# )
response: Response
if bucket["tokens"] < 1:
# logger.warning(
# f"Rate limit exceeded for user {user_key}. No tokens available."
# )
response = JSONResponse(
content={"error": "Rate limit exceeded"}, status_code=429
)
else:
bucket["tokens"] -= 1
response = await call_next(request)
# logger.info(
# f"Token deducted for user {user_key}. Tokens remaining: {bucket['tokens']}."
# )
# Update response headers for client info
response.headers["X-RateLimit-Limit"] = str(capacity)
response.headers["X-RateLimit-Remaining"] = str(
max(0, int(bucket["tokens"]))
)
response.headers["X-RateLimit-Reset"] = str(
int(now + (1 - bucket["tokens"]) * (1 / rate))
)
return response
@staticmethod
async def get_rate_limit_key(request: Request, settings: Settings) -> str:
"""Get the rate limit key."""
try:
api_key = request.headers.get("x-api-key")
###########################
# Auth0 used for UI #######
###########################
# oauth2_scheme = OAuth2AuthorizationCodeBearer(
# authorizationUrl=settings.api.auth0.authorize_url,
# tokenUrl=settings.api.auth0.token_url,
# auto_error=False,
# )
# token = await oauth2_scheme(request)
if api_key:
return api_key
# elif token:
# if (
# settings.api.auth0.domain is None
# or settings.api.auth0.audience is None
# or settings.api.auth0.algorithm is None
# ):
# raise ValueError(
# "Auth0 domain, audience, and algorithm required"
# )
# domain = settings.api.auth0.domain.get_secret_value()
# audience = settings.api.auth0.audience.get_secret_value()
# algorithm = settings.api.auth0.algorithm
# signing_key = (
# request.app.state.jwks_client.get_signing_key_from_jwt(
# token
# ).key
# )
# payload = jwt.decode(
# token,
# signing_key,
# algorithms=[algorithm],
# audience=audience,
# issuer=f"https://{domain}/",
# )
# return payload["sub"]
else:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="API key or token is required",
)
except HTTPException as http_e:
raise http_e
except Exception as e:
logger.error(
f"Error getting rate limit key (unable to authorize): {e}"
)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Unable to authorize",
)

View File

@ -0,0 +1 @@
"""Models."""

View File

@ -0,0 +1,194 @@
# whyhow_api/models/common.py
"""Shared models."""
from typing import Any, Dict, List, Union
from openai import AsyncAzureOpenAI, AsyncOpenAI
from pydantic import BaseModel, Field, ConfigDict
from whyhow_api.schemas.users import (
BYOAzureOpenAIMetadata,
BYOOpenAIMetadata,
WhyHowOpenAIMetadata,
)
from whyhow_api.config import Settings
settings = Settings()
class LLMClient:
"""LLM client."""
def __init__(
self,
client: AsyncOpenAI | AsyncAzureOpenAI,
metadata: Union[
BYOOpenAIMetadata, BYOAzureOpenAIMetadata, WhyHowOpenAIMetadata
],
) -> None:
"""Initialize the LLM client."""
self.client = client
self.metadata = metadata
class Node(BaseModel):
"""Schema for a single node."""
name: str = Field(..., description="The name of the node.", examples=["Python"])
label: str | None = Field(
None,
description="The label (e.g., person, organization, location).",
examples=["Programming Language"],
)
properties: dict[str, Any] = Field(default_factory=dict, description="Properties of the node.")
class Relation(BaseModel):
"""Schema for a single relationship."""
label: str
start_node: Node
end_node: Node
properties: dict[str, Any] = Field(default_factory=dict)
class Entity(BaseModel):
"""Schema for a single entity (text required)."""
text: str = Field(..., examples=["Python"])
label: str | None = None
properties: dict[str, Any] = Field(default_factory=dict)
class Triple(BaseModel):
"""Schema for a single triple."""
head: str = Field(..., min_length=1)
head_type: str = Field(default="Entity", min_length=1)
relation: str = Field(..., min_length=1)
tail: str = Field(..., min_length=1)
tail_type: str = Field(default="Entity", min_length=1)
head_properties: Dict[Any, Any] = Field(default={})
relation_properties: Dict[Any, Any] = Field(default={})
tail_properties: Dict[Any, Any] = Field(default={})
def __str__(self) -> str:
return self.model_dump_json(indent=2)
class EntityField(BaseModel):
name: str
properties: List[str] = []
class SchemaEntity(BaseModel):
name: str
description: str
fields: List[EntityField] = Field(default=[])
class SchemaRelation(BaseModel):
name: str
description: str
class TriplePattern(BaseModel):
head: str
relation: str
tail: str
description: str
class GeneratedSchema(BaseModel):
entities: List[SchemaEntity] = []
relations: List[SchemaRelation] = []
patterns: List[TriplePattern] = []
model_config = ConfigDict(extra="ignore")
class SchemaTriplePattern(BaseModel):
head: SchemaEntity
relation: SchemaRelation
tail: SchemaEntity
description: str
class StructuredSchemaEntity(BaseModel):
name: str
field: EntityField
properties: List[str] = Field(default_factory=list)
class StructuredSchemaTriplePattern(BaseModel):
head: StructuredSchemaEntity
relation: str
tail: StructuredSchemaEntity
class Schema(BaseModel):
entities: List[SchemaEntity] = Field(default_factory=list)
relations: List[SchemaRelation] = Field(default_factory=list)
patterns: List[Union[SchemaTriplePattern, TriplePattern]] = Field(default_factory=list)
def get_entity(self, name: str) -> SchemaEntity | None:
return next((e for e in self.entities if e.name == name), None)
def get_relation(self, name: str) -> SchemaRelation | None:
return next((r for r in self.relations if r.name == name), None)
class OpenAICompletionsConfig(BaseModel):
"""OpenAI completions configuration."""
model: str = Field(default="gpt-4o")
temperature: float = Field(default=0.1)
max_tokens: int = Field(default=2000)
class OpenAIDirectivesConfig(BaseModel):
"""OpenAI directives configuration."""
entity_questions: str = ""
triple_questions: str = ""
entity_concepts: str = ""
triple_concepts: str = ""
merge_graph: str = ""
specific_query: str = ""
improve_matched_relations: str = ""
improve_matched_entities: str = ""
class MasterOpenAICompletionsConfig(BaseModel):
"""Master OpenAI completions configuration."""
default: OpenAICompletionsConfig
entity: OpenAICompletionsConfig
triple: OpenAICompletionsConfig
entity_questions: OpenAICompletionsConfig
triple_questions: OpenAICompletionsConfig
entity_concepts: OpenAICompletionsConfig
triple_concepts: OpenAICompletionsConfig
merge_graph: OpenAICompletionsConfig
class DatasetModel(BaseModel):
"""Dataset model."""
dataset: Union[Dict[str, List[str]], List[str]]
class PDFProcessorConfig(BaseModel):
"""PDF processor configuration."""
file_path: str = Field(..., description="File path to PDF document")
chunk_size: int = Field(default=settings.api.max_chars_per_chunk)
chunk_overlap: int = Field(0)
class TextWithEntities(BaseModel):
"""Text with extracted entities."""
text: str = Field(
...,
description=(
"The original text that was analyzed to identify entities. This"
" text contains one or more entities that have been classified and"
" extracted."
),
)
entities: List[Entity] = Field(
...,
description=(
"A list of entities extracted from the original text. Each entity"
" is represented as a combination of the entity's surface form and"
" its classification label."
),
)

View File

@ -0,0 +1,155 @@
aiohappyeyeballs==2.6.1
aiohttp==3.12.13
aiosignal==1.4.0
annotated-types==0.7.0
anyio==4.9.0
asgi-correlation-id==4.3.4
asgiref==3.9.1
asttokens==3.0.0
asyncpg==0.30.0
attrs==25.3.0
auth0-python==4.7.1
blis==1.3.0
boto3==1.39.3
botocore==1.39.3
catalogue==2.0.10
certifi==2025.7.9
cffi==1.17.1
charset-normalizer==3.4.2
click==8.2.1
cloudpathlib==0.21.1
confection==0.1.5
cryptography==42.0.8
cymem==2.0.11
dataclasses-json==0.6.7
DateTime==5.5
decorator==5.2.1
distro==1.9.0
dnspython==2.7.0
executing==2.2.0
fastapi==0.110.3
frozenlist==1.7.0
googleapis-common-protos==1.70.0
greenlet==3.2.3
h11==0.16.0
httpcore==1.0.9
httpx==0.28.1
httpx-sse==0.4.1
idna==3.10
importlib_metadata==8.7.0
iniconfig==2.1.0
ipython==9.4.0
ipython_pygments_lexers==1.1.1
jedi==0.19.2
Jinja2==3.1.6
jiter==0.10.0
jmespath==1.0.1
jsonpatch==1.33
jsonpointer==3.0.0
langchain==0.3.26
langchain-community==0.3.27
langchain-core==0.3.68
langchain-openai==0.3.27
langchain-text-splitters==0.3.8
langcodes==3.5.0
langsmith==0.4.4
language_data==1.3.0
logfire==4.2.0
marisa-trie==1.2.1
markdown-it-py==3.0.0
MarkupSafe==3.0.2
marshmallow==3.26.1
matplotlib-inline==0.1.7
mdurl==0.1.2
motor==3.4.0
multidict==6.6.3
murmurhash==1.0.13
mypy_extensions==1.1.0
neo4j==5.28.1
numpy @ file:///home/conda/feedstock_root/build_artifacts/numpy_1751342119755/work/dist/numpy-2.3.1-cp311-cp311-linux_x86_64.whl#sha256=a37f9653bc04b775179c760731e9ebaa9a9df01a73604f60d6391300f21db5d4
openai==1.93.2
opentelemetry-api==1.34.1
opentelemetry-exporter-otlp-proto-common==1.34.1
opentelemetry-exporter-otlp-proto-http==1.34.1
opentelemetry-instrumentation==0.55b1
opentelemetry-instrumentation-asgi==0.55b1
opentelemetry-instrumentation-fastapi==0.55b1
opentelemetry-instrumentation-pymongo==0.55b1
opentelemetry-instrumentation-system-metrics==0.55b1
opentelemetry-proto==1.34.1
opentelemetry-sdk==1.34.1
opentelemetry-semantic-conventions==0.55b1
opentelemetry-util-http==0.55b1
orjson==3.10.18
packaging==24.2
pandas==2.3.1
parso==0.8.4
pexpect==4.9.0
pinecone-client==6.0.0
pinecone-plugin-interface==0.0.7
pluggy==1.6.0
preshed==3.0.10
prompt_toolkit==3.0.51
propcache==0.3.2
protobuf==5.29.5
psutil==7.0.0
ptyprocess==0.7.0
pure_eval==0.2.3
pycparser==2.22
pydantic==2.11.7
pydantic-settings==2.10.1
pydantic_core==2.33.2
Pygments==2.19.2
PyJWT==2.10.1
pymongo==4.8.0
pypdf==5.7.0
PyPDF2==3.0.1
PySocks==1.7.1
pytest==8.4.1
pytest-mock==3.14.1
python-dateutil==2.9.0.post0
python-dotenv==1.1.1
python-multipart==0.0.20
pytz==2025.2
PyYAML==6.0.2
regex==2024.11.6
requests==2.32.4
requests-toolbelt==1.0.0
rich==14.0.0
s3transfer==0.13.0
shellingham==1.5.4
six==1.17.0
smart_open==7.3.0.post1
sniffio==1.3.1
socksio==1.0.0
spacy==3.8.7
spacy-legacy==3.0.12
spacy-loggers==1.0.5
SQLAlchemy==2.0.41
srsly==2.5.1
stack-data==0.6.3
starlette==0.37.2
tenacity==9.1.2
thinc==8.3.6
tiktoken==0.7.0
tqdm==4.67.1
traitlets==5.14.3
typer==0.16.0
types-requests==2.32.4.20250611
typing-inspect==0.9.0
typing-inspection==0.4.1
typing_extensions==4.14.1
tzdata==2025.2
urllib3==2.5.0
uuid==1.30
uvicorn==0.35.0
wasabi==1.1.3
wcwidth==0.2.13
weasel==0.4.1
whyhow==0.1.12
whyhow-api @ file:///root/knowledge-graph-studio
wrapt==1.17.2
yarl==1.20.1
zipp==3.23.0
zope.interface==7.2
zstandard==0.23.0

View File

@ -0,0 +1 @@
"""Definition of routers for the API."""

View File

@ -0,0 +1,252 @@
"""Chunks router (openGauss/PostgreSQL, API-key auth)."""
from __future__ import annotations
import logging
from typing import Any, Dict, List, Optional
from uuid import UUID
from fastapi import (
APIRouter,
Depends,
HTTPException,
Query,
Body,
Header,
UploadFile,
File,
)
from sqlalchemy.ext.asyncio import AsyncSession
from whyhow_api.dependencies import get_pg, get_llm_client, LLMClient
from whyhow_api.services.crud.user_pg import get_user_by_api_key
# ---- schemas (Pydantic) ----
from whyhow_api.schemas.chunks import (
AddChunkModel,
UpdateChunkModel,
ChunkOut,
ChunksOutWithWorkspaceDetails,
ChunkDocumentModel,
)
from whyhow_api.schemas.base import File_Extensions
# ---- services (you provided in services/crud/chunks_pg.py) ----
from whyhow_api.services.crud.chunks_pg import (
get_chunks,
get_chunk_basic,
prepare_chunks,
add_chunks,
process_chunks,
assign_chunks_to_workspace,
unassign_chunks_from_workspace,
update_chunk as svc_update_chunk,
delete_chunk as svc_delete_chunk,
)
logger = logging.getLogger(__name__)
router = APIRouter(tags=["Chunks"], prefix="/chunks")
# ---- common auth ----
async def _require_user_id(session: AsyncSession, api_key: str) -> UUID:
u = await get_user_by_api_key(session, api_key)
if not u:
raise HTTPException(status_code=401, detail="Invalid API key")
return u["id"]
# ==============================
# READ
# ==============================
@router.get("")
async def list_chunks(
workspace_id: Optional[UUID] = Query(None, description="按 workspace 过滤"),
data_type: Optional[str] = Query(None, description="string|object 等"),
document_id: Optional[UUID] = Query(None, description="按 document 过滤"),
chunk_id: Optional[UUID] = Query(None, alias="_id", description="按 chunk id 过滤"),
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=-1, le=200),
order: int = Query(1, description="1 升序, -1 降序"),
populate: bool = Query(True, description="是否展开 workspace/document 信息"),
include_embeddings: bool = Query(True, description="返回是否包含 embedding通常 False"),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
filters: Dict[str, Any] = {}
if workspace_id:
filters["workspaces"] = workspace_id
if data_type:
filters["data_type"] = data_type
if document_id:
filters["document_id"] = document_id
if chunk_id:
filters["_id"] = chunk_id
rows = await get_chunks(
session=session,
user_id=user_id,
llm_client=None,
include_embeddings=include_embeddings,
filters=filters,
skip=skip,
limit=limit,
order=order,
populate=populate,
)
return {
"message": "ok",
"status": "success",
"count": len(rows or []),
"chunks": rows or [],
}
@router.get("/{chunk_id}")
async def get_chunk(
chunk_id: UUID,
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
row = await get_chunk_basic(session, chunk_id=chunk_id, user_id=user_id)
if not row:
raise HTTPException(status_code=404, detail="Chunk not found")
return {"message": "ok", "status": "success", "chunk": row}
# ==============================
# CREATE
# ==============================
@router.post("")
async def add_chunks_manual(
workspace_id: UUID = Query(..., description="目标 workspace"),
chunks_in: List[AddChunkModel] = Body(..., embed=True),
session: AsyncSession = Depends(get_pg),
llm_client: LLMClient = Depends(get_llm_client),
api_key: str = Header(..., alias="x-api-key"),
):
"""
手动添加字符串/对象Chunks
- 先用 prepare_chunks 组装为 ChunkDocumentModel tags/user_metadata workspace 维度存储
- 再调用 add_chunks 计算 embedding 并入库
"""
user_id = await _require_user_id(session, api_key)
prepared = prepare_chunks(chunks_in, workspace_id, user_id)
inserted = await add_chunks(session, llm_client, prepared)
return {"message": "created", "status": "success", "count": len(inserted), "chunks": inserted}
@router.post("/upload")
async def upload_file_to_chunks(
workspace_id: UUID = Query(...),
document_id: UUID = Query(...),
extension: File_Extensions = Query(..., description="csv|json|pdf|txt"),
file: UploadFile = File(...),
session: AsyncSession = Depends(get_pg),
llm_client: LLMClient = Depends(get_llm_client),
api_key: str = Header(..., alias="x-api-key"),
):
"""
上传文件并自动切块csv/json/pdf/txt
内部会根据扩展名走结构化或非结构化处理然后 add_chunks 写入
"""
_ = await _require_user_id(session, api_key)
content = await file.read()
await process_chunks(
session=session,
content=content,
document_id=document_id,
llm_client=llm_client,
workspace_id=workspace_id,
user_id=_, # 当前用户
extension=extension,
)
return {"message": "uploaded", "status": "success"}
# ==============================
# ASSIGN / UNASSIGN
# ==============================
@router.post("/assign")
async def assign_chunks(
workspace_id: UUID = Query(...),
chunk_ids: List[UUID] = Body(..., embed=True),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
result = await assign_chunks_to_workspace(session, chunk_ids, workspace_id, user_id)
return {"message": "ok", "status": "success", "result": result}
@router.post("/unassign")
async def unassign_chunks(
workspace_id: UUID = Query(...),
chunk_ids: List[UUID] = Body(..., embed=True),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
result = await unassign_chunks_from_workspace(session, chunk_ids, workspace_id, user_id)
return {"message": "ok", "status": "success", "result": result}
# ==============================
# UPDATE / DELETE
# ==============================
@router.patch("/{chunk_id}")
async def update_chunk(
chunk_id: UUID,
workspace_id: UUID = Query(..., description="要写入 tags/user_metadata 的 workspace"),
body: UpdateChunkModel = Body(...),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
msg, rows = await svc_update_chunk(session, chunk_id, workspace_id, body, user_id)
return {"message": msg, "status": "success", "count": len(rows or []), "chunks": rows or []}
@router.delete("/{chunk_id}")
async def delete_chunk(
chunk_id: UUID,
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
row = await svc_delete_chunk(session, chunk_id, user_id)
if not row:
raise HTTPException(status_code=404, detail="Chunk not found")
return {"message": "deleted", "status": "success", "chunk": row}
@router.post("/embeddings/rebuild")
async def rebuild_embeddings_endpoint(
workspace_id: UUID | None = Query(None),
chunk_ids: str | None = Query(None),
force: bool = Query(False),
session: AsyncSession = Depends(get_pg),
llm_client: LLMClient = Depends(get_llm_client),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
ids = [UUID(s.strip()) for s in (chunk_ids or "").split(",") if s.strip()] or None
from whyhow_api.services.crud.chunks_pg import rebuild_chunk_embeddings
result = await rebuild_chunk_embeddings(
session,
user_id=user_id,
llm_client=llm_client,
workspace_id=workspace_id,
chunk_ids=ids,
force=force,
)
return {"message": "embeddings rebuilt", "status": "success", **result}

View File

@ -0,0 +1,150 @@
"""Documents router (openGauss/PostgreSQL)."""
from uuid import UUID
from typing import Any, Dict, List, Optional
import logging
import sqlalchemy as sa
from fastapi import APIRouter, Depends, HTTPException, Query, Body, Header, status
from sqlalchemy.ext.asyncio import AsyncSession
from whyhow_api.dependencies import get_pg
from whyhow_api.services.crud.document_pg import (
list_documents,
get_document,
create_document,
update_document_state_errors,
assign_documents_to_workspace,
unassign_documents_from_workspace,
)
from whyhow_api.services.crud.user_pg import get_user_by_api_key
logger = logging.getLogger(__name__)
router = APIRouter(tags=["Documents"], prefix="/documents")
async def _require_user_id(session: AsyncSession, api_key: str) -> UUID:
u = await get_user_by_api_key(session, api_key)
if not u:
raise HTTPException(status_code=401, detail="Invalid API key")
return u["id"]
@router.get("")
async def list_docs(
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=-1, le=50),
order: int = Query(1, description="1 升序, -1 降序"),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
docs = await list_documents(session, user_id, skip=skip, limit=limit, order=order)
return {
"message": "ok",
"status": "success",
"count": len(docs or []),
"documents": docs or [],
}
@router.post("")
async def create_doc(
body: Dict[str, Any] = Body(
...,
example={
"workspace_id": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
"title": "doc-1",
"source": "inline",
"meta": {},
},
),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
title: Optional[str] = body.get("title")
source: Optional[str] = body.get("source")
meta: Dict[str, Any] = body.get("meta") or {}
ws_id_raw: Optional[str] = body.get("workspace_id")
if not title or not source:
raise HTTPException(status_code=422, detail="title 和 source 不能为空")
metadata = {"title": title, "source": source, **meta}
doc = await create_document(
session=session,
user_id=user_id,
status="uploaded",
metadata=metadata,
workspace_id=UUID(ws_id_raw) if ws_id_raw else None,
)
if ws_id_raw and doc:
try:
await assign_documents_to_workspace(
session=session,
document_ids=[doc["id"]],
workspace_id=UUID(ws_id_raw),
created_by=user_id,
)
except Exception:
logger.exception("assign_documents_to_workspace failed")
return {"message": "created", "status": "success", "document": doc}
@router.get("/{document_id}")
async def get_doc(
document_id: UUID,
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
doc = await get_document(session, user_id, document_id)
if not doc:
raise HTTPException(status_code=404, detail="Document not found")
return {"message": "ok", "status": "success", "document": doc}
@router.post("/{document_id}/state")
async def update_state(
document_id: UUID,
status_value: Optional[str] = Body(None, embed=True),
errors: Optional[List[Dict[str, Any]]] = Body(None, embed=True),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
doc = await update_document_state_errors(
session, user_id, document_id, status_value, errors
)
if not doc:
raise HTTPException(status_code=404, detail="Document not found")
return {"message": "ok", "status": "success", "document": doc}
@router.post("/assign")
async def assign_docs(
workspace_id: UUID = Query(...),
document_ids: List[UUID] = Body(..., embed=True),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
res = await assign_documents_to_workspace(session, document_ids, workspace_id, user_id)
return {"message": "ok", "status": "success", "documents": res}
@router.post("/unassign")
async def unassign_docs(
workspace_id: UUID = Query(...),
document_ids: List[UUID] = Body(..., embed=True),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
res = await unassign_documents_from_workspace(session, document_ids, workspace_id, user_id)
return {"message": "ok", "status": "success", "documents": res}

View File

@ -0,0 +1,143 @@
# routers/graphs.py
import logging
from typing import List
from uuid import UUID, uuid4
from fastapi import APIRouter, Depends, HTTPException, Query, Body, Header, status
from sqlalchemy.ext.asyncio import AsyncSession
import sqlalchemy as sa
from sqlalchemy.exc import IntegrityError
from whyhow_api.dependencies import get_pg, get_llm_client
from whyhow_api.models.common import LLMClient
from whyhow_api.schemas.graphs import Triple
from whyhow_api.services.crud.graph_pg import (
list_all_graphs, get_graph, list_nodes, list_relations, delete_graphs, graphs, workspaces
)
from whyhow_api.services.graph_service_pg import build_graph_pg
from whyhow_api.services.crud.user_pg import get_user_by_api_key
from whyhow_api.services.crud.base_pg import insert_returning
from whyhow_api.schemas.graphs import CreateGraphBody
logger = logging.getLogger(__name__)
router = APIRouter(tags=["Graphs"], prefix="/graphs")
async def _require_user_id(session: AsyncSession, api_key: str) -> UUID:
u = await get_user_by_api_key(session, api_key)
if not u:
raise HTTPException(status_code=401, detail="Invalid API key")
return u["id"]
@router.get("")
async def list_graphs_endpoint(
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=-1, le=50),
order: int = Query(-1),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
graphs = await list_all_graphs(session, user_id=user_id, filters=None, skip=skip, limit=limit, order=order)
return {"message": "ok", "status": "success", "count": len(graphs or []), "graphs": graphs or []}
@router.post("", status_code=status.HTTP_201_CREATED)
async def create_graph_endpoint(
body: CreateGraphBody = Body(...),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
ws = (await session.execute(
sa.select(workspaces.c.id).where(workspaces.c.id == body.workspace)
)).scalar_one_or_none()
if not ws:
raise HTTPException(status_code=404, detail="Workspace not found")
dup = (await session.execute(
sa.select(graphs.c.id).where(
graphs.c.workspace_id == body.workspace,
graphs.c.created_by == user_id,
graphs.c.name == body.name.strip(),
).limit(1)
)).scalar_one_or_none()
if dup:
raise HTTPException(status_code=409, detail="Graph with the same name already exists in this workspace")
values = {
"id": uuid4(),
"schema_id": body.schema_,
"workspace_id": body.workspace,
"created_by": user_id,
"public": False,
"name": body.name.strip(),
}
try:
row = await insert_returning(session, graphs, values)
await session.commit()
except IntegrityError as e:
await session.rollback()
raise HTTPException(status_code=400, detail=f"Failed to create graph: {e.orig}")
return {"message": "created", "status": "success", "graph": row}
@router.get("/{graph_id}")
async def read_graph_endpoint(
graph_id: UUID,
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
g = await get_graph(session, graph_id=graph_id, user_id=user_id, public=False)
if not g:
raise HTTPException(status_code=404, detail="Graph not found.")
return {"message": "ok", "status": "success", "count": 1, "graphs": [g]}
@router.get("/{graph_id}/relations")
async def relations_endpoint(
graph_id: UUID,
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=-1, le=200),
order: int = Query(-1),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
rels, total = await list_relations(session, user_id=user_id, graph_id=graph_id, skip=skip, limit=limit, order=order)
return {"message": "ok", "status": "success", "count": total, "relations": rels}
@router.get("/{graph_id}/nodes")
async def nodes_endpoint(
graph_id: UUID,
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=-1, le=200),
order: int = Query(-1),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
nodes_, total = await list_nodes(session, user_id=user_id, graph_id=graph_id, skip=skip, limit=limit, order=order)
return {"message": "ok", "status": "success", "count": total, "nodes": nodes_}
@router.post("/from_triples")
async def create_graph_from_triples_endpoint(
graph_id: UUID = Query(..., description="已有图 ID"),
triples: List[Triple] = Body(..., embed=True),
session: AsyncSession = Depends(get_pg),
llm_client: LLMClient = Depends(get_llm_client),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
await build_graph_pg(session=session, llm_client=llm_client, graph_id=graph_id, user_id=user_id, triples_in=triples)
return {"message": "ok", "status": "success"}
@router.delete("/{graph_id}")
async def delete_graph_endpoint(
graph_id: UUID,
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
await delete_graphs(session, user_id=user_id, graph_ids=[graph_id])
return {"message": "Graph deleted successfully.", "status": "success"}

View File

@ -0,0 +1,245 @@
"""Node CRUD routes (openGauss / SQLAlchemy via services.crud.node_pg)."""
from __future__ import annotations
from typing import Any, Dict, List, Optional
from uuid import UUID
from fastapi import APIRouter, Depends, Header, HTTPException, Query, Body
from sqlalchemy.ext.asyncio import AsyncSession
from whyhow_api.dependencies import get_pg
from whyhow_api.services.crud.user_pg import get_user_by_api_key
from whyhow_api.services.crud import node_pg
from whyhow_api.schemas.nodes import (
NodeCreate,
NodeUpdate,
NodesResponse,
NodeChunksResponse,
)
from whyhow_api.services.crud.graph_pg import chunks
import sqlalchemy as sa
router = APIRouter(tags=["Nodes"], prefix="/nodes")
async def _require_user_id(session: AsyncSession, api_key: str) -> UUID:
"""检查 x-api-key 并返回 user_id。"""
user = await get_user_by_api_key(session, api_key)
if not user:
raise HTTPException(status_code=401, detail="Invalid API key")
return user["id"]
def _normalize_label_from_payload(payload: Dict[str, Any]) -> None:
"""
某些旧 Schema 里节点类型叫 `type`PG 实现里用 `label`
为了兼容 body/type 存在则映射到 label
"""
if payload is None:
return
if "label" not in payload and "type" in payload and payload["type"] is not None:
payload["label"] = payload["type"]
# ------------------------- Create -------------------------
@router.post("", response_model=NodesResponse)
async def create_node(
body: NodeCreate,
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
) -> NodesResponse:
"""
创建节点PG
- NodeCreate 若为 { graph, name, type, properties?, chunks? }将把 type 映射到 label
"""
user_id = await _require_user_id(session, api_key)
payload = body.model_dump()
_normalize_label_from_payload(payload)
for k in ("graph", "name", "label"):
if payload.get(k) in (None, ""):
raise HTTPException(status_code=422, detail=f"Missing field: {k}")
node = await node_pg.create_node(
session=session,
graph_id=UUID(str(payload["graph"])),
created_by=user_id,
name=str(payload["name"]),
label=str(payload["label"]),
properties=payload.get("properties") or {},
chunks=payload.get("chunks") or [],
)
return NodesResponse(
message="Node created successfully",
status="success",
count=1,
nodes=[node],
)
# ------------------------- List -------------------------
@router.get("", response_model=NodesResponse)
async def list_nodes_endpoint(
graph_id: UUID = Query(..., description="图 ID"),
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=-1, le=50),
order: int = Query(-1, description="排序:-1=按创建时间倒序1=正序"),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
) -> NodesResponse:
"""
按图分页列出节点PG
Mongo 版不同这里最小实现先要求提供 graph_id
后续你若需要 name/type/workspace 等更多过滤可在 node_pg.py 中扩展 SQL 条件
"""
user_id = await _require_user_id(session, api_key)
nodes, total = await node_pg.list_nodes(
session=session,
graph_id=graph_id,
user_id=user_id,
skip=skip,
limit=limit,
order=order,
)
return NodesResponse(
message="Nodes retrieved successfully",
status="success",
nodes=nodes,
count=total,
)
# ------------------------- Read One -------------------------
@router.get("/{node_id}", response_model=NodesResponse)
async def read_node_endpoint(
node_id: UUID,
graph_id: Optional[UUID] = Query(None, description="可选:校验该节点属于的图"),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
) -> NodesResponse:
"""获取单节点(可选校验 graph_id"""
user_id = await _require_user_id(session, api_key)
node = await node_pg.get_node(
session=session,
node_id=node_id,
graph_id=graph_id,
user_id=user_id,
)
if not node:
raise HTTPException(status_code=404, detail="Node not found.")
return NodesResponse(
message="Node retrieved successfully",
status="success",
count=1,
nodes=[node],
)
# ------------------------- Update -------------------------
@router.put("/{node_id}", response_model=NodesResponse)
async def update_node_endpoint(
node_id: UUID,
body: NodeUpdate,
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
) -> NodesResponse:
"""
更新节点PG
- NodeUpdate 使用 `type` 字段这里会映射到 label
"""
user_id = await _require_user_id(session, api_key)
payload = body.model_dump(exclude_unset=True)
_normalize_label_from_payload(payload)
node = await node_pg.update_node(
session=session,
node_id=node_id,
user_id=user_id,
name=payload.get("name"),
label=payload.get("label"),
properties=payload.get("properties"),
chunks=payload.get("chunks"),
)
if not node:
raise HTTPException(status_code=404, detail="Node not found or not owned by user.")
return NodesResponse(
message="Node updated successfully",
status="success",
count=1,
nodes=[node],
)
# ------------------------- Delete -------------------------
@router.delete(
"/{node_id}",
response_model=NodesResponse,
description="删除节点;如需同时清理关联三元组,可在 services 层串联 triple 清理。",
)
async def delete_node_endpoint(
node_id: UUID,
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
) -> NodesResponse:
"""删除当前用户创建的节点。"""
user_id = await _require_user_id(session, api_key)
old = await node_pg.get_node(session=session, node_id=node_id, user_id=user_id)
if not old:
raise HTTPException(status_code=404, detail="Node not found or not owned by user.")
ok = await node_pg.delete_node(session=session, node_id=node_id, user_id=user_id)
if not ok:
raise HTTPException(status_code=404, detail="Node not found or not owned by user.")
return NodesResponse(
message="Node deleted successfully",
status="success",
count=1,
nodes=[old],
)
# ------------------------- /{node_id}/chunks -------------------------
@router.get("/{node_id}/chunks", response_model=NodeChunksResponse)
async def read_node_with_chunks_endpoint(
node_id: UUID,
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
) -> NodeChunksResponse:
"""
读取节点引用到的 chunks
- 先从 nodes chunks 数组再到 chunks 表做 IN 查询
"""
user_id = await _require_user_id(session, api_key)
node = await node_pg.get_node(session=session, node_id=node_id, user_id=user_id)
if not node:
raise HTTPException(status_code=404, detail="Node not found.")
chunk_ids = list(node.chunks or [])
if not chunk_ids:
return NodeChunksResponse(message="No chunks found for the node.", status="success", count=0, chunks=[])
stmt = sa.select(
chunks.c.id,
chunks.c.data_type,
chunks.c.content,
chunks.c.content_obj,
chunks.c.metadata,
chunks.c.created_at,
).where(chunks.c.id.in_(chunk_ids))
rows = (await session.execute(stmt)).mappings().all()
return NodeChunksResponse(
message="Node with chunks retrieved successfully.",
status="success",
count=len(rows),
chunks=[dict(r) for r in rows],
)

View File

@ -0,0 +1,387 @@
# whyhow_api/routers/queries.py
from __future__ import annotations
from typing import Any, Dict, Optional, List, Tuple
from uuid import UUID, uuid4
from fastapi import APIRouter, Depends, Header, HTTPException, Query, Body
from sqlalchemy.ext.asyncio import AsyncSession
from whyhow_api.dependencies import get_pg, get_llm_client
from whyhow_api.models.common import LLMClient
from whyhow_api.services.crud.user_pg import get_user_by_api_key
from whyhow_api.services.crud.queries_pg import (
list_queries, count_queries, get_query, delete_query, queries,
)
import sqlalchemy as sa
metadata = sa.MetaData()
UUIDT = sa.dialects.postgresql.UUID(as_uuid=True)
chunks = sa.Table(
"chunks", metadata,
sa.Column("id", UUIDT, primary_key=True),
sa.Column("workspaces", sa.ARRAY(UUIDT)),
sa.Column("data_type", sa.Text),
sa.Column("content", sa.Text),
sa.Column("content_obj", sa.JSON),
sa.Column("embedding", sa.JSON),
sa.Column("tags", sa.JSON),
sa.Column("user_metadata", sa.JSON),
sa.Column("document_id", UUIDT),
sa.Column("created_by", UUIDT),
sa.Column("created_at", sa.DateTime(timezone=True)),
sa.Column("updated_at", sa.DateTime(timezone=True)),
)
router = APIRouter(tags=["Queries"], prefix="/queries")
async def _require_user_id(session: AsyncSession, api_key: str) -> UUID:
u = await get_user_by_api_key(session, api_key)
if not u:
raise HTTPException(status_code=401, detail="Invalid API key")
return u["id"]
async def _run_rag_core(
*,
session: AsyncSession,
api_key: str,
llm_client: LLMClient,
workspace_id: UUID,
text: str,
top_k: int = 5,
filters: Dict[str, Any] | None = None,
model: str = "gpt-4o-mini",
max_tokens: int = 512,
temperature: float = 0.2,
) -> Dict[str, Any]:
user_id = await _require_user_id(session, api_key)
if not text.strip():
raise HTTPException(status_code=422, detail="text is required")
top_k = max(1, min(int(top_k), 50))
filters = filters or {}
from whyhow_api.utilities.common import embed_texts
qv = (await embed_texts(llm_client=llm_client, texts=[text]))[0]
ws_bind = sa.bindparam("ws_id", workspace_id)
where = [
chunks.c.created_by == user_id,
chunks.c.embedding.is_not(None),
sa.cast(ws_bind, UUIDT) == sa.any_(chunks.c.workspaces),
]
if filters.get("document_id"):
where.append(chunks.c.document_id == UUID(str(filters["document_id"])))
cand_rows = (await session.execute(
sa.select(
chunks.c.id, chunks.c.data_type, chunks.c.content, chunks.c.content_obj,
chunks.c.embedding, chunks.c.tags, chunks.c.user_metadata, chunks.c.document_id
)
.where(*where)
.order_by(chunks.c.created_at.desc(), chunks.c.id.desc())
.limit(500)
)).mappings().all()
import math
def cos(a: list[float], b: list[float]) -> float:
if not a or not b: return -1.0
s = sum(x*y for x, y in zip(a, b))
na = math.sqrt(sum(x*x for x in a))
nb = math.sqrt(sum(y*y for y in b))
return (s/(na*nb)) if na and nb else -1.0
scored = []
for r in cand_rows:
emb = r["embedding"]
if isinstance(emb, dict): emb = emb.get("vector")
if isinstance(emb, list): scored.append((cos(qv, emb), r))
scored.sort(key=lambda x: x[0], reverse=True)
top = scored[:top_k]
import json
def chunk_to_text(r: Dict[str, Any]) -> str:
if r["data_type"] == "string":
return r["content"] or ""
return json.dumps(r["content_obj"] or r["content"] or {}, ensure_ascii=False)
context_lines, picked = [], []
for s, r in top:
picked.append({"id": str(r["id"]),
"score": float(s),
"data_type": r["data_type"],
"document_id": str(r["document_id"]) if r["document_id"] else None})
context_lines.append(f"- [{s:.3f}] {chunk_to_text(r)[:800]}")
context = "\n".join(context_lines) if context_lines else "(no context)"
answer = None
try:
msgs = [
{"role": "system", "content": "你是专业知识助手,只基于提供的上下文回答;如果无法确定就直说。"},
{"role": "user", "content": f"问题:{text}\n\n可用上下文:\n{context}"},
]
if hasattr(llm_client, "client"):
resp = await llm_client.client.chat.completions.create(
model=model, messages=msgs, temperature=temperature, max_tokens=max_tokens
)
answer = (resp.choices[0].message.content or "").strip()
elif hasattr(llm_client, "chat"):
resp = await llm_client.chat(messages=msgs, model=model,
temperature=temperature, max_tokens=max_tokens)
answer = (resp["content"] if isinstance(resp, dict) else str(resp)).strip()
except Exception:
answer = None
payload = {
"request": {"workspace_id": str(workspace_id), "text": text, "top_k": top_k, "filters": filters},
"hits": picked,
"answer": answer,
}
row = (await session.execute(
sa.insert(queries).values(
id=uuid4(), user_id=user_id, graph_id=None, status="completed",
name=text[:255], payload=payload
).returning(*queries.c)
)).mappings().one()
await session.commit()
return {"message": "ok", "status": "success",
"query_id": str(row["id"]), "answer": answer, "top_chunks": picked}
# ---------- RAG One-shotPOST /queries ----------
@router.post("/rag")
async def rag_post(
body: Dict[str, Any],
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
llm_client: LLMClient = Depends(get_llm_client),
):
if not body.get("workspace_id") or not body.get("text"):
raise HTTPException(status_code=422, detail="workspace_id 与 text 均为必填")
return await _run_rag_core(
session=session, api_key=api_key, llm_client=llm_client,
workspace_id=UUID(str(body["workspace_id"])),
text=str(body["text"]),
top_k=int(body.get("top_k") or 5),
filters=body.get("filters") or {},
model=body.get("model") or "gpt-4o-mini",
max_tokens=int(body.get("max_tokens") or 512),
temperature=float(body.get("temperature") or 0.2),
)
@router.get("/rag")
async def rag_get(
workspace_id: UUID = Query(...),
text: str = Query(...),
top_k: int = Query(5),
document_id: UUID | None = Query(None),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
llm_client: LLMClient = Depends(get_llm_client),
):
filters = {}
if document_id: filters["document_id"] = str(document_id)
return await _run_rag_core(
session=session, api_key=api_key, llm_client=llm_client,
workspace_id=workspace_id, text=text, top_k=top_k, filters=filters,
)
@router.post("")
async def run_rag_query(
body: Dict[str, Any] = Body(..., description="RAG 查询体"),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
llm_client: LLMClient = Depends(get_llm_client),
):
"""
一步完成 workspace 过滤 chunk -> 语义相似度选 TopK -> 组上下文 -> 调用 LLM
请求体示例:
{
"workspace_id": "<uuid>",
"text": "openGauss 的优势是什么?",
"top_k": 5,
"filters": { "document_id": "<uuid-可选>" },
"model": "gpt-4o-mini", # 可选
"max_tokens": 512, # 可选
"temperature": 0.2 # 可选
}
"""
user_id = await _require_user_id(session, api_key)
# ---- 校验/默认值
ws_id: Optional[UUID] = body.get("workspace_id")
text: str = (body.get("text") or "").strip()
if not ws_id or not text:
raise HTTPException(status_code=422, detail="workspace_id 与 text 均为必填")
top_k = int(body.get("top_k") or 5)
top_k = max(1, min(top_k, 50))
filters: Dict[str, Any] = body.get("filters") or {}
model = body.get("model") or "gpt-4o-mini"
max_tokens = int(body.get("max_tokens") or 512)
temperature = float(body.get("temperature") or 0.2)
# ---- 1) 生成查询向量
from whyhow_api.utilities.common import embed_texts
qv = (await embed_texts(llm_client=llm_client, texts=[text]))[0] # list[float]
# ---- 2) 取候选 chunks只看当前用户写入的且 embedding 非空,且包含 workspace_id
where = [
chunks.c.created_by == user_id,
chunks.c.embedding.is_not(None),
# workspaces 包含 ws_id。openGauss/PG 兼容ARRAY CONTAINS
chunks.c.workspaces.contains([ws_id]),
]
if "document_id" in filters and filters["document_id"]:
where.append(chunks.c.document_id == UUID(str(filters["document_id"])))
# 先取前 500 作为候选,避免全表扫
cand_rows = (await session.execute(
sa.select(
chunks.c.id, chunks.c.data_type, chunks.c.content, chunks.c.content_obj,
chunks.c.embedding, chunks.c.tags, chunks.c.user_metadata, chunks.c.document_id
)
.where(*where)
.order_by(chunks.c.created_at.desc(), chunks.c.id.desc())
.limit(500)
)).mappings().all()
# ---- 3) 计算相似度(余弦),选 TopK
import math
def cos(a: list[float], b: list[float]) -> float:
if not a or not b:
return -1.0
s = sum(x*y for x, y in zip(a, b))
na = math.sqrt(sum(x*x for x in a))
nb = math.sqrt(sum(y*y for y in b))
return (s / (na*nb)) if na and nb else -1.0
scored = []
for r in cand_rows:
emb = r["embedding"]
if isinstance(emb, dict): # 兼容 {"vector":[...]}
emb = emb.get("vector")
if not isinstance(emb, list):
continue
scored.append((cos(qv, emb), r))
scored.sort(key=lambda x: x[0], reverse=True)
top = scored[:top_k]
# ---- 4) 组上下文
def chunk_to_text(r: Dict[str, Any]) -> str:
if r["data_type"] == "string":
return r["content"] or ""
# 非字符串做个压缩展示
import json
return json.dumps(r["content_obj"] or r["content"] or {}, ensure_ascii=False)
context_lines = []
picked_ids: List[str] = []
for score, r in top:
picked_ids.append(str(r["id"]))
snippet = chunk_to_text(r)[:800] # 控制单段长度
context_lines.append(f"- [{score:.3f}] {snippet}")
context = "\n".join(context_lines) if context_lines else "(no context)"
# ---- 5) 调用 LLM 生成答案(若失败也正常返回 top chunks
answer = None
try:
# 你自家的 LLMClient 一般暴露 openai 兼容接口llm_client.client.chat.completions.create(...)
# 保守写法:同时兼容 .client 和直接 .chat 两种封装
msgs = [
{"role": "system", "content": "你是一个专业的知识助手。请只基于提供的上下文回答;若上下文没有答案,就明确说无法确定。"},
{"role": "user", "content": f"问题:{text}\n\n可用上下文:\n{context}"},
]
if hasattr(llm_client, "client"): # OpenAI 风格
resp = await llm_client.client.chat.completions.create(
model=model, messages=msgs, temperature=temperature, max_tokens=max_tokens
)
answer = (resp.choices[0].message.content or "").strip()
elif hasattr(llm_client, "chat"): # 备用
resp = await llm_client.chat(messages=msgs, model=model, temperature=temperature, max_tokens=max_tokens)
answer = (resp["content"] if isinstance(resp, dict) else str(resp)).strip()
except Exception as e:
# 不抛错,继续返回检索结果,便于排查
answer = None
# ---- 6) 记录查询payload 里把请求/命中/答案都塞进去)
payload = {
"request": {"workspace_id": str(ws_id), "text": text, "top_k": top_k, "filters": filters},
"hits": [
{
"id": str(r["id"]),
"score": float(score),
"data_type": r["data_type"],
"document_id": str(r["document_id"]) if r["document_id"] else None,
} for score, r in top
],
"answer": answer,
}
row = (await session.execute(
sa.insert(queries).values(
id=uuid4(), user_id=user_id, graph_id=None, status="completed",
name=text[:255], payload=payload
).returning(*queries.c)
)).mappings().one()
await session.commit()
return {
"message": "ok",
"status": "success",
"query_id": str(row["id"]),
"answer": answer,
"top_chunks": payload["hits"],
}
# ---------- 列表 ----------
@router.get("")
async def list_queries_endpoint(
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=-1, le=50),
order: int = Query(-1),
status: Optional[str] = Query(None),
graph_id: Optional[UUID] = Query(None),
graph_name: Optional[str] = Query(None),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
rows = await list_queries(
session, created_by=user_id, skip=skip, limit=limit, order=order,
status=status, graph_id=graph_id, graph_name=graph_name
)
total = await count_queries(
session, created_by=user_id, status=status, graph_id=graph_id, graph_name=graph_name
)
return {"message": "ok", "status": "success", "count": total, "queries": rows}
# ---------- 读取/删除 ----------
@router.get("/{query_id}")
async def get_query_endpoint(
query_id: UUID,
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
row = await get_query(session, query_id=query_id, created_by=user_id)
if not row:
raise HTTPException(status_code=404, detail="Query not found.")
return {"message": "ok", "status": "success", "count": 1, "queries": [row]}
@router.delete("/{query_id}")
async def delete_query_endpoint(
query_id: UUID,
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
user_id = await _require_user_id(session, api_key)
row = await delete_query(session, query_id=query_id, created_by=user_id)
if not row:
raise HTTPException(status_code=404, detail="Query not found.")
return {"message": "deleted", "status": "success", "count": 1, "queries": [row]}

View File

@ -0,0 +1,95 @@
# routers/rules.py
from typing import Any, Dict, Optional, Union
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, Query, Body, Header, status
from sqlalchemy.ext.asyncio import AsyncSession
from whyhow_api.dependencies import get_pg
from whyhow_api.services.crud.rule_pg import (
create_rule as pg_create_rule,
get_workspace_rules as pg_get_workspace_rules,
get_graph_rules as pg_get_graph_rules,
delete_rule as pg_delete_rule,
)
from whyhow_api.services.crud.user_pg import get_user_by_api_key
from whyhow_api.schemas.rules import RuleWrapper
router = APIRouter(tags=["Rules"], prefix="/rules")
async def _require_user_id(session: AsyncSession, api_key: str) -> UUID:
u = await get_user_by_api_key(session, api_key)
if not u:
raise HTTPException(status_code=401, detail="Invalid API key")
return u["id"]
def _xor(a: Optional[Any], b: Optional[Any]) -> bool:
return (a is None) ^ (b is None)
@router.post("")
async def create_rule_endpoint(
workspace_id: Optional[UUID] = Query(None, description="工作区ID二选一"),
graph_id: Optional[UUID] = Query(None, description="图ID二选一"),
payload: Union[Dict[str, Any], RuleWrapper] = Body(..., description="规则 JSON可以是裸对象或 {'rule': {...}}"),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
if not _xor(workspace_id, graph_id):
raise HTTPException(status_code=400, detail="workspace_id 与 graph_id 必须二选一,且只能提供一个。")
user_id = await _require_user_id(session, api_key)
rule_body: Dict[str, Any]
if isinstance(payload, dict):
rule_body = payload
else:
rule_body = payload.rule
new_rule = await pg_create_rule(
session=session,
user_id=user_id,
rule_body=rule_body,
workspace_id=workspace_id,
graph_id=graph_id,
)
return {"message": "Rule created successfully.", "status": "success", "count": 1, "rules": [new_rule]}
@router.get("")
async def read_rules_endpoint(
workspace_id: Optional[UUID] = Query(None, description="工作区ID二选一"),
graph_id: Optional[UUID] = Query(None, description="图ID二选一"),
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=-1, le=200),
order: int = Query(-1),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
"""查询规则PG使用 API Key 识别用户)"""
if not _xor(workspace_id, graph_id):
raise HTTPException(status_code=400, detail="workspace_id 与 graph_id 必须二选一,且只能提供一个。")
user_id = await _require_user_id(session, api_key)
if workspace_id:
rules = await pg_get_workspace_rules(
session, user_id=user_id, workspace_id=workspace_id, skip=skip, limit=limit, order=order
)
else:
rules = await pg_get_graph_rules(
session, graph_id=graph_id, user_id=user_id, skip=skip, limit=limit, order=order
)
return {"message": "Rules retrieved successfully.", "status": "success", "count": len(rules), "rules": rules}
@router.delete("/{rule_id}")
async def delete_rule_endpoint(
rule_id: UUID,
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
"""删除规则PG使用 API Key 识别用户)"""
user_id = await _require_user_id(session, api_key)
ok = await pg_delete_rule(session, user_id=user_id, rule_id=rule_id)
if not ok:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Rule not found.")
return {"message": "Rule deleted successfully.", "status": "success", "count": 1}

View File

@ -0,0 +1,168 @@
"""Schemas router (openGauss/PostgreSQL)"""
import logging
from typing import Optional, Dict, Any
from uuid import UUID
import sqlalchemy as sa
from fastapi import APIRouter, Depends, HTTPException, Query, Body, Header, status
from sqlalchemy.ext.asyncio import AsyncSession
from whyhow_api.dependencies import get_pg, get_llm_client, LLMClient
from whyhow_api.services.crud.schema_pg import (
list_schemas,
get_schema,
create_schema,
delete_schema,
)
from whyhow_api.services.crud.user_pg import get_user_by_api_key
from whyhow_api.utilities.builders import OpenAIBuilder
logger = logging.getLogger(__name__)
router = APIRouter(tags=["Schemas"], prefix="/schemas")
async def _require_user_id(session: AsyncSession, api_key: str) -> UUID:
user = await get_user_by_api_key(session, api_key)
if not user:
raise HTTPException(status_code=401, detail="Invalid API key")
return user["id"]
@router.get("")
async def read_schemas_endpoint(
workspace_id: UUID = Query(..., description="工作区ID"),
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=-1, le=200),
order: int = Query(-1),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
"""列出当前用户在指定 workspace 下的 schemasPG"""
user_id = await _require_user_id(session, api_key)
schemas = await list_schemas(
session, user_id=user_id, workspace_id=workspace_id,
skip=skip, limit=limit, order=order
)
return {
"message": "Schemas retrieved successfully.",
"status": "success",
"count": len(schemas),
"schemas": schemas,
}
@router.get("/{schema_id}")
async def read_schema_endpoint(
schema_id: UUID,
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
"""获取单个 schemaPG"""
user_id = await _require_user_id(session, api_key)
s = await get_schema(session, schema_id=schema_id, user_id=user_id)
if not s:
raise HTTPException(status_code=404, detail="Schema not found")
return {
"message": "Schema retrieved successfully.",
"status": "success",
"count": 1,
"schemas": [s],
}
@router.post("")
async def create_schema_endpoint(
workspace_id: UUID = Query(...),
name: Optional[str] = Query(None),
body: Optional[Dict[str, Any]] = Body(None),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
"""创建 schemaPG"""
user_id = await _require_user_id(session, api_key)
s = await create_schema(
session, user_id=user_id, workspace_id=workspace_id, name=name, body=body
)
return {
"message": "Schema created successfully.",
"status": "success",
"count": 1,
"schemas": [s],
}
@router.put("/{schema_id}")
async def update_schema_endpoint(
schema_id: UUID,
name: Optional[str] = Query(None),
body: Optional[Dict[str, Any]] = Body(None),
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
"""更新 schemaPG"""
from whyhow_api.services.crud.schema_pg import schemas as schemas_tbl
user_id = await _require_user_id(session, api_key)
values: Dict[str, Any] = {}
if name is not None:
values["name"] = name
if body is not None:
values["body"] = body
if not values:
raise HTTPException(status_code=400, detail="Nothing to update")
stmt = (
sa.update(schemas_tbl)
.where(schemas_tbl.c.id == schema_id, schemas_tbl.c.created_by == user_id)
.values(**values)
.returning(*schemas_tbl.c)
)
row = (await session.execute(stmt)).mappings().first()
if not row:
raise HTTPException(status_code=404, detail="Schema not found")
await session.commit()
return {
"message": "Schema updated successfully.",
"status": "success",
"count": 1,
"schemas": [dict(row)],
}
@router.delete("/{schema_id}")
async def delete_schema_endpoint(
schema_id: UUID,
session: AsyncSession = Depends(get_pg),
api_key: str = Header(..., alias="x-api-key"),
):
"""删除 schemaPG"""
user_id = await _require_user_id(session, api_key)
ok = await pg_delete_schema(session, user_id=user_id, schema_id=schema_id)
if not ok:
# 被图引用时返回 409
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Cannot delete schema with associated graphs.",
)
return {"message": "Schema deleted successfully.", "status": "success", "count": 1}
@router.post("/generate")
async def generate_schema_endpoint(
questions: list[str] = Body(..., embed=True),
llm_client: LLMClient = Depends(get_llm_client),
):
"""用 LLM 生成 schema无 DB 依赖)"""
generated_schema, errors = await OpenAIBuilder.generate_schema(
llm_client=llm_client,
questions=questions,
)
return {
"message": "Schema generated successfully.",
"status": "success",
"questions": questions,
"generated_schema": generated_schema,
"errors": errors,
}

View File

@ -0,0 +1,39 @@
"""Task CRUD router (openGauss)"""
from uuid import UUID
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from whyhow_api.dependencies import get_pg
from whyhow_api.services.crud.task_pg import get_task, create_task
router = APIRouter(prefix="/tasks", tags=["tasks"])
@router.get("/{task_id}")
async def get_task_by_id(
task_id: UUID,
created_by: UUID = Query(..., description="任务创建者用户ID"),
session: AsyncSession = Depends(get_pg),
):
"""查询单个任务"""
data = await get_task(session, task_id=task_id, created_by=created_by)
if not data:
raise HTTPException(status_code=404, detail="Task not found")
return {"message": "ok", "status": "success", "task": data}
@router.post("")
async def create_task_api(
created_by: UUID = Query(..., description="任务创建者用户ID"),
title: str = Query(..., description="任务标题"),
description: Optional[str] = Query(None, description="任务描述"),
session: AsyncSession = Depends(get_pg),
):
"""创建任务"""
task = await create_task(
session,
created_by=created_by,
title=title,
description=description,
)
return {"message": "created", "status": "success", "task": task}

Some files were not shown because too many files have changed in this diff Show More