a63e645513
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Failing after 1m36s
CD Pipeline / build-and-deploy (push) Has been skipped
CD Pipeline / post-deploy-checks (push) Has been skipped
1425 lines
52 KiB
Python
1425 lines
52 KiB
Python
"""
|
||
Knowledge Service - 業務邏輯層
|
||
===============================
|
||
Knowledge Base Phase 1: CRUD + 狀態流轉 + 搜尋
|
||
|
||
建立時間: 2026-04-02 (台北時區)
|
||
建立者: Claude Code (Knowledge Base Phase 1)
|
||
|
||
遵循 leWOOOgo 積木化原則:
|
||
- Service 層封裝業務邏輯
|
||
- 依賴 IKnowledgeRepository Protocol
|
||
- Router 層禁止直接存取 DB
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
from collections.abc import Mapping
|
||
from typing import Any, cast
|
||
|
||
import structlog
|
||
|
||
from src.core.config import settings
|
||
from src.core.context import get_current_project_id
|
||
from src.db.base import get_db_context
|
||
from src.models.knowledge import (
|
||
CategoryCount,
|
||
EntrySource,
|
||
EntryStatus,
|
||
EntryType,
|
||
KnowledgeAssetTaxonomyCount,
|
||
KnowledgeEntry,
|
||
KnowledgeEntryCreate,
|
||
KnowledgeEntryUpdate,
|
||
KnowledgeListResponse,
|
||
)
|
||
from src.repositories.interfaces import IKnowledgeRepository
|
||
from src.repositories.knowledge_repository import KnowledgeDBRepository
|
||
from src.services.embedding_service import OllamaEmbeddingService
|
||
from src.services.model_registry import get_model as _get_model
|
||
from src.utils.timezone import now_taipei
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
_DEGRADED_CATEGORY_FALLBACKS = (
|
||
"project",
|
||
"product",
|
||
"website",
|
||
"service",
|
||
"package",
|
||
"tool",
|
||
"log",
|
||
"alert",
|
||
"playbook",
|
||
"rag",
|
||
"mcp",
|
||
"schedule",
|
||
"general",
|
||
)
|
||
|
||
_ASSET_TAXONOMY_FALLBACKS = _DEGRADED_CATEGORY_FALLBACKS[:-1]
|
||
_PRIMARY_KM_LIST_TIMEOUT_SECONDS = 2.0
|
||
_PRIMARY_KM_LIST_RETRY_TIMEOUT_SECONDS = 1.0
|
||
_PRIMARY_KM_RETRY_DELAY_SECONDS = 0.1
|
||
_PRIMARY_KM_SIDE_READ_TIMEOUT_SECONDS = 1.2
|
||
_PRIMARY_KM_TAXONOMY_TIMEOUT_SECONDS = 1.5
|
||
_PRIMARY_KM_DIRECT_CONNECT_TIMEOUT_SECONDS = 1.5
|
||
_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS = 3.0
|
||
_PRIMARY_KM_DIRECT_CLOSE_TIMEOUT_SECONDS = 0.5
|
||
_PRIMARY_KM_DIRECT_STATEMENT_TIMEOUT_MS = 3500
|
||
_ASSET_TAXONOMY_CATEGORY_HINTS: dict[str, tuple[str, ...]] = {
|
||
"service": (
|
||
"infrastructure",
|
||
"application",
|
||
"ai_system",
|
||
"database",
|
||
"host_resource",
|
||
"kubernetes",
|
||
"alert_handling",
|
||
),
|
||
"tool": ("devops_tool", "AI自動化/Ansible受控修復"),
|
||
"alert": ("alert_handling",),
|
||
"playbook": ("auto_repair",),
|
||
}
|
||
_ASSET_TAXONOMY_TERMS: dict[str, tuple[str, ...]] = {
|
||
"project": (
|
||
"project",
|
||
"repo",
|
||
"repository",
|
||
"gitea",
|
||
"branch",
|
||
"workflow",
|
||
"source_control",
|
||
),
|
||
"product": (
|
||
"product",
|
||
"awoooi",
|
||
"awooop",
|
||
"iwooos",
|
||
"stockplatform",
|
||
"momo",
|
||
"awooogo",
|
||
"agent-bounty",
|
||
"agent_bounty",
|
||
"tsenyang",
|
||
"vibework",
|
||
"2026fifa",
|
||
),
|
||
"website": ("website", "site", "nginx", "ssl", "domain", "route", "frontend"),
|
||
"service": ("service", "daemon", "api", "worker", "runtime", "container", "pod"),
|
||
"package": (
|
||
"package",
|
||
"dependency",
|
||
"npm",
|
||
"pnpm",
|
||
"node",
|
||
"python",
|
||
"pip",
|
||
"prisma",
|
||
"next",
|
||
"library",
|
||
),
|
||
"tool": (
|
||
"tool",
|
||
"ansible",
|
||
"mcp",
|
||
"playbook",
|
||
"telegram",
|
||
"wazuh",
|
||
"kali",
|
||
"sentry",
|
||
"signoz",
|
||
"runner",
|
||
),
|
||
"log": (
|
||
"log",
|
||
"logs",
|
||
"event",
|
||
"timeline",
|
||
"trace",
|
||
"audit",
|
||
"callback",
|
||
"conversation_event",
|
||
"telemetry",
|
||
),
|
||
"alert": (
|
||
"alert",
|
||
"telegram",
|
||
"sentry",
|
||
"signoz",
|
||
"notification",
|
||
"warning",
|
||
"critical",
|
||
),
|
||
"playbook": ("playbook", "runbook", "sop"),
|
||
"rag": ("rag", "vector", "embedding", "semantic", "retrieval"),
|
||
"mcp": ("mcp", "connector", "gateway", "tool integration", "tool-integration"),
|
||
"schedule": ("schedule", "cron", "job", "worker", "patrol", "recurrence", "cadence"),
|
||
}
|
||
|
||
_SOURCE_BACKED_KNOWLEDGE_SPECS: tuple[dict[str, object], ...] = (
|
||
{
|
||
"id": "source-backed-project-awoooi",
|
||
"title": "AWOOOI Gitea source-to-runtime truth",
|
||
"category": "project",
|
||
"entry_type": EntryType.BEST_PRACTICE,
|
||
"content": (
|
||
"Source-backed KM readback for AWOOOI mainline: Gitea SSH is the "
|
||
"source of truth, deploy markers and public runtime readbacks remain "
|
||
"separate evidence layers, and GitHub is frozen."
|
||
),
|
||
"tags": ["project", "gitea", "deploy_marker", "source_control", "ai_automation"],
|
||
},
|
||
{
|
||
"id": "source-backed-product-awooop",
|
||
"title": "AwoooP AI controlled operations loop",
|
||
"category": "product",
|
||
"entry_type": EntryType.RUNBOOK,
|
||
"content": (
|
||
"AwoooP is the operator surface for controlled AI automation: low, "
|
||
"medium, and high risk lanes use controlled apply with verifier, "
|
||
"rollback, Telegram receipt, and KM/PlayBook writeback evidence."
|
||
),
|
||
"tags": ["product", "awooop", "controlled_apply", "ai_loop_agent"],
|
||
},
|
||
{
|
||
"id": "source-backed-website-awoooi-public",
|
||
"title": "awoooi.wooo.work public runtime surfaces",
|
||
"category": "website",
|
||
"entry_type": EntryType.BEST_PRACTICE,
|
||
"content": (
|
||
"Public website readbacks must prove the deployed API/page behavior "
|
||
"instead of treating source tests as production truth."
|
||
),
|
||
"tags": ["website", "awoooi.wooo.work", "route", "frontend", "readback"],
|
||
},
|
||
{
|
||
"id": "source-backed-service-telegram-alert-receipts",
|
||
"title": "Telegram alert receipt services",
|
||
"category": "service",
|
||
"entry_type": EntryType.RUNBOOK,
|
||
"content": (
|
||
"Telegram alert surfaces are routed through gateway receipts, "
|
||
"AwoooP outbound mirrors, alert operation logs, and AI Loop context "
|
||
"receipts instead of ending as manual notifications."
|
||
),
|
||
"tags": ["service", "telegram", "alert", "receipt", "awooop"],
|
||
},
|
||
{
|
||
"id": "source-backed-package-workspace-governance",
|
||
"title": "Workspace package and dependency governance",
|
||
"category": "package",
|
||
"entry_type": EntryType.BEST_PRACTICE,
|
||
"content": (
|
||
"Workspace packages, Python services, pnpm workspaces, and generated "
|
||
"readbacks are classified as package evidence for AI automation."
|
||
),
|
||
"tags": ["package", "dependency", "pnpm", "python", "typescript"],
|
||
},
|
||
{
|
||
"id": "source-backed-tool-mcp-runner-gateway",
|
||
"title": "MCP, runner, and Telegram gateway tools",
|
||
"category": "tool",
|
||
"entry_type": EntryType.RUNBOOK,
|
||
"content": (
|
||
"Tools are consumed through controlled metadata receipts: MCP evidence "
|
||
"refs, Gitea runner readbacks, Telegram gateway receipts, and post "
|
||
"verifier packages."
|
||
),
|
||
"tags": ["tool", "mcp", "runner", "telegram", "verifier"],
|
||
},
|
||
{
|
||
"id": "source-backed-log-intelligence-taxonomy",
|
||
"title": "LOG intelligence label taxonomy",
|
||
"category": "log",
|
||
"entry_type": EntryType.BEST_PRACTICE,
|
||
"content": (
|
||
"Logs are grouped by project, product, website, service, package, "
|
||
"tool, alert, playbook, RAG, MCP, and schedule labels so AI Agent "
|
||
"can reuse them for decisions and learning."
|
||
),
|
||
"tags": ["log", "telemetry", "trace", "audit", "label_taxonomy"],
|
||
},
|
||
{
|
||
"id": "source-backed-alert-telegram-monitoring-coverage",
|
||
"title": "Telegram monitoring AI automation coverage",
|
||
"category": "alert",
|
||
"entry_type": EntryType.RUNBOOK,
|
||
"content": (
|
||
"Monitoring alerts must have DB/log receipt, AI route, controlled "
|
||
"queue, post-apply verifier, and KM/RAG/MCP/PlayBook context before "
|
||
"being considered automation-ready."
|
||
),
|
||
"tags": ["alert", "telegram", "monitoring", "controlled_queue", "ai_agent"],
|
||
},
|
||
{
|
||
"id": "source-backed-playbook-controlled-apply",
|
||
"title": "Controlled PlayBook apply and verifier loop",
|
||
"category": "playbook",
|
||
"entry_type": EntryType.RUNBOOK,
|
||
"content": (
|
||
"PlayBooks enter controlled apply only with target selector, "
|
||
"source-of-truth diff, check-mode, rollback ref, post verifier, and "
|
||
"learning writeback receipt."
|
||
),
|
||
"tags": ["playbook", "runbook", "sop", "controlled_apply"],
|
||
"related_playbook_id": "playbook://awoooi/controlled-apply/verifier-loop",
|
||
},
|
||
{
|
||
"id": "source-backed-rag-km-retrieval-context",
|
||
"title": "KM / RAG retrieval context",
|
||
"category": "rag",
|
||
"entry_type": EntryType.BEST_PRACTICE,
|
||
"content": (
|
||
"RAG context must use public-safe metadata refs and source-backed "
|
||
"knowledge receipts; raw sessions, secrets, and unredacted payloads "
|
||
"are not learning inputs."
|
||
),
|
||
"tags": ["rag", "km", "embedding", "retrieval", "redaction"],
|
||
},
|
||
{
|
||
"id": "source-backed-mcp-tool-audit-context",
|
||
"title": "MCP evidence refs and tool audit context",
|
||
"category": "mcp",
|
||
"entry_type": EntryType.BEST_PRACTICE,
|
||
"content": (
|
||
"MCP-related evidence is stored as redacted metadata references for "
|
||
"tool audit and AI decision context; tool execution remains gated by "
|
||
"controlled routes."
|
||
),
|
||
"tags": ["mcp", "connector", "tool_integration", "audit", "redaction"],
|
||
},
|
||
{
|
||
"id": "source-backed-schedule-report-monitoring",
|
||
"title": "Report and monitoring schedules",
|
||
"category": "schedule",
|
||
"entry_type": EntryType.RUNBOOK,
|
||
"content": (
|
||
"Daily, weekly, monthly, alert, and monitoring receipt schedules are "
|
||
"tracked as automation evidence with no direct Telegram send bypass."
|
||
),
|
||
"tags": ["schedule", "cron", "cadence", "worker", "telegram"],
|
||
},
|
||
{
|
||
"id": "source-backed-general-km-readback-contract",
|
||
"title": "Source-backed KM readback contract",
|
||
"category": "general",
|
||
"entry_type": EntryType.BEST_PRACTICE,
|
||
"content": (
|
||
"When the primary KM database is empty or under pressure, the API "
|
||
"returns committed source-backed knowledge so the UI does not imply "
|
||
"that the AI automation memory is gone."
|
||
),
|
||
"tags": ["general", "knowledge_readback", "source_backed", "no_false_zero"],
|
||
},
|
||
)
|
||
|
||
|
||
def _classify_knowledge_readback_degraded_reason(reason: str) -> str:
|
||
normalized = reason.lower()
|
||
if (
|
||
not normalized
|
||
or "pool" in normalized
|
||
or "timeout" in normalized
|
||
or "timed out" in normalized
|
||
or "timeouterror" in normalized
|
||
):
|
||
return "primary_km_db_timeout_or_pool_exhausted"
|
||
if "missing tenant context" in normalized or "project_id" in normalized or "unauthorized" in normalized:
|
||
return "primary_km_project_context_missing"
|
||
if "undefinedcolumn" in normalized or "does not exist" in normalized or "missing column" in normalized:
|
||
return "primary_km_schema_mismatch"
|
||
if "validation" in normalized or "pydantic" in normalized:
|
||
return "primary_km_legacy_row_validation"
|
||
return "primary_km_readback_exception"
|
||
|
||
|
||
def _knowledge_readback_exception_reason(exc: BaseException) -> str:
|
||
if isinstance(exc, TimeoutError):
|
||
return "TimeoutError"
|
||
return str(exc) or type(exc).__name__
|
||
|
||
|
||
def _current_knowledge_project_id() -> str | None:
|
||
project_id = get_current_project_id()
|
||
return str(project_id).strip() if project_id else None
|
||
|
||
|
||
def _normalize_direct_tags(value: object) -> list[str]:
|
||
if value is None:
|
||
return []
|
||
if isinstance(value, list | tuple | set):
|
||
return [str(tag).strip() for tag in value if str(tag).strip()]
|
||
if isinstance(value, str):
|
||
stripped = value.strip()
|
||
if not stripped:
|
||
return []
|
||
try:
|
||
parsed = json.loads(stripped)
|
||
except json.JSONDecodeError:
|
||
parsed = None
|
||
if isinstance(parsed, list):
|
||
return [str(tag).strip() for tag in parsed if str(tag).strip()]
|
||
return [stripped]
|
||
return [str(value).strip()] if str(value).strip() else []
|
||
|
||
|
||
def _direct_enum_member(value: object, enum_cls: type[Any], fallback: Any) -> Any:
|
||
raw = str(getattr(value, "value", value) or "").lower()
|
||
for member in enum_cls:
|
||
if raw in {member.value.lower(), member.name.lower()}:
|
||
return member
|
||
return fallback
|
||
|
||
|
||
def _direct_knowledge_entry_from_row(row: Mapping[str, Any]) -> KnowledgeEntry:
|
||
return KnowledgeEntry(
|
||
id=str(row.get("id")),
|
||
title=str(row.get("title") or "Untitled KM entry"),
|
||
content=str(row.get("content") or ""),
|
||
entry_type=_direct_enum_member(
|
||
row.get("entry_type"),
|
||
EntryType,
|
||
EntryType.BEST_PRACTICE,
|
||
),
|
||
category=str(row.get("category") or "general").strip() or "general",
|
||
tags=_normalize_direct_tags(row.get("tags")),
|
||
source=_direct_enum_member(
|
||
row.get("source"),
|
||
EntrySource,
|
||
EntrySource.AI_EXTRACTED,
|
||
),
|
||
status=_direct_enum_member(
|
||
row.get("status"),
|
||
EntryStatus,
|
||
EntryStatus.REVIEW,
|
||
),
|
||
related_incident_id=row.get("related_incident_id"),
|
||
related_playbook_id=row.get("related_playbook_id"),
|
||
related_approval_id=row.get("related_approval_id"),
|
||
path_type=row.get("path_type"),
|
||
symptoms_hash=row.get("symptoms_hash"),
|
||
view_count=int(row.get("view_count") or 0),
|
||
created_by=row.get("created_by"),
|
||
created_at=row.get("created_at") or now_taipei(),
|
||
updated_at=row.get("updated_at") or row.get("created_at") or now_taipei(),
|
||
)
|
||
|
||
|
||
def _direct_tag_needle(tag: str) -> str:
|
||
return f'"{tag.replace(chr(34), chr(92) + chr(34))}"'
|
||
|
||
|
||
def _direct_base_where(project_id: str) -> tuple[list[str], list[Any]]:
|
||
return [
|
||
"project_id = $1",
|
||
"lower(status::text) != 'archived'",
|
||
], [project_id]
|
||
|
||
|
||
def _append_direct_filter(
|
||
clauses: list[str],
|
||
params: list[Any],
|
||
*,
|
||
category: str | None = None,
|
||
entry_type: EntryType | None = None,
|
||
status: EntryStatus | None = None,
|
||
tags: list[str] | None = None,
|
||
q: str | None = None,
|
||
) -> None:
|
||
if category:
|
||
params.append(str(category).strip() or "general")
|
||
clauses.append(
|
||
"coalesce(nullif(trim(category), ''), 'general') = "
|
||
f"${len(params)}"
|
||
)
|
||
if entry_type:
|
||
params.append(entry_type.value)
|
||
clauses.append(f"lower(entry_type::text) = ${len(params)}")
|
||
if status:
|
||
params.append(status.value)
|
||
clauses.append(f"lower(status::text) = ${len(params)}")
|
||
for tag in tags or []:
|
||
params.append(_direct_tag_needle(tag))
|
||
clauses.append(f"strpos(tags::text, ${len(params)}) > 0")
|
||
if q:
|
||
params.append(f"%{q}%")
|
||
index = len(params)
|
||
clauses.append(
|
||
"("
|
||
f"title ILIKE ${index} OR "
|
||
f"content ILIKE ${index} OR "
|
||
f"tags::text ILIKE ${index} OR "
|
||
f"coalesce(nullif(trim(category), ''), 'general') ILIKE ${index}"
|
||
")"
|
||
)
|
||
|
||
|
||
def _direct_where_sql(clauses: list[str]) -> str:
|
||
return " WHERE " + " AND ".join(clauses)
|
||
|
||
|
||
def _sql_literal(value: str) -> str:
|
||
return "'" + value.replace("'", "''") + "'"
|
||
|
||
|
||
def _asset_taxonomy_direct_condition(key: str) -> str:
|
||
category_expr = "coalesce(nullif(trim(category), ''), 'general')"
|
||
category_text = f"lower({category_expr}::text)"
|
||
conditions: list[str] = []
|
||
category_hints = _ASSET_TAXONOMY_CATEGORY_HINTS.get(key, ())
|
||
if category_hints:
|
||
conditions.append(
|
||
f"{category_text} IN ("
|
||
+ ", ".join(_sql_literal(hint.lower()) for hint in category_hints)
|
||
+ ")"
|
||
)
|
||
for term in _ASSET_TAXONOMY_TERMS.get(key, ()):
|
||
pattern = _sql_literal(f"%{term.lower()}%")
|
||
conditions.extend(
|
||
[
|
||
f"lower(coalesce(title, '')) LIKE {pattern}",
|
||
f"lower(coalesce(content, '')) LIKE {pattern}",
|
||
f"lower(coalesce(tags::text, '')) LIKE {pattern}",
|
||
f"{category_text} LIKE {pattern}",
|
||
]
|
||
)
|
||
if key == "playbook":
|
||
conditions.append("related_playbook_id IS NOT NULL")
|
||
return " OR ".join(conditions) if conditions else f"{category_text} = {_sql_literal(key)}"
|
||
|
||
# =============================================================================
|
||
# Singleton
|
||
# =============================================================================
|
||
|
||
_knowledge_service: "KnowledgeService | None" = None
|
||
|
||
|
||
def build_knowledge_list_readback_degraded_response(
|
||
reason: str,
|
||
*,
|
||
category: str | None = None,
|
||
entry_type: EntryType | None = None,
|
||
status: EntryStatus | None = None,
|
||
tags: list[str] | None = None,
|
||
q: str | None = None,
|
||
limit: int = 20,
|
||
offset: int = 0,
|
||
readback_status: str = "source_backed_degraded",
|
||
) -> KnowledgeListResponse:
|
||
"""主 KM readback 失敗時回保守 payload，避免前端誤判成知識庫歸零。"""
|
||
entries = _filter_source_backed_entries(
|
||
category=category,
|
||
entry_type=entry_type,
|
||
status=status,
|
||
tags=tags,
|
||
q=q,
|
||
)
|
||
total = len(entries)
|
||
page = entries[offset: offset + limit]
|
||
return KnowledgeListResponse(
|
||
items=page,
|
||
total=total,
|
||
categories=_source_category_counts(entries),
|
||
asset_taxonomy=_source_asset_taxonomy_counts(entries),
|
||
readback_status=readback_status,
|
||
primary_readback_ready=False,
|
||
degraded_reason_code=_classify_knowledge_readback_degraded_reason(reason),
|
||
operator_stage="knowledge_readback_source_backed_ai_controlled_repair",
|
||
next_step=(
|
||
"repair_primary_km_db_readback_then_promote_source_backed_receipts_to_persistent_km"
|
||
),
|
||
writes_on_read=False,
|
||
manual_review_required=False,
|
||
)
|
||
|
||
|
||
def _source_backed_entries() -> list[KnowledgeEntry]:
|
||
entries: list[KnowledgeEntry] = []
|
||
for spec in _SOURCE_BACKED_KNOWLEDGE_SPECS:
|
||
tags = [str(tag) for tag in cast(list[object], spec.get("tags", []))]
|
||
related_playbook_id = spec.get("related_playbook_id")
|
||
entries.append(
|
||
KnowledgeEntry(
|
||
id=str(spec["id"]),
|
||
title=str(spec["title"]),
|
||
content=str(spec["content"]),
|
||
entry_type=cast(EntryType, spec["entry_type"]),
|
||
category=str(spec["category"]),
|
||
tags=tags,
|
||
source=EntrySource.AI_EXTRACTED,
|
||
status=EntryStatus.APPROVED,
|
||
related_playbook_id=(
|
||
str(related_playbook_id) if related_playbook_id else None
|
||
),
|
||
view_count=0,
|
||
created_by="ai_agent_source_backed_km_readback",
|
||
)
|
||
)
|
||
return entries
|
||
|
||
|
||
def _filter_source_backed_entries(
|
||
*,
|
||
category: str | None = None,
|
||
entry_type: EntryType | None = None,
|
||
status: EntryStatus | None = None,
|
||
tags: list[str] | None = None,
|
||
q: str | None = None,
|
||
) -> list[KnowledgeEntry]:
|
||
entries = _source_backed_entries()
|
||
if category:
|
||
entries = [entry for entry in entries if entry.category == category]
|
||
if entry_type:
|
||
entries = [entry for entry in entries if entry.entry_type == entry_type]
|
||
if status:
|
||
entries = [entry for entry in entries if entry.status == status]
|
||
if tags:
|
||
wanted = {tag.lower() for tag in tags}
|
||
entries = [
|
||
entry
|
||
for entry in entries
|
||
if wanted.issubset({tag.lower() for tag in entry.tags})
|
||
]
|
||
if q:
|
||
needle = q.lower()
|
||
entries = [
|
||
entry
|
||
for entry in entries
|
||
if needle
|
||
in " ".join(
|
||
[
|
||
entry.id,
|
||
entry.title,
|
||
entry.content,
|
||
entry.category,
|
||
entry.entry_type.value,
|
||
*entry.tags,
|
||
]
|
||
).lower()
|
||
]
|
||
return entries
|
||
|
||
|
||
def _source_category_counts(entries: list[KnowledgeEntry]) -> list[CategoryCount]:
|
||
counts = {category: 0 for category in _DEGRADED_CATEGORY_FALLBACKS}
|
||
for entry in entries:
|
||
counts[entry.category] = counts.get(entry.category, 0) + 1
|
||
ordered = [
|
||
CategoryCount(category=category, count=counts.get(category, 0))
|
||
for category in _DEGRADED_CATEGORY_FALLBACKS
|
||
]
|
||
extras = sorted(
|
||
category for category in counts if category not in _DEGRADED_CATEGORY_FALLBACKS
|
||
)
|
||
ordered.extend(CategoryCount(category=category, count=counts[category]) for category in extras)
|
||
return ordered
|
||
|
||
|
||
def _source_asset_taxonomy_counts(
|
||
entries: list[KnowledgeEntry],
|
||
) -> list[KnowledgeAssetTaxonomyCount]:
|
||
return [
|
||
KnowledgeAssetTaxonomyCount(
|
||
key=key,
|
||
count=sum(1 for entry in entries if _entry_matches_asset_key(entry, key)),
|
||
)
|
||
for key in _ASSET_TAXONOMY_FALLBACKS
|
||
]
|
||
|
||
|
||
def _entry_matches_asset_key(entry: KnowledgeEntry, key: str) -> bool:
|
||
if entry.category == key:
|
||
return True
|
||
if entry.category.lower() in {
|
||
hint.lower() for hint in _ASSET_TAXONOMY_CATEGORY_HINTS.get(key, ())
|
||
}:
|
||
return True
|
||
if key in {tag.lower() for tag in entry.tags}:
|
||
return True
|
||
if key == "playbook" and entry.related_playbook_id:
|
||
return True
|
||
return False
|
||
|
||
|
||
def get_knowledge_service() -> "KnowledgeService":
|
||
"""取得 Knowledge Service 實例"""
|
||
global _knowledge_service
|
||
if _knowledge_service is None:
|
||
_knowledge_service = KnowledgeService()
|
||
return _knowledge_service
|
||
|
||
|
||
class KnowledgeService:
|
||
"""Knowledge Base 業務邏輯"""
|
||
|
||
def __init__(self) -> None:
|
||
# I2: 注入 embedding service，避免每次呼叫 new 實例
|
||
# D1 集中化 2026-04-11: 從 models.json providers.ollama.models.embedding 讀取
|
||
self._embed_svc = OllamaEmbeddingService(model=_get_model("ollama", "embedding"), timeout=15.0)
|
||
# I1: 持有背景 Task 引用，防止 GC 提前回收
|
||
self._pending_tasks: set[asyncio.Task] = set() # type: ignore[type-arg]
|
||
|
||
async def create_entry(self, data: KnowledgeEntryCreate) -> KnowledgeEntry:
|
||
"""建立知識條目，建立後背景自動產生 embedding"""
|
||
async with get_db_context() as db:
|
||
repo: IKnowledgeRepository = KnowledgeDBRepository(db)
|
||
entry = await repo.create(data)
|
||
logger.info(
|
||
"knowledge_entry_created",
|
||
entry_id=entry.id,
|
||
entry_type=entry.entry_type,
|
||
source=entry.source,
|
||
)
|
||
|
||
# 背景產生 embedding (不阻塞回應)；持有引用防 GC 回收
|
||
task = asyncio.create_task(self._embed_entry(entry.id, data.title, data.content))
|
||
self._pending_tasks.add(task)
|
||
task.add_done_callback(self._pending_tasks.discard)
|
||
return entry
|
||
|
||
async def _embed_entry(self, entry_id: str, title: str, content: str) -> None:
|
||
"""背景任務：產生並儲存 embedding"""
|
||
try:
|
||
text = f"search_document: {title}\n\n{content[:2000]}"
|
||
embedding = await self._embed_svc.embed_text(text)
|
||
if not embedding:
|
||
logger.warning("knowledge_embedding_empty", entry_id=entry_id)
|
||
return
|
||
async with get_db_context() as db:
|
||
repo = KnowledgeDBRepository(db)
|
||
await repo.save_embedding(entry_id, embedding)
|
||
logger.info("knowledge_embedding_saved", entry_id=entry_id)
|
||
except Exception as e:
|
||
logger.warning("knowledge_embedding_failed", entry_id=entry_id, error=str(e))
|
||
|
||
async def get_entry(self, entry_id: str) -> KnowledgeEntry | None:
|
||
"""取得知識條目 (view_count +1)"""
|
||
async with get_db_context() as db:
|
||
repo: IKnowledgeRepository = KnowledgeDBRepository(db)
|
||
entry = await repo.get_by_id(entry_id)
|
||
if entry:
|
||
await repo.increment_view_count(entry_id)
|
||
entry.view_count += 1
|
||
return entry
|
||
|
||
async def update_entry(
|
||
self, entry_id: str, data: KnowledgeEntryUpdate
|
||
) -> KnowledgeEntry | None:
|
||
"""更新知識條目"""
|
||
update_data = data.model_dump(exclude_none=True)
|
||
async with get_db_context() as db:
|
||
repo: IKnowledgeRepository = KnowledgeDBRepository(db)
|
||
if not update_data:
|
||
return await repo.get_by_id(entry_id)
|
||
return await repo.update(entry_id, update_data)
|
||
|
||
async def approve_entry(self, entry_id: str) -> KnowledgeEntry | None:
|
||
"""審核通過 (draft/review → approved)"""
|
||
async with get_db_context() as db:
|
||
repo: IKnowledgeRepository = KnowledgeDBRepository(db)
|
||
entry = await repo.get_by_id(entry_id)
|
||
if not entry:
|
||
return None
|
||
if entry.status == EntryStatus.APPROVED:
|
||
return entry
|
||
return await repo.update(entry_id, {"status": EntryStatus.APPROVED})
|
||
|
||
async def archive_entry(self, entry_id: str) -> bool:
|
||
"""封存 (軟刪除)"""
|
||
async with get_db_context() as db:
|
||
repo: IKnowledgeRepository = KnowledgeDBRepository(db)
|
||
return await repo.delete(entry_id)
|
||
|
||
async def _list_entries_from_primary(
|
||
self,
|
||
*,
|
||
category: str | None,
|
||
entry_type: EntryType | None,
|
||
status: EntryStatus | None,
|
||
tags: list[str] | None,
|
||
q: str | None,
|
||
limit: int,
|
||
offset: int,
|
||
) -> KnowledgeListResponse:
|
||
async with get_db_context() as db:
|
||
repo: IKnowledgeRepository = KnowledgeDBRepository(db)
|
||
items, total = await repo.list_entries(
|
||
category=category,
|
||
entry_type=entry_type,
|
||
status=status,
|
||
tags=tags,
|
||
q=q,
|
||
limit=limit,
|
||
offset=offset,
|
||
)
|
||
if total == 0:
|
||
return build_knowledge_list_readback_degraded_response(
|
||
"knowledge_db_empty",
|
||
category=category,
|
||
entry_type=entry_type,
|
||
status=status,
|
||
tags=tags,
|
||
q=q,
|
||
limit=limit,
|
||
offset=offset,
|
||
readback_status="source_backed_db_empty",
|
||
)
|
||
|
||
readback_status = "ready"
|
||
degraded_reason_code: str | None = None
|
||
operator_stage: str | None = None
|
||
next_step: str | None = None
|
||
|
||
try:
|
||
categories = await asyncio.wait_for(
|
||
self._read_primary_categories(),
|
||
timeout=_PRIMARY_KM_TAXONOMY_TIMEOUT_SECONDS,
|
||
)
|
||
except Exception as exc: # noqa: BLE001 - primary entries must stay visible
|
||
logger.warning(
|
||
"knowledge_list_categories_readback_degraded",
|
||
error=str(exc),
|
||
total=total,
|
||
limit=limit,
|
||
offset=offset,
|
||
)
|
||
categories = _source_category_counts(items)
|
||
readback_status = "ready_partial_taxonomy_degraded"
|
||
degraded_reason_code = "primary_km_taxonomy_readback_degraded"
|
||
operator_stage = "knowledge_primary_entries_ready_taxonomy_degraded"
|
||
next_step = "repair_primary_km_taxonomy_readback_without_hiding_entries"
|
||
|
||
try:
|
||
asset_taxonomy = await asyncio.wait_for(
|
||
self._read_primary_asset_taxonomy(),
|
||
timeout=_PRIMARY_KM_TAXONOMY_TIMEOUT_SECONDS,
|
||
)
|
||
except Exception as exc: # noqa: BLE001 - primary entries must stay visible
|
||
logger.warning(
|
||
"knowledge_list_asset_taxonomy_readback_degraded",
|
||
error=str(exc),
|
||
total=total,
|
||
limit=limit,
|
||
offset=offset,
|
||
)
|
||
asset_taxonomy = _source_asset_taxonomy_counts(items)
|
||
readback_status = "ready_partial_taxonomy_degraded"
|
||
degraded_reason_code = "primary_km_taxonomy_readback_degraded"
|
||
operator_stage = "knowledge_primary_entries_ready_taxonomy_degraded"
|
||
next_step = "repair_primary_km_taxonomy_readback_without_hiding_entries"
|
||
|
||
return KnowledgeListResponse(
|
||
items=items,
|
||
total=total,
|
||
categories=categories,
|
||
asset_taxonomy=asset_taxonomy,
|
||
readback_status=readback_status,
|
||
primary_readback_ready=True,
|
||
degraded_reason_code=degraded_reason_code,
|
||
operator_stage=operator_stage,
|
||
next_step=next_step,
|
||
)
|
||
|
||
async def _read_primary_categories(self) -> list[CategoryCount]:
|
||
async with get_db_context() as db:
|
||
repo: IKnowledgeRepository = KnowledgeDBRepository(db)
|
||
categories_raw = await repo.get_categories()
|
||
categories = [
|
||
CategoryCount(category=cat, count=cnt) for cat, cnt in categories_raw
|
||
]
|
||
return categories
|
||
|
||
async def _read_primary_asset_taxonomy(self) -> list[KnowledgeAssetTaxonomyCount]:
|
||
async with get_db_context() as db:
|
||
repo: IKnowledgeRepository = KnowledgeDBRepository(db)
|
||
asset_taxonomy_raw = await repo.get_asset_taxonomy_counts()
|
||
return [
|
||
KnowledgeAssetTaxonomyCount(key=key, count=count)
|
||
for key, count in asset_taxonomy_raw
|
||
]
|
||
|
||
async def _search_primary_entries(
|
||
self,
|
||
query: str,
|
||
limit: int,
|
||
) -> list[KnowledgeEntry]:
|
||
async with get_db_context() as db:
|
||
repo: IKnowledgeRepository = KnowledgeDBRepository(db)
|
||
return await repo.search(query, limit)
|
||
|
||
async def _list_entries_from_primary_bounded(
|
||
self,
|
||
*,
|
||
category: str | None,
|
||
entry_type: EntryType | None,
|
||
status: EntryStatus | None,
|
||
tags: list[str] | None,
|
||
q: str | None,
|
||
limit: int,
|
||
offset: int,
|
||
timeout: float,
|
||
) -> KnowledgeListResponse:
|
||
return await asyncio.wait_for(
|
||
self._list_entries_from_primary(
|
||
category=category,
|
||
entry_type=entry_type,
|
||
status=status,
|
||
tags=tags,
|
||
q=q,
|
||
limit=limit,
|
||
offset=offset,
|
||
),
|
||
timeout=timeout,
|
||
)
|
||
|
||
async def _list_entries_from_direct(
|
||
self,
|
||
*,
|
||
category: str | None,
|
||
entry_type: EntryType | None,
|
||
status: EntryStatus | None,
|
||
tags: list[str] | None,
|
||
q: str | None,
|
||
limit: int,
|
||
offset: int,
|
||
) -> KnowledgeListResponse | None:
|
||
"""Pool 壓力下以短連線只讀救回 primary KM，不把 727 筆誤降成 13 筆。"""
|
||
project_id = _current_knowledge_project_id()
|
||
if not project_id:
|
||
return None
|
||
|
||
db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://")
|
||
try:
|
||
import asyncpg # type: ignore[import-untyped]
|
||
|
||
conn = await asyncio.wait_for(
|
||
asyncpg.connect(db_url),
|
||
timeout=_PRIMARY_KM_DIRECT_CONNECT_TIMEOUT_SECONDS,
|
||
)
|
||
try:
|
||
await asyncio.wait_for(
|
||
conn.execute("SELECT set_config('app.project_id', $1, TRUE)", project_id),
|
||
timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
|
||
)
|
||
await asyncio.wait_for(
|
||
conn.execute(
|
||
"SET statement_timeout = "
|
||
f"'{int(_PRIMARY_KM_DIRECT_STATEMENT_TIMEOUT_MS)}ms'"
|
||
),
|
||
timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
|
||
)
|
||
clauses, params = _direct_base_where(project_id)
|
||
_append_direct_filter(
|
||
clauses,
|
||
params,
|
||
category=category,
|
||
entry_type=entry_type,
|
||
status=status,
|
||
tags=tags,
|
||
q=q,
|
||
)
|
||
where_sql = _direct_where_sql(clauses)
|
||
total = int(
|
||
await asyncio.wait_for(
|
||
conn.fetchval(
|
||
f"SELECT count(*)::int FROM knowledge_entries{where_sql}",
|
||
*params,
|
||
),
|
||
timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
|
||
)
|
||
or 0
|
||
)
|
||
if total == 0:
|
||
return build_knowledge_list_readback_degraded_response(
|
||
"knowledge_db_empty_direct_connection",
|
||
category=category,
|
||
entry_type=entry_type,
|
||
status=status,
|
||
tags=tags,
|
||
q=q,
|
||
limit=limit,
|
||
offset=offset,
|
||
readback_status="source_backed_db_empty_direct_connection",
|
||
)
|
||
|
||
list_params = [*params, limit, offset]
|
||
limit_index = len(params) + 1
|
||
offset_index = len(params) + 2
|
||
rows = await asyncio.wait_for(
|
||
conn.fetch(
|
||
f"""
|
||
SELECT
|
||
id::text AS id,
|
||
title,
|
||
content,
|
||
entry_type::text AS entry_type,
|
||
coalesce(nullif(trim(category), ''), 'general') AS category,
|
||
tags,
|
||
source::text AS source,
|
||
status::text AS status,
|
||
related_incident_id,
|
||
related_playbook_id,
|
||
related_approval_id,
|
||
path_type,
|
||
symptoms_hash,
|
||
coalesce(view_count, 0)::int AS view_count,
|
||
created_by,
|
||
created_at,
|
||
updated_at
|
||
FROM knowledge_entries
|
||
{where_sql}
|
||
ORDER BY updated_at DESC NULLS LAST, created_at DESC NULLS LAST
|
||
LIMIT ${limit_index} OFFSET ${offset_index}
|
||
""",
|
||
*list_params,
|
||
),
|
||
timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
|
||
)
|
||
items = [_direct_knowledge_entry_from_row(dict(row)) for row in rows]
|
||
categories = await self._read_categories_direct_with_conn(
|
||
conn,
|
||
project_id=project_id,
|
||
)
|
||
asset_taxonomy = await self._read_asset_taxonomy_direct_with_conn(
|
||
conn,
|
||
project_id=project_id,
|
||
)
|
||
return KnowledgeListResponse(
|
||
items=items,
|
||
total=total,
|
||
categories=categories or _source_category_counts(items),
|
||
asset_taxonomy=asset_taxonomy or _source_asset_taxonomy_counts(items),
|
||
readback_status="ready_direct_connection_after_session_timeout",
|
||
primary_readback_ready=True,
|
||
operator_stage="knowledge_readback_direct_connection_recovered",
|
||
next_step="repair_session_pool_readback_without_hiding_primary_km",
|
||
writes_on_read=False,
|
||
manual_review_required=False,
|
||
)
|
||
finally:
|
||
try:
|
||
await asyncio.wait_for(
|
||
conn.close(),
|
||
timeout=_PRIMARY_KM_DIRECT_CLOSE_TIMEOUT_SECONDS,
|
||
)
|
||
except Exception as close_exc: # pragma: no cover - live cleanup
|
||
logger.warning(
|
||
"knowledge_direct_readback_close_failed",
|
||
project_id=project_id,
|
||
error_type=type(close_exc).__name__,
|
||
)
|
||
except Exception as exc: # pragma: no cover - live DB pressure
|
||
logger.warning(
|
||
"knowledge_direct_readback_failed",
|
||
project_id=project_id,
|
||
error_type=type(exc).__name__,
|
||
)
|
||
return None
|
||
|
||
async def _read_categories_direct(self) -> list[CategoryCount] | None:
|
||
project_id = _current_knowledge_project_id()
|
||
if not project_id:
|
||
return None
|
||
db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://")
|
||
try:
|
||
import asyncpg # type: ignore[import-untyped]
|
||
|
||
conn = await asyncio.wait_for(
|
||
asyncpg.connect(db_url),
|
||
timeout=_PRIMARY_KM_DIRECT_CONNECT_TIMEOUT_SECONDS,
|
||
)
|
||
try:
|
||
await asyncio.wait_for(
|
||
conn.execute("SELECT set_config('app.project_id', $1, TRUE)", project_id),
|
||
timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
|
||
)
|
||
await asyncio.wait_for(
|
||
conn.execute(
|
||
"SET statement_timeout = "
|
||
f"'{int(_PRIMARY_KM_DIRECT_STATEMENT_TIMEOUT_MS)}ms'"
|
||
),
|
||
timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
|
||
)
|
||
return await self._read_categories_direct_with_conn(
|
||
conn,
|
||
project_id=project_id,
|
||
)
|
||
finally:
|
||
await asyncio.wait_for(
|
||
conn.close(),
|
||
timeout=_PRIMARY_KM_DIRECT_CLOSE_TIMEOUT_SECONDS,
|
||
)
|
||
except Exception as exc: # pragma: no cover - live DB pressure
|
||
logger.warning(
|
||
"knowledge_categories_direct_readback_failed",
|
||
project_id=project_id,
|
||
error_type=type(exc).__name__,
|
||
)
|
||
return None
|
||
|
||
async def _read_asset_taxonomy_direct(self) -> list[KnowledgeAssetTaxonomyCount] | None:
|
||
project_id = _current_knowledge_project_id()
|
||
if not project_id:
|
||
return None
|
||
db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://")
|
||
try:
|
||
import asyncpg # type: ignore[import-untyped]
|
||
|
||
conn = await asyncio.wait_for(
|
||
asyncpg.connect(db_url),
|
||
timeout=_PRIMARY_KM_DIRECT_CONNECT_TIMEOUT_SECONDS,
|
||
)
|
||
try:
|
||
await asyncio.wait_for(
|
||
conn.execute("SELECT set_config('app.project_id', $1, TRUE)", project_id),
|
||
timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
|
||
)
|
||
await asyncio.wait_for(
|
||
conn.execute(
|
||
"SET statement_timeout = "
|
||
f"'{int(_PRIMARY_KM_DIRECT_STATEMENT_TIMEOUT_MS)}ms'"
|
||
),
|
||
timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
|
||
)
|
||
return await self._read_asset_taxonomy_direct_with_conn(
|
||
conn,
|
||
project_id=project_id,
|
||
)
|
||
finally:
|
||
await asyncio.wait_for(
|
||
conn.close(),
|
||
timeout=_PRIMARY_KM_DIRECT_CLOSE_TIMEOUT_SECONDS,
|
||
)
|
||
except Exception as exc: # pragma: no cover - live DB pressure
|
||
logger.warning(
|
||
"knowledge_asset_taxonomy_direct_readback_failed",
|
||
project_id=project_id,
|
||
error_type=type(exc).__name__,
|
||
)
|
||
return None
|
||
|
||
async def _read_categories_direct_with_conn(
|
||
self,
|
||
conn: Any,
|
||
*,
|
||
project_id: str,
|
||
) -> list[CategoryCount]:
|
||
rows = await asyncio.wait_for(
|
||
conn.fetch(
|
||
"""
|
||
SELECT
|
||
coalesce(nullif(trim(category), ''), 'general') AS category,
|
||
count(*)::int AS cnt
|
||
FROM knowledge_entries
|
||
WHERE project_id = $1
|
||
AND lower(status::text) != 'archived'
|
||
GROUP BY 1
|
||
ORDER BY count(*) DESC, category ASC
|
||
""",
|
||
project_id,
|
||
),
|
||
timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
|
||
)
|
||
return [
|
||
CategoryCount(
|
||
category=str(row["category"] or "general"),
|
||
count=int(row["cnt"] or 0),
|
||
)
|
||
for row in rows
|
||
]
|
||
|
||
async def _read_asset_taxonomy_direct_with_conn(
|
||
self,
|
||
conn: Any,
|
||
*,
|
||
project_id: str,
|
||
) -> list[KnowledgeAssetTaxonomyCount]:
|
||
select_columns = ",\n".join(
|
||
(
|
||
"count(*) FILTER (WHERE "
|
||
f"({_asset_taxonomy_direct_condition(key)}))::int AS {key}"
|
||
)
|
||
for key in _ASSET_TAXONOMY_FALLBACKS
|
||
)
|
||
row = await asyncio.wait_for(
|
||
conn.fetchrow(
|
||
f"""
|
||
SELECT
|
||
{select_columns}
|
||
FROM knowledge_entries
|
||
WHERE project_id = $1
|
||
AND lower(status::text) != 'archived'
|
||
""",
|
||
project_id,
|
||
),
|
||
timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
|
||
)
|
||
if row is None:
|
||
return []
|
||
return [
|
||
KnowledgeAssetTaxonomyCount(key=key, count=int(row[key] or 0))
|
||
for key in _ASSET_TAXONOMY_FALLBACKS
|
||
]
|
||
|
||
async def list_entries(
|
||
self,
|
||
category: str | None = None,
|
||
entry_type: EntryType | None = None,
|
||
status: EntryStatus | None = None,
|
||
tags: list[str] | None = None,
|
||
q: str | None = None,
|
||
limit: int = 20,
|
||
offset: int = 0,
|
||
) -> KnowledgeListResponse:
|
||
"""列出知識條目 + 分類統計"""
|
||
try:
|
||
return await self._list_entries_from_primary_bounded(
|
||
category=category,
|
||
entry_type=entry_type,
|
||
status=status,
|
||
tags=tags,
|
||
q=q,
|
||
limit=limit,
|
||
offset=offset,
|
||
timeout=_PRIMARY_KM_LIST_TIMEOUT_SECONDS,
|
||
)
|
||
except Exception as exc: # noqa: BLE001 - production readback must fail soft
|
||
degraded_reason = _knowledge_readback_exception_reason(exc)
|
||
degraded_reason_code = _classify_knowledge_readback_degraded_reason(degraded_reason)
|
||
if degraded_reason_code == "primary_km_db_timeout_or_pool_exhausted":
|
||
try:
|
||
await asyncio.sleep(_PRIMARY_KM_RETRY_DELAY_SECONDS)
|
||
retry_response = await self._list_entries_from_primary_bounded(
|
||
category=category,
|
||
entry_type=entry_type,
|
||
status=status,
|
||
tags=tags,
|
||
q=q,
|
||
limit=limit,
|
||
offset=offset,
|
||
timeout=_PRIMARY_KM_LIST_RETRY_TIMEOUT_SECONDS,
|
||
)
|
||
retry_response.operator_stage = "knowledge_readback_primary_retry_recovered"
|
||
return retry_response
|
||
except Exception as retry_exc: # noqa: BLE001 - keep source-backed fallback
|
||
retry_reason = _knowledge_readback_exception_reason(retry_exc)
|
||
logger.warning(
|
||
"knowledge_list_readback_retry_degraded",
|
||
error=retry_reason,
|
||
degraded_reason_code=_classify_knowledge_readback_degraded_reason(retry_reason),
|
||
q=q,
|
||
limit=limit,
|
||
offset=offset,
|
||
)
|
||
direct_response = await self._list_entries_from_direct(
|
||
category=category,
|
||
entry_type=entry_type,
|
||
status=status,
|
||
tags=tags,
|
||
q=q,
|
||
limit=limit,
|
||
offset=offset,
|
||
)
|
||
if direct_response is not None:
|
||
return direct_response
|
||
logger.warning(
|
||
"knowledge_list_readback_degraded",
|
||
error=degraded_reason,
|
||
degraded_reason_code=degraded_reason_code,
|
||
category=category,
|
||
entry_type=entry_type.value if entry_type else None,
|
||
status=status.value if status else None,
|
||
q=q,
|
||
limit=limit,
|
||
offset=offset,
|
||
)
|
||
return build_knowledge_list_readback_degraded_response(
|
||
degraded_reason,
|
||
category=category,
|
||
entry_type=entry_type,
|
||
status=status,
|
||
tags=tags,
|
||
q=q,
|
||
limit=limit,
|
||
offset=offset,
|
||
)
|
||
|
||
async def get_asset_taxonomy(self) -> list[KnowledgeAssetTaxonomyCount]:
|
||
"""取得 AI 自動化資產維度統計。"""
|
||
try:
|
||
taxonomy = await asyncio.wait_for(
|
||
self._read_primary_asset_taxonomy(),
|
||
timeout=_PRIMARY_KM_SIDE_READ_TIMEOUT_SECONDS,
|
||
)
|
||
if taxonomy and any(row.count > 0 for row in taxonomy):
|
||
return taxonomy
|
||
return _source_asset_taxonomy_counts(_source_backed_entries())
|
||
except Exception as exc: # noqa: BLE001 - taxonomy must not 500 the KM UI
|
||
reason = _knowledge_readback_exception_reason(exc)
|
||
if _classify_knowledge_readback_degraded_reason(reason) == "primary_km_db_timeout_or_pool_exhausted":
|
||
direct_taxonomy = await self._read_asset_taxonomy_direct()
|
||
if direct_taxonomy and any(row.count > 0 for row in direct_taxonomy):
|
||
return direct_taxonomy
|
||
logger.warning(
|
||
"knowledge_asset_taxonomy_readback_degraded",
|
||
error=reason,
|
||
degraded_reason_code=_classify_knowledge_readback_degraded_reason(reason),
|
||
)
|
||
return _source_asset_taxonomy_counts(_source_backed_entries())
|
||
|
||
async def get_categories(self) -> list[CategoryCount]:
|
||
"""取得分類統計（直接呼叫 repo，不走 list_entries）"""
|
||
try:
|
||
categories = await asyncio.wait_for(
|
||
self._read_primary_categories(),
|
||
timeout=_PRIMARY_KM_SIDE_READ_TIMEOUT_SECONDS,
|
||
)
|
||
if categories:
|
||
return categories
|
||
return _source_category_counts(_source_backed_entries())
|
||
except Exception as exc: # noqa: BLE001 - categories must not 500 the KM UI
|
||
reason = _knowledge_readback_exception_reason(exc)
|
||
if _classify_knowledge_readback_degraded_reason(reason) == "primary_km_db_timeout_or_pool_exhausted":
|
||
direct_categories = await self._read_categories_direct()
|
||
if direct_categories and any(row.count > 0 for row in direct_categories):
|
||
return direct_categories
|
||
logger.warning(
|
||
"knowledge_categories_readback_degraded",
|
||
error=reason,
|
||
degraded_reason_code=_classify_knowledge_readback_degraded_reason(reason),
|
||
)
|
||
return _source_category_counts(_source_backed_entries())
|
||
|
||
async def search(self, query: str, limit: int = 20) -> list[KnowledgeEntry]:
|
||
"""關鍵字搜尋"""
|
||
try:
|
||
return await asyncio.wait_for(
|
||
self._search_primary_entries(query, limit),
|
||
timeout=_PRIMARY_KM_SIDE_READ_TIMEOUT_SECONDS,
|
||
)
|
||
except Exception as exc: # noqa: BLE001 - KM search must not 500 the UI
|
||
reason = _knowledge_readback_exception_reason(exc)
|
||
logger.warning(
|
||
"knowledge_search_readback_degraded",
|
||
error=reason,
|
||
q=query,
|
||
limit=limit,
|
||
degraded_reason_code=_classify_knowledge_readback_degraded_reason(reason),
|
||
)
|
||
return _filter_source_backed_entries(q=query)[:limit]
|
||
|
||
async def semantic_search(
|
||
self,
|
||
query: str,
|
||
limit: int = 10,
|
||
threshold: float = 0.5,
|
||
) -> list[tuple[KnowledgeEntry, float]]:
|
||
"""
|
||
語意搜尋 (pgvector cosine similarity)
|
||
|
||
Returns:
|
||
list of (entry, score) 已按相似度降序排列
|
||
"""
|
||
query_text = f"search_query: {query}"
|
||
embedding = await self._embed_svc.embed_text(query_text)
|
||
if not embedding:
|
||
logger.warning("semantic_search_embedding_failed", query=query)
|
||
return []
|
||
|
||
async with get_db_context() as db:
|
||
repo: IKnowledgeRepository = KnowledgeDBRepository(db)
|
||
return await repo.semantic_search(embedding, limit=limit, threshold=threshold)
|
||
|
||
async def embed_all_entries(self) -> dict[str, int]:
|
||
"""
|
||
批次為所有未 embed 的條目產生 embedding (管理用)
|
||
|
||
Returns:
|
||
{"total": N, "success": N, "failed": N}
|
||
"""
|
||
# C2 修復: 透過 Repository 取得資料，Service 不直接執行 raw SQL
|
||
async with get_db_context() as db:
|
||
repo: IKnowledgeRepository = KnowledgeDBRepository(db)
|
||
rows = await repo.list_unembedded_entries()
|
||
|
||
success = failed = 0
|
||
for entry_id, title, content in rows:
|
||
try:
|
||
text = f"search_document: {title}\n\n{content[:2000]}"
|
||
embedding = await self._embed_svc.embed_text(text)
|
||
if embedding:
|
||
async with get_db_context() as db:
|
||
repo = KnowledgeDBRepository(db)
|
||
await repo.save_embedding(entry_id, embedding)
|
||
success += 1
|
||
else:
|
||
logger.warning("embed_all_empty_vector", entry_id=entry_id)
|
||
failed += 1
|
||
except Exception as e:
|
||
logger.warning("embed_all_failed", entry_id=entry_id, error=str(e))
|
||
failed += 1
|
||
|
||
logger.info("embed_all_complete", total=len(rows), success=success, failed=failed)
|
||
return {"total": len(rows), "success": success, "failed": failed}
|
||
|
||
async def check_anti_pattern(
|
||
self,
|
||
symptoms_hash: str,
|
||
days: int = 7,
|
||
) -> list[KnowledgeEntry]:
|
||
"""
|
||
2026-04-04 Claude Code: Phase 25 P1 — Anti-Pattern 閉環閘門
|
||
根據 symptoms_hash 查找近期失敗案例，供 auto_repair decide() 攔截用
|
||
|
||
Args:
|
||
symptoms_hash: SymptomPattern.compute_hash() 的 16 字元 hash
|
||
days: 查找幾天內的記錄（預設 7 天）
|
||
|
||
Returns:
|
||
list[KnowledgeEntry] — ANTI_PATTERN 條目，空表示無已知失敗案例
|
||
"""
|
||
from datetime import timedelta
|
||
|
||
from sqlalchemy import text as sa_text
|
||
|
||
from src.utils.timezone import now_taipei
|
||
|
||
cutoff = now_taipei() - timedelta(days=days)
|
||
|
||
async with get_db_context() as db:
|
||
result = await db.execute(
|
||
sa_text(
|
||
"SELECT id FROM knowledge_entries "
|
||
"WHERE LOWER(CAST(entry_type AS TEXT)) = :entry_type "
|
||
"AND symptoms_hash = :hash "
|
||
"AND created_at >= :cutoff "
|
||
"AND LOWER(CAST(status AS TEXT)) != :archived "
|
||
"ORDER BY created_at DESC LIMIT 5"
|
||
),
|
||
{
|
||
"entry_type": EntryType.ANTI_PATTERN.value,
|
||
"hash": symptoms_hash,
|
||
"cutoff": cutoff,
|
||
"archived": EntryStatus.ARCHIVED.value,
|
||
},
|
||
)
|
||
entry_ids = [row.id for row in result.fetchall()]
|
||
|
||
if not entry_ids:
|
||
return []
|
||
|
||
entries = []
|
||
for eid in entry_ids:
|
||
entry = await self.get_entry(eid)
|
||
if entry:
|
||
entries.append(entry)
|
||
|
||
logger.info(
|
||
"anti_pattern_check",
|
||
symptoms_hash=symptoms_hash,
|
||
days=days,
|
||
found=len(entries),
|
||
)
|
||
return entries
|