jp170na/dp-zp-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

upravy

Browse Source
This commit is contained in:
Ján Pták 2026-08-13 21:39:42 +02:00
parent 32004e89bf
commit f67de6dcaa
8 changed files with 1176 additions and 417 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

258
a nachadza v staged obsahu"
View File

@ -1,258 +0,0 @@
SSUUMMMMAARRYY OOFF LLEESSSS CCOOMMMMAANNDDSS
Commands marked with * may be preceded by a number, _N.
Notes in parentheses indicate the behavior if _N is given.
A key preceded by a caret indicates the Ctrl key; thus ^K is ctrl-K.
h H Display this help.
q :q Q :Q ZZ Exit.
---------------------------------------------------------------------------
MMOOVVIINNGG
e ^E j ^N CR * Forward one line (or _N lines).
y ^Y k ^K ^P * Backward one line (or _N lines).
f ^F ^V SPACE * Forward one window (or _N lines).
b ^B ESC-v * Backward one window (or _N lines).
z * Forward one window (and set window to _N).
w * Backward one window (and set window to _N).
ESC-SPACE * Forward one window, but don't stop at end-of-file.
d ^D * Forward one half-window (and set half-window to _N).
u ^U * Backward one half-window (and set half-window to _N).
ESC-) RightArrow * Right one half screen width (or _N positions).
ESC-( LeftArrow * Left one half screen width (or _N positions).
ESC-} ^RightArrow Right to last column displayed.
ESC-{ ^LeftArrow Left to first column.
F Forward forever; like "tail -f".
ESC-F Like F but stop when search pattern is found.
r ^R ^L Repaint screen.
R Repaint screen, discarding buffered input.
---------------------------------------------------
Default "window" is the screen height.
Default "half-window" is half of the screen height.
---------------------------------------------------------------------------
SSEEAARRCCHHIINNGG
/_p_a_t_t_e_r_n * Search forward for (_N-th) matching line.
?_p_a_t_t_e_r_n * Search backward for (_N-th) matching line.
n * Repeat previous search (for _N-th occurrence).
N * Repeat previous search in reverse direction.
ESC-n * Repeat previous search, spanning files.
ESC-N * Repeat previous search, reverse dir. & spanning files.
ESC-u Undo (toggle) search highlighting.
ESC-U Clear search highlighting.
&_p_a_t_t_e_r_n * Display only matching lines.
---------------------------------------------------
A search pattern may begin with one or more of:
^N or ! Search for NON-matching lines.
^E or * Search multiple files (pass thru END OF FILE).
^F or @ Start search at FIRST file (for /) or last file (for ?).
^K Highlight matches, but don't move (KEEP position).
^R Don't use REGULAR EXPRESSIONS.
^W WRAP search if no match found.
---------------------------------------------------------------------------
JJUUMMPPIINNGG
g < ESC-< * Go to first line in file (or line _N).
G > ESC-> * Go to last line in file (or line _N).
p % * Go to beginning of file (or _N percent into file).
t * Go to the (_N-th) next tag.
T * Go to the (_N-th) previous tag.
{ ( [ * Find close bracket } ) ].
} ) ] * Find open bracket { ( [.
ESC-^F _<_c_1_> _<_c_2_> * Find close bracket _<_c_2_>.
ESC-^B _<_c_1_> _<_c_2_> * Find open bracket _<_c_1_>.
---------------------------------------------------
Each "find close bracket" command goes forward to the close bracket
matching the (_N-th) open bracket in the top line.
Each "find open bracket" command goes backward to the open bracket
matching the (_N-th) close bracket in the bottom line.
m_<_l_e_t_t_e_r_> Mark the current top line with <letter>.
M_<_l_e_t_t_e_r_> Mark the current bottom line with <letter>.
'_<_l_e_t_t_e_r_> Go to a previously marked position.
'' Go to the previous position.
^X^X Same as '.
ESC-M_<_l_e_t_t_e_r_> Clear a mark.
---------------------------------------------------
A mark is any upper-case or lower-case letter.
Certain marks are predefined:
^ means beginning of the file
$ means end of the file
---------------------------------------------------------------------------
CCHHAANNGGIINNGG FFIILLEESS
:e [_f_i_l_e] Examine a new file.
^X^V Same as :e.
:n * Examine the (_N-th) next file from the command line.
:p * Examine the (_N-th) previous file from the command line.
:x * Examine the first (or _N-th) file from the command line.
:d Delete the current file from the command line list.
= ^G :f Print current file name.
---------------------------------------------------------------------------
MMIISSCCEELLLLAANNEEOOUUSS CCOOMMMMAANNDDSS
-_<_f_l_a_g_> Toggle a command line option [see OPTIONS below].
--_<_n_a_m_e_> Toggle a command line option, by name.
__<_f_l_a_g_> Display the setting of a command line option.
___<_n_a_m_e_> Display the setting of an option, by name.
+_c_m_d Execute the less cmd each time a new file is examined.
!_c_o_m_m_a_n_d Execute the shell command with $SHELL.
|XX_c_o_m_m_a_n_d Pipe file between current pos & mark XX to shell command.
s _f_i_l_e Save input to a file.
v Edit the current file with $VISUAL or $EDITOR.
V Print version number of "less".
---------------------------------------------------------------------------
OOPPTTIIOONNSS
Most options may be changed either on the command line,
or from within less by using the - or -- command.
Options may be given in one of two forms: either a single
character preceded by a -, or a name preceded by --.
-? ........ --help
Display help (from command line).
-a ........ --search-skip-screen
Search skips current screen.
-A ........ --SEARCH-SKIP-SCREEN
Search starts just after target line.
-b [_N] .... --buffers=[_N]
Number of buffers.
-B ........ --auto-buffers
Don't automatically allocate buffers for pipes.
-c ........ --clear-screen
Repaint by clearing rather than scrolling.
-d ........ --dumb
Dumb terminal.
-D xx_c_o_l_o_r . --color=xx_c_o_l_o_r
Set screen colors.
-e -E .... --quit-at-eof --QUIT-AT-EOF
Quit at end of file.
-f ........ --force
Force open non-regular files.
-F ........ --quit-if-one-screen
Quit if entire file fits on first screen.
-g ........ --hilite-search
Highlight only last match for searches.
-G ........ --HILITE-SEARCH
Don't highlight any matches for searches.
-h [_N] .... --max-back-scroll=[_N]
Backward scroll limit.
-i ........ --ignore-case
Ignore case in searches that do not contain uppercase.
-I ........ --IGNORE-CASE
Ignore case in all searches.
-j [_N] .... --jump-target=[_N]
Screen position of target lines.
-J ........ --status-column
Display a status column at left edge of screen.
-k [_f_i_l_e] . --lesskey-file=[_f_i_l_e]
Use a lesskey file.
-K ........ --quit-on-intr
Exit less in response to ctrl-C.
-L ........ --no-lessopen
Ignore the LESSOPEN environment variable.
-m -M .... --long-prompt --LONG-PROMPT
Set prompt style.
-n -N .... --line-numbers --LINE-NUMBERS
Don't use line numbers.
-o [_f_i_l_e] . --log-file=[_f_i_l_e]
Copy to log file (standard input only).
-O [_f_i_l_e] . --LOG-FILE=[_f_i_l_e]
Copy to log file (unconditionally overwrite).
-p [_p_a_t_t_e_r_n] --pattern=[_p_a_t_t_e_r_n]
Start at pattern (from command line).
-P [_p_r_o_m_p_t] --prompt=[_p_r_o_m_p_t]
Define new prompt.
-q -Q .... --quiet --QUIET --silent --SILENT
Quiet the terminal bell.
-r -R .... --raw-control-chars --RAW-CONTROL-CHARS
Output "raw" control characters.
-s ........ --squeeze-blank-lines
Squeeze multiple blank lines.
-S ........ --chop-long-lines
Chop (truncate) long lines rather than wrapping.
-t [_t_a_g] .. --tag=[_t_a_g]
Find a tag.
-T [_t_a_g_s_f_i_l_e] --tag-file=[_t_a_g_s_f_i_l_e]
Use an alternate tags file.
-u -U .... --underline-special --UNDERLINE-SPECIAL
Change handling of backspaces.
-V ........ --version
Display the version number of "less".
-w ........ --hilite-unread
Highlight first new line after forward-screen.
-W ........ --HILITE-UNREAD
Highlight first new line after any forward movement.
-x [_N[,...]] --tabs=[_N[,...]]
Set tab stops.
-X ........ --no-init
Don't use termcap init/deinit strings.
-y [_N] .... --max-forw-scroll=[_N]
Forward scroll limit.
-z [_N] .... --window=[_N]
Set size of window.
-" [_c[_c]] . --quotes=[_c[_c]]
Set shell quote characters.
-~ ........ --tilde
Don't display tildes after end of file.
-# [_N] .... --shift=[_N]
Set horizontal scroll amount (0 = one half screen width).
--file-size
Automatically determine the size of the input file.
--follow-name
The F command changes files if the input file is renamed.
--incsearch
Search file as each pattern character is typed in.
--line-num-width=N
Set the width of the -N line number field to N characters.
--mouse
Enable mouse input.
--no-keypad
Don't send termcap keypad init/deinit strings.
--no-histdups
Remove duplicates from command history.
--rscroll=C
Set the character used to mark truncated lines.
--save-marks
Retain marks across invocations of less.
--status-col-width=N
Set the width of the -J status column to N characters.
--use-backslash
Subsequent options use backslash as escape char.
--use-color
Enables colored text.
--wheel-lines=N
Each click of the mouse wheel moves N lines.
---------------------------------------------------------------------------
LLIINNEE EEDDIITTIINNGG
These keys can be used to edit text being entered
on the "command line" at the bottom of the screen.
RightArrow ..................... ESC-l ... Move cursor right one character.
LeftArrow ...................... ESC-h ... Move cursor left one character.
ctrl-RightArrow ESC-RightArrow ESC-w ... Move cursor right one word.
ctrl-LeftArrow ESC-LeftArrow ESC-b ... Move cursor left one word.
HOME ........................... ESC-0 ... Move cursor to start of line.
END ............................ ESC-$ ... Move cursor to end of line.
BACKSPACE ................................ Delete char to left of cursor.
DELETE ......................... ESC-x ... Delete char under cursor.
ctrl-BACKSPACE ESC-BACKSPACE ........... Delete word to left of cursor.
ctrl-DELETE .... ESC-DELETE .... ESC-X ... Delete word under cursor.
ctrl-U ......... ESC (MS-DOS only) ....... Delete entire line.
UpArrow ........................ ESC-k ... Retrieve previous command line.
DownArrow ...................... ESC-j ... Retrieve next command line.
TAB ...................................... Complete filename & cycle.
SHIFT-TAB ...................... ESC-TAB Complete filename & reverse cycle.
ctrl-L ................................... Complete filename, list all.

480
app/main.py
View File

@ -19,16 +19,29 @@ from fastapi import (
Security,
status,
)
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi.security import APIKeyHeader
from fastapi.security import (
APIKeyHeader,
HTTPAuthorizationCredentials,
HTTPBearer,
)
from pydantic import BaseModel, Field
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
sys.path.insert(
0,
str(PROJECT_ROOT),
)
from scripts.common import DB_FILE, ZPWIKI_ROOT
from scripts.common import (
DB_FILE,
ZPWIKI_ROOT,
)
from scripts.embedding_utils import embed_query
from scripts.rag_utils import build_rag_context
from scripts.rebuild_index import (
ReindexInProgressError,
rebuild_index,
@ -41,17 +54,38 @@ MIN_SECRET_LENGTH = 32
SEARCH_API_KEY_HEADER = "X-API-Key"
SYNC_API_KEY_HEADER = "X-API-Key"
OPENWEBUI_ORIGIN = (
"https://ui.tukekemt.xyz"
)
search_api_key_scheme = APIKeyHeader(
name=SEARCH_API_KEY_HEADER,
scheme_name="SearchApiKey",
auto_error=False,
description="API kľúč pre vyhľadávanie v zpwiki.",
description=(
"API kľúč pre vyhľadávanie v ZP Wiki "
"cez hlavičku X-API-Key."
),
)
search_bearer_scheme = HTTPBearer(
scheme_name="SearchBearer",
auto_error=False,
description=(
"Bearer token pre vyhľadávanie v ZP Wiki. "
"Používa hodnotu SEARCH_API_KEY."
),
)
sync_api_key_scheme = APIKeyHeader(
name=SYNC_API_KEY_HEADER,
scheme_name="SyncApiKey",
auto_error=False,
description="API kľúč pre manuálne spustenie reindexovania.",
description=(
"API kľúč pre manuálne spustenie "
"reindexovania."
),
)
@ -68,36 +102,91 @@ class SearchRequest(BaseModel):
)
published_only: bool = False
max_per_document: int = Field(
default=3,
default=1,
ge=0,
le=10,
)
class RagRequest(BaseModel):
query: str = Field(
...,
min_length=1,
max_length=500,
description=(
"Otázka alebo vyhľadávací dotaz "
"používateľa nad ZP Wiki."
),
)
limit: int = Field(
default=5,
ge=1,
le=20,
description=(
"Maximálny počet relevantných "
"zdrojov pre RAG kontext."
),
)
published_only: bool = Field(
default=False,
description=(
"Ak je true, použijú sa iba "
"publikované dokumenty."
),
)
max_per_document: int = Field(
default=1,
ge=0,
le=10,
description=(
"Maximálny počet chunkov z jedného "
"dokumentu. Hodnota 1 preferuje "
"rôzne dokumenty."
),
)
class SyncRequest(BaseModel):
pull_git: bool = Field(
default=False,
description="Pred reindexovaním vykoná git pull --ff-only.",
description=(
"Pred reindexovaním vykoná "
"git pull --ff-only."
),
)
def required_environment_value(name: str) -> str:
value = os.getenv(name, "").strip()
def required_environment_value(
name: str,
) -> str:
value = os.getenv(
name,
"",
).strip()
if not value:
raise RuntimeError(
f"Chýba povinná environment premenná {name}"
"Chýba povinná environment "
f"premenná {name}"
)
return value
def validate_secret(name: str) -> str:
value = required_environment_value(name)
def validate_secret(
name: str,
) -> str:
value = required_environment_value(
name
)
if len(value) < MIN_SECRET_LENGTH:
raise RuntimeError(
f"{name} musí mať aspoň {MIN_SECRET_LENGTH} znakov"
f"{name} musí mať aspoň "
f"{MIN_SECRET_LENGTH} znakov"
)
return value
@ -110,8 +199,8 @@ def expected_gitea_repository() -> str:
if "/" not in value:
raise RuntimeError(
"EXPECTED_GITEA_REPOSITORY musí mať tvar "
"vlastník/repozitár"
"EXPECTED_GITEA_REPOSITORY musí mať "
"tvar vlastník/repozitár"
)
return value
@ -132,50 +221,115 @@ def webhook_should_pull_git() -> bool:
def validate_security_configuration() -> None:
validate_secret("WEBHOOK_SECRET")
validate_secret("SYNC_API_KEY")
validate_secret("SEARCH_API_KEY")
validate_secret(
"WEBHOOK_SECRET"
)
validate_secret(
"SYNC_API_KEY"
)
validate_secret(
"SEARCH_API_KEY"
)
expected_gitea_repository()
@asynccontextmanager
async def lifespan(_: FastAPI):
async def lifespan(
_: FastAPI,
):
validate_security_configuration()
# Predhriatie embedding modelu.
await asyncio.to_thread(
embed_query,
"warmup",
)
yield
app = FastAPI(
title="ZP Agent API",
description=(
"API pre vyhľadávanie v repozitári "
"záverečných prác zpwiki."
"RAG API pre vyhľadávanie "
"a získavanie informácií "
"z repozitára záverečných prác "
"ZP Wiki."
),
version="0.6.0",
version="0.8.0",
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=[
OPENWEBUI_ORIGIN,
],
allow_credentials=True,
allow_methods=[
"GET",
"POST",
"OPTIONS",
],
allow_headers=["*"],
allow_private_network=True,
)
def require_search_api_key(
api_key: str | None = Security(
search_api_key_scheme
),
bearer: (
HTTPAuthorizationCredentials | None
) = Security(
search_bearer_scheme
),
) -> None:
expected = validate_secret(
"SEARCH_API_KEY"
)
supplied_credentials: list[str] = []
if api_key:
supplied_credentials.append(
api_key
)
if (
not api_key
or not hmac.compare_digest(
api_key,
bearer is not None
and bearer.scheme.casefold()
== "bearer"
and bearer.credentials
):
supplied_credentials.append(
bearer.credentials
)
valid = any(
hmac.compare_digest(
supplied,
expected,
)
):
for supplied
in supplied_credentials
)
if not valid:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Neplatný alebo chýbajúci API kľúč",
status_code=(
status.HTTP_401_UNAUTHORIZED
),
detail=(
"Neplatný alebo chýbajúci "
"API kľúč"
),
headers={
"WWW-Authenticate": "ApiKey",
"WWW-Authenticate": (
"Bearer"
),
},
)
@ -197,10 +351,17 @@ def require_sync_api_key(
)
):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Neplatný alebo chýbajúci API kľúč",
status_code=(
status.HTTP_401_UNAUTHORIZED
),
detail=(
"Neplatný alebo chýbajúci "
"API kľúč"
),
headers={
"WWW-Authenticate": "ApiKey",
"WWW-Authenticate": (
"ApiKey"
),
},
)
@ -213,10 +374,15 @@ def verify_gitea_signature(
if not signature:
return False
supplied = signature.strip().casefold()
supplied = (
signature
.strip()
.casefold()
)
# Kompatibilita podpisu.
if supplied.startswith("sha256="):
if supplied.startswith(
"sha256="
):
supplied = supplied.removeprefix(
"sha256="
)
@ -225,13 +391,18 @@ def verify_gitea_signature(
return False
try:
int(supplied, 16)
int(
supplied,
16,
)
except ValueError:
return False
expected = hmac.new(
secret.encode("utf-8"),
secret.encode(
"utf-8"
),
raw_body,
hashlib.sha256,
).hexdigest()
@ -245,17 +416,29 @@ def verify_gitea_signature(
def repository_name_from_payload(
payload: dict[str, Any],
) -> str | None:
repository = payload.get("repository")
repository = payload.get(
"repository"
)
if not isinstance(repository, dict):
if not isinstance(
repository,
dict,
):
return None
value = (
repository.get("full_name")
or repository.get("name")
repository.get(
"full_name"
)
or repository.get(
"name"
)
)
if not isinstance(value, str):
if not isinstance(
value,
str,
):
return None
value = value.strip()
@ -273,15 +456,29 @@ def same_repository(
)
@app.get("/health")
@app.get(
"/health",
include_in_schema=False,
)
def health() -> dict[str, Any]:
return {
"status": "ok",
"database_exists": DB_FILE.exists(),
"database_path": str(DB_FILE),
"search_engine": "sqlite_fts5",
"zpwiki_root": str(ZPWIKI_ROOT),
"zpwiki_exists": ZPWIKI_ROOT.exists(),
"database_exists": (
DB_FILE.exists()
),
"database_path": str(
DB_FILE
),
"search_engine": (
"hybrid_fts5_embeddings"
),
"rag_enabled": True,
"zpwiki_root": str(
ZPWIKI_ROOT
),
"zpwiki_exists": (
ZPWIKI_ROOT.exists()
),
"security_configured": all(
bool(
os.getenv(
@ -299,11 +496,80 @@ def health() -> dict[str, Any]:
}
@app.post(
"/rag",
operation_id=(
"retrieve_zpwiki_context"
),
summary=(
"Vyhľadaj informácie v ZP Wiki"
),
description=(
"Použi tento nástroj pri otázkach "
"o ZP Wiki, študentoch, autoroch, "
"záverečných prácach, témach, rokoch, "
"projektoch alebo dokumentoch. "
"Nástroj vykoná hybridné FTS5 a "
"embeddingové vyhľadávanie a pripraví "
"zdrojovo podložený RAG kontext."
),
dependencies=[
Depends(
require_search_api_key
)
],
)
def rag(
request: RagRequest,
) -> dict[str, Any]:
try:
response = build_rag_context(
DB_FILE,
request.query,
limit=request.limit,
published_only=(
request.published_only
),
max_per_document=(
request.max_per_document
),
)
except FileNotFoundError as error:
raise HTTPException(
status_code=500,
detail=str(
error
),
) from error
except ValueError as error:
raise HTTPException(
status_code=400,
detail=str(
error
),
) from error
except RuntimeError as error:
raise HTTPException(
status_code=500,
detail=str(
error
),
) from error
return response
@app.post(
"/search",
dependencies=[
Depends(require_search_api_key)
Depends(
require_search_api_key
)
],
include_in_schema=False,
)
def search(
request: SearchRequest,
@ -324,28 +590,42 @@ def search(
except FileNotFoundError as error:
raise HTTPException(
status_code=500,
detail=str(error),
detail=str(
error
),
) from error
except ValueError as error:
raise HTTPException(
status_code=400,
detail=str(error),
detail=str(
error
),
) from error
except RuntimeError as error:
raise HTTPException(
status_code=500,
detail=str(error),
detail=str(
error
),
) from error
results = response["results"]
results = response[
"results"
]
return {
"query": request.query,
"engine": response["engine"],
"strategies": response["strategies"],
"count": len(results),
"engine": response[
"engine"
],
"strategies": response[
"strategies"
],
"count": len(
results
),
"results": results,
}
@ -353,42 +633,58 @@ def search(
@app.post(
"/sync",
dependencies=[
Depends(require_sync_api_key)
Depends(
require_sync_api_key
)
],
include_in_schema=False,
)
def sync(
request: SyncRequest,
) -> dict[str, Any]:
try:
result = rebuild_index(
pull_git=request.pull_git
pull_git=(
request.pull_git
)
)
except ReindexInProgressError as error:
raise HTTPException(
status_code=409,
detail=str(error),
detail=str(
error
),
) from error
except RuntimeError as error:
raise HTTPException(
status_code=500,
detail=str(error),
detail=str(
error
),
) from error
return {
"status": "ok",
"pull_git": request.pull_git,
"duration_seconds": (
result["duration_seconds"]
"pull_git": (
request.pull_git
),
"counts": result["counts"],
"duration_seconds": (
result[
"duration_seconds"
]
),
"counts": result[
"counts"
],
}
@app.post(
"/webhook/gitea",
response_model=None,
include_in_schema=False,
)
async def gitea_webhook(
request: Request,
@ -416,12 +712,16 @@ async def gitea_webhook(
status_code=(
status.HTTP_401_UNAUTHORIZED
),
detail="Neplatný webhook podpis",
detail=(
"Neplatný webhook podpis"
),
)
try:
payload = json.loads(
raw_body.decode("utf-8")
raw_body.decode(
"utf-8"
)
)
except (
@ -436,7 +736,10 @@ async def gitea_webhook(
),
) from error
if not isinstance(payload, dict):
if not isinstance(
payload,
dict,
):
raise HTTPException(
status_code=400,
detail=(
@ -454,15 +757,22 @@ async def gitea_webhook(
),
)
if x_gitea_event.casefold() != "push":
if (
x_gitea_event.casefold()
!= "push"
):
return JSONResponse(
status_code=(
status.HTTP_202_ACCEPTED
),
content={
"status": "ignored",
"reason": "unsupported_event",
"event": x_gitea_event,
"reason": (
"unsupported_event"
),
"event": (
x_gitea_event
),
},
)
@ -508,22 +818,36 @@ async def gitea_webhook(
except ReindexInProgressError as error:
raise HTTPException(
status_code=409,
detail=str(error),
detail=str(
error
),
) from error
except RuntimeError as error:
raise HTTPException(
status_code=500,
detail=str(error),
detail=str(
error
),
) from error
return {
"status": "ok",
"event": x_gitea_event,
"repository": repository_name,
"verified_by": "hmac_sha256",
"duration_seconds": (
result["duration_seconds"]
"event": (
x_gitea_event
),
"counts": result["counts"],
"repository": (
repository_name
),
"verified_by": (
"hmac_sha256"
),
"duration_seconds": (
result[
"duration_seconds"
]
),
"counts": result[
"counts"
],
}

5
docker-compose.yml
View File

@ -14,9 +14,14 @@ services:
CHUNK_OVERLAP_TOKENS: "70"
CHUNK_MIN_TOKENS: "80"
CHUNK_TOKEN_ENCODING: cl100k_base
HF_HOME: /cache/huggingface
volumes:
- ./data:/app/data
- ../zpwiki:/zpwiki
- hf-cache:/cache/huggingface
restart: unless-stopped
volumes:
hf-cache:

40
et -a
View File

@ -1,40 +0,0 @@
README.md:66:WEBHOOK_SECRET=<náhodná hodnota s minimálne 32 znakmi>
README.md:67:SYNC_API_KEY=<iná náhodná hodnota s minimálne 32 znakmi>
README.md:68:SEARCH_API_KEY=<ďalšia náhodná hodnota s minimálne 32 znakmi>
README.md:162: -H "X-API-Key: $SEARCH_API_KEY" \
README.md:182: -H "X-API-Key: $SYNC_API_KEY" \
app/main.py:41:SEARCH_API_KEY_HEADER = "X-API-Key"
app/main.py:42:SYNC_API_KEY_HEADER = "X-API-Key"
app/main.py:45:search_api_key_scheme = APIKeyHeader(
app/main.py:46: name=SEARCH_API_KEY_HEADER,
app/main.py:51:sync_api_key_scheme = APIKeyHeader(
app/main.py:52: name=SYNC_API_KEY_HEADER,
app/main.py:135: validate_secret("WEBHOOK_SECRET")
app/main.py:136: validate_secret("SYNC_API_KEY")
app/main.py:137: validate_secret("SEARCH_API_KEY")
app/main.py:158:def require_search_api_key(
app/main.py:160: search_api_key_scheme
app/main.py:164: "SEARCH_API_KEY"
app/main.py:183:def require_sync_api_key(
app/main.py:185: sync_api_key_scheme
app/main.py:189: "SYNC_API_KEY"
app/main.py:293: "WEBHOOK_SECRET",
app/main.py:294: "SYNC_API_KEY",
app/main.py:295: "SEARCH_API_KEY",
app/main.py:305: Depends(require_search_api_key)
app/main.py:356: Depends(require_sync_api_key)
app/main.py:407: "WEBHOOK_SECRET"
test/conftest.py:21: "WEBHOOK_SECRET",
test/conftest.py:26: "SYNC_API_KEY",
test/conftest.py:31: "SEARCH_API_KEY",
test/test_api.py:14:WEBHOOK_SECRET = "w" * 64
test/test_api.py:15:SYNC_API_KEY = "s" * 64
test/test_api.py:16:SEARCH_API_KEY = "a" * 64
test/test_api.py:34: WEBHOOK_SECRET.encode("utf-8"),
test/test_api.py:49: monkeypatch.delenv("WEBHOOK_SECRET")
test/test_api.py:53: match="WEBHOOK_SECRET",
test/test_api.py:63: "SYNC_API_KEY",
test/test_api.py:111: "X-API-Key": SEARCH_API_KEY,
test/test_api.py:133: "X-API-Key": SEARCH_API_KEY,
test/test_api.py:216: "X-API-Key": SYNC_API_KEY,
test/test_api.py:245: "X-API-Key": SYNC_API_KEY,

268
scripts/rag_utils.py Normal file
View File

@ -0,0 +1,268 @@
from __future__ import annotations
from pathlib import Path
from typing import Any
from scripts.search_utils import search_database
RAG_INSTRUCTIONS = [
(
"Odpovedaj výhradne podľa informácií vrátených "
"v poli context a sources zo ZP Wiki."
),
(
"Nepoužívaj vlastnú pamäť modelu, domnienky ani "
"informácie, ktoré sa v poskytnutých zdrojoch nenachádzajú."
),
(
"Dôsledne rozlišuj názov dokumentu, autora dokumentu, "
"osobu, o ktorej dokument pojednáva, rok začiatku štúdia "
"a rok záverečnej práce."
),
(
"Rok začiatku štúdia nie je automaticky rokom "
"záverečnej práce."
),
(
"Názov študentskej stránky nie je automaticky názvom "
"záverečnej práce."
),
(
"Autor dokumentu nemusí byť osoba, o ktorej dokument "
"pojednáva."
),
(
"Ak zdroje obsahujú viac možných interpretácií, "
"stručne vysvetli rozdiel a nevytváraj jednoznačný "
"záver bez opory v zdrojoch."
),
(
"Ak odpoveď nemožno zo zdrojov spoľahlivo určiť, povedz: "
"V dostupných dokumentoch ZP Wiki sa túto informáciu "
"nepodarilo spoľahlivo nájsť."
),
(
"Odpovedaj stručne, prirodzene a vetne po slovensky. "
"Pri jednoduchej otázke zvyčajne stačí jedna alebo dve vety."
),
(
"Nepoužívaj odrážky, tabuľky, tučné písmo ani iné "
"Markdown formátovanie pri jednoduchej faktickej odpovedi."
),
(
"Odpoveď formuluj prirodzenou vetou. Napríklad: "
"'Ján Holp vypracovával diplomovú prácu v roku 2021.'"
),
(
"Dodržuj prirodzené medzery medzi slovami a číslami. "
"Píš napríklad 'v roku 2021' a 'bol 2016'. "
"Nikdy nepíš 'v roku2021', 'roku2021', 'bol2016' "
"ani podobne spojené výrazy."
),
(
"Interné označenia zdrojov S1, S2, S3 a podobne slúžia "
"iba na rozlíšenie vstupných zdrojov. "
"Nevypisuj ich v konečnej odpovedi."
),
(
"V konečnej odpovedi nevypisuj interné retrieval údaje, "
"ako sú fts_rank, vector_rank, vector_score, "
"hybrid_score alebo match_strategy."
),
(
"Na konci odpovede uveď iba source_url zdrojov, "
"z ktorých odpoveď skutočne vychádza."
),
(
"Pri jednom použitom zdroji po hlavnej odpovedi "
"uveď samostatný riadok vo formáte "
"'Zdroj: <source_url>'."
),
(
"Pri viacerých použitých zdrojoch napíš 'Zdroje:' "
"a každý source_url uveď na samostatnom riadku."
),
(
"Medzi hlavnou odpoveďou a riadkom so zdrojom "
"ponechaj prázdny riadok."
),
]
ANSWER_FORMAT = {
"language": "slovak",
"style": "stručný, prirodzený a vetný text",
"internal_source_ids_visible": False,
"source_section": True,
"template": (
"<stručná odpoveď v jednej alebo dvoch vetách>\n\n"
"Zdroj: <source_url>"
),
"single_source_template": (
"<stručná odpoveď v jednej alebo dvoch vetách>\n\n"
"Zdroj: <source_url>"
),
"multiple_sources_template": (
"<stručná odpoveď v prirodzených vetách>\n\n"
"Zdroje:\n"
"<source_url_1>\n"
"<source_url_2>"
),
}
def build_source(
result: dict[str, Any],
number: int,
) -> dict[str, Any]:
source_id = f"S{number}"
return {
"source_id": source_id,
"title": result.get("title"),
"author": result.get("author"),
"document_path": result.get("document_path"),
"source_url": result.get("source_url"),
"published": result.get("published"),
"section": result.get(
"heading_paths",
[],
),
"text": result.get(
"text",
"",
),
"retrieval": {
"match_strategy": result.get(
"match_strategy"
),
"fts_rank": result.get(
"fts_rank"
),
"vector_rank": result.get(
"vector_rank"
),
"vector_score": result.get(
"vector_score"
),
"hybrid_score": result.get(
"hybrid_score"
),
},
}
def build_context_text(
sources: list[dict[str, Any]],
) -> str:
if not sources:
return (
"V dostupných dokumentoch ZP Wiki "
"sa k dotazu nenašli relevantné zdroje."
)
blocks: list[str] = []
for source in sources:
source_id = source[
"source_id"
]
title = (
source.get("title")
or "Neuvedené"
)
author = (
source.get("author")
or "Neuvedený"
)
source_url = (
source.get("source_url")
or "Neuvedené"
)
text = (
source.get("text")
or ""
)
block = (
f"ZDROJ {source_id}\n"
f"Názov dokumentu: {title}\n"
f"Autor dokumentu: {author}\n"
f"Source URL: {source_url}\n"
"\n"
f"{text}"
)
blocks.append(
block
)
return (
"\n\n"
"--------------------"
"\n\n"
).join(
blocks
)
def build_rag_context(
db_path: Path,
query: str,
*,
limit: int = 5,
published_only: bool = False,
max_per_document: int = 1,
) -> dict[str, Any]:
response = search_database(
db_path,
query,
limit,
published_only=published_only,
max_per_document=max_per_document,
)
results = response[
"results"
]
sources = [
build_source(
result,
index,
)
for index, result in enumerate(
results,
start=1,
)
]
context = build_context_text(
sources
)
return {
"query": query,
"engine": response[
"engine"
],
"strategies": response[
"strategies"
],
"source_count": len(
sources
),
"instructions": (
RAG_INSTRUCTIONS
),
"answer_format": (
ANSWER_FORMAT
),
"context": context,
"sources": sources,
}

40
tatus
View File

@ -1,40 +0,0 @@
README.md:66:WEBHOOK_SECRET=<náhodná hodnota s minimálne 32 znakmi>
README.md:67:SYNC_API_KEY=<iná náhodná hodnota s minimálne 32 znakmi>
README.md:68:SEARCH_API_KEY=<ďalšia náhodná hodnota s minimálne 32 znakmi>
README.md:162: -H "X-API-Key: $SEARCH_API_KEY" \
README.md:182: -H "X-API-Key: $SYNC_API_KEY" \
app/main.py:41:SEARCH_API_KEY_HEADER = "X-API-Key"
app/main.py:42:SYNC_API_KEY_HEADER = "X-API-Key"
app/main.py:45:search_api_key_scheme = APIKeyHeader(
app/main.py:46: name=SEARCH_API_KEY_HEADER,
app/main.py:51:sync_api_key_scheme = APIKeyHeader(
app/main.py:52: name=SYNC_API_KEY_HEADER,
app/main.py:135: validate_secret("WEBHOOK_SECRET")
app/main.py:136: validate_secret("SYNC_API_KEY")
app/main.py:137: validate_secret("SEARCH_API_KEY")
app/main.py:158:def require_search_api_key(
app/main.py:160: search_api_key_scheme
app/main.py:164: "SEARCH_API_KEY"
app/main.py:183:def require_sync_api_key(
app/main.py:185: sync_api_key_scheme
app/main.py:189: "SYNC_API_KEY"
app/main.py:293: "WEBHOOK_SECRET",
app/main.py:294: "SYNC_API_KEY",
app/main.py:295: "SEARCH_API_KEY",
app/main.py:305: Depends(require_search_api_key)
app/main.py:356: Depends(require_sync_api_key)
app/main.py:407: "WEBHOOK_SECRET"
test/conftest.py:21: "WEBHOOK_SECRET",
test/conftest.py:26: "SYNC_API_KEY",
test/conftest.py:31: "SEARCH_API_KEY",
test/test_api.py:14:WEBHOOK_SECRET = "w" * 64
test/test_api.py:15:SYNC_API_KEY = "s" * 64
test/test_api.py:16:SEARCH_API_KEY = "a" * 64
test/test_api.py:34: WEBHOOK_SECRET.encode("utf-8"),
test/test_api.py:49: monkeypatch.delenv("WEBHOOK_SECRET")
test/test_api.py:53: match="WEBHOOK_SECRET",
test/test_api.py:63: "SYNC_API_KEY",
test/test_api.py:111: "X-API-Key": SEARCH_API_KEY,
test/test_api.py:133: "X-API-Key": SEARCH_API_KEY,
test/test_api.py:216: "X-API-Key": SYNC_API_KEY,
test/test_api.py:245: "X-API-Key": SYNC_API_KEY,

2
test/test_api.py
View File

@ -82,7 +82,7 @@ def test_health_endpoint(
payload = response.json()
assert payload["status"] == "ok"
assert payload["search_engine"] == "sqlite_fts5"
assert payload["search_engine"] == "hybrid_fts5_embeddings"
assert payload["security_configured"] is True

500
test/test_rag.py Normal file
View File

@ -0,0 +1,500 @@
from __future__ import annotations
from pathlib import Path
from typing import Any
import pytest
from fastapi.testclient import TestClient
import app.main as main_module
import scripts.rag_utils as rag_utils
from scripts.rag_utils import (
ANSWER_FORMAT,
RAG_INSTRUCTIONS,
build_context_text,
build_rag_context,
build_source,
)
SEARCH_API_KEY = "a" * 64
@pytest.fixture
def client(
security_environment,
) -> TestClient:
return TestClient(
main_module.app
)
def sample_result() -> dict[str, Any]:
return {
"chunk_id": (
"pages/students/2016/"
"jan_holp/README.md::chunk-0"
),
"document_path": (
"pages/students/2016/"
"jan_holp/README.md"
),
"title": "Ján Holp",
"author": "Daniel Hladek",
"published": True,
"heading_paths": [
[
"Ján Holp",
"Diplomová práca 2021",
],
],
"text": (
"Dokument: Ján Holp\n"
"Sekcia: Diplomová práca 2021\n\n"
"Rok začiatku štúdia: 2016\n"
"Názov diplomovej práce: "
"Systém získavania informácií "
"v slovenskom jazyku."
),
"source_url": (
"https://zp.kemt.fei.tuke.sk/"
"students/2016/jan_holp"
),
"match_strategy": "any_term",
"fts_rank": 11,
"vector_rank": 1,
"vector_score": 0.863072,
"hybrid_score": 0.02811129,
}
def test_build_source() -> None:
result = sample_result()
source = build_source(
result,
1,
)
assert source["source_id"] == "S1"
assert source["title"] == "Ján Holp"
assert source["author"] == "Daniel Hladek"
assert source["source_url"] == (
"https://zp.kemt.fei.tuke.sk/"
"students/2016/jan_holp"
)
assert source["published"] is True
assert source["retrieval"] == {
"match_strategy": "any_term",
"fts_rank": 11,
"vector_rank": 1,
"vector_score": 0.863072,
"hybrid_score": 0.02811129,
}
# Interná identifikácia zdroja nemá byť
# používateľská citation hodnota.
assert "citation" not in source
def test_build_context_text() -> None:
source = build_source(
sample_result(),
1,
)
context = build_context_text(
[source]
)
assert "ZDROJ S1" in context
assert "Názov dokumentu: Ján Holp" in context
assert (
"Autor dokumentu: Daniel Hladek"
in context
)
assert (
"Sekcia: Diplomová práca 2021"
in context
)
assert (
"Rok začiatku štúdia: 2016"
in context
)
assert (
"https://zp.kemt.fei.tuke.sk/"
"students/2016/jan_holp"
in context
)
def test_build_context_text_empty() -> None:
context = build_context_text([])
assert (
"nenašli relevantné zdroje"
in context
)
def test_rag_instructions_require_grounding() -> None:
instructions = " ".join(
RAG_INSTRUCTIONS
)
assert (
"výhradne podľa informácií"
in instructions
)
assert (
"v roku 2021"
in instructions
)
assert (
"roku2021"
in instructions
)
assert (
"Nevypisuj ich v konečnej odpovedi"
in instructions
)
assert "source_url" in instructions
def test_answer_format() -> None:
assert (
ANSWER_FORMAT[
"internal_source_ids_visible"
]
is False
)
assert (
ANSWER_FORMAT[
"source_section"
]
is True
)
assert (
"<source_url>"
in ANSWER_FORMAT["template"]
)
def test_build_rag_context(
monkeypatch: pytest.MonkeyPatch,
) -> None:
captured: dict[str, Any] = {}
def fake_search_database(
db_path: Path,
query: str,
limit: int,
*,
published_only: bool,
max_per_document: int,
) -> dict[str, Any]:
captured["db_path"] = db_path
captured["query"] = query
captured["limit"] = limit
captured["published_only"] = (
published_only
)
captured["max_per_document"] = (
max_per_document
)
return {
"engine": (
"hybrid_fts5_embeddings"
),
"strategies": [
"any_term"
],
"results": [
sample_result()
],
}
monkeypatch.setattr(
rag_utils,
"search_database",
fake_search_database,
)
db_path = Path(
"/tmp/test.sqlite"
)
response = build_rag_context(
db_path,
(
"V akom roku robil Ján Holp "
"diplomovú prácu?"
),
limit=5,
published_only=True,
max_per_document=1,
)
assert captured == {
"db_path": db_path,
"query": (
"V akom roku robil Ján Holp "
"diplomovú prácu?"
),
"limit": 5,
"published_only": True,
"max_per_document": 1,
}
assert response["engine"] == (
"hybrid_fts5_embeddings"
)
assert response["strategies"] == [
"any_term"
]
assert response["source_count"] == 1
assert (
response["sources"][0]["title"]
== "Ján Holp"
)
assert (
"Diplomová práca 2021"
in response["context"]
)
assert (
response["answer_format"][
"internal_source_ids_visible"
]
is False
)
def test_build_rag_context_without_results(
monkeypatch: pytest.MonkeyPatch,
) -> None:
def fake_search_database(
db_path: Path,
query: str,
limit: int,
*,
published_only: bool,
max_per_document: int,
) -> dict[str, Any]:
return {
"engine": (
"hybrid_fts5_embeddings"
),
"strategies": [],
"results": [],
}
monkeypatch.setattr(
rag_utils,
"search_database",
fake_search_database,
)
response = build_rag_context(
Path("/tmp/test.sqlite"),
"neexistujúca téma",
)
assert response["source_count"] == 0
assert response["sources"] == []
assert (
"nenašli relevantné zdroje"
in response["context"]
)
def test_rag_endpoint(
client: TestClient,
monkeypatch: pytest.MonkeyPatch,
) -> None:
expected = {
"query": "Ján Holp",
"engine": (
"hybrid_fts5_embeddings"
),
"strategies": [
"all_terms"
],
"source_count": 1,
"instructions": (
RAG_INSTRUCTIONS
),
"answer_format": (
ANSWER_FORMAT
),
"context": (
"ZDROJ S1\n"
"Názov dokumentu: Ján Holp"
),
"sources": [
{
"source_id": "S1",
"title": "Ján Holp",
"source_url": (
"https://example.test/"
"jan_holp"
),
},
],
}
def fake_build_rag_context(
db_path: Path,
query: str,
*,
limit: int,
published_only: bool,
max_per_document: int,
) -> dict[str, Any]:
assert query == "Ján Holp"
assert limit == 5
assert published_only is False
assert max_per_document == 1
return expected
monkeypatch.setattr(
main_module,
"build_rag_context",
fake_build_rag_context,
)
response = client.post(
"/rag",
headers={
"X-API-Key": (
SEARCH_API_KEY
),
},
json={
"query": "Ján Holp",
},
)
assert response.status_code == 200
assert response.json() == expected
def test_rag_endpoint_with_bearer(
client: TestClient,
monkeypatch: pytest.MonkeyPatch,
) -> None:
def fake_build_rag_context(
db_path: Path,
query: str,
*,
limit: int,
published_only: bool,
max_per_document: int,
) -> dict[str, Any]:
return {
"query": query,
"engine": (
"hybrid_fts5_embeddings"
),
"strategies": [],
"source_count": 0,
"instructions": (
RAG_INSTRUCTIONS
),
"answer_format": (
ANSWER_FORMAT
),
"context": "bez výsledkov",
"sources": [],
}
monkeypatch.setattr(
main_module,
"build_rag_context",
fake_build_rag_context,
)
response = client.post(
"/rag",
headers={
"Authorization": (
f"Bearer {SEARCH_API_KEY}"
),
},
json={
"query": "test",
},
)
assert response.status_code == 200
def test_rag_endpoint_without_api_key(
client: TestClient,
) -> None:
response = client.post(
"/rag",
json={
"query": "Ján Holp",
},
)
assert response.status_code == 401
def test_rag_endpoint_empty_query(
client: TestClient,
) -> None:
response = client.post(
"/rag",
headers={
"X-API-Key": (
SEARCH_API_KEY
),
},
json={
"query": "",
},
)
assert response.status_code == 422
def test_openapi_exposes_rag_only(
client: TestClient,
) -> None:
response = client.get(
"/openapi.json"
)
assert response.status_code == 200
paths = response.json()[
"paths"
]
assert "/rag" in paths
assert "/search" not in paths
assert "/sync" not in paths
assert "/health" not in paths
assert "/webhook/gitea" not in paths