[Phase 5] feat(ai): 법률 RAG 임베딩 적재 구현

backend-ai/app/clients/supabase_client.py

@@ -1,3 +1,4 @@
+import json
 from typing import Any
 
 import psycopg
@@ -58,8 +59,106 @@ def similarity_search_legal_documents(
                 cursor.execute(sql, (vector_literal, vector_literal, top_k))
                 return list(cursor.fetchall())
 
+    def upsert_legal_document_chunks(self, rows: list[dict[str, Any]]) -> int:
+        if not rows:
+            return 0
+
+        sql = """
+            insert into public.legal_document_chunks as existing (
+                law_id,
+                law_name,
+                article_no,
+                article_title,
+                effective_date,
+                source_name,
+                source_url,
+                chunk_index,
+                content,
+                content_hash,
+                embedding,
+                metadata_json
+            )
+            values (
+                %(law_id)s,
+                %(law_name)s,
+                %(article_no)s,
+                %(article_title)s,
+                %(effective_date)s,
+                %(source_name)s,
+                %(source_url)s,
+                %(chunk_index)s,
+                %(content)s,
+                %(content_hash)s,
+                %(embedding)s::vector,
+                %(metadata_json)s::jsonb
+            )
+            on conflict (content_hash) do update set
+                law_id = excluded.law_id,
+                law_name = excluded.law_name,
+                article_no = excluded.article_no,
+                article_title = excluded.article_title,
+                effective_date = excluded.effective_date,
+                source_name = excluded.source_name,
+                source_url = excluded.source_url,
+                chunk_index = excluded.chunk_index,
+                content = excluded.content,
+                embedding = excluded.embedding,
+                metadata_json = excluded.metadata_json,
+                updated_at = now()
+            where
+                existing.law_id is distinct from excluded.law_id
+                or existing.law_name is distinct from excluded.law_name
+                or existing.article_no is distinct from excluded.article_no
+                or existing.article_title is distinct from excluded.article_title
+                or existing.effective_date is distinct from excluded.effective_date
+                or existing.source_name is distinct from excluded.source_name
+                or existing.source_url is distinct from excluded.source_url
+                or existing.chunk_index is distinct from excluded.chunk_index
+                or existing.content is distinct from excluded.content
+                or existing.embedding is distinct from excluded.embedding
+                or existing.metadata_json is distinct from excluded.metadata_json
+        """
+        params = [legal_chunk_upsert_params(row) for row in rows]
+        affected_rows = 0
+        with psycopg.connect(
+            self.database_url,
+            row_factory=dict_row,
+            connect_timeout=self.connect_timeout_seconds,
+        ) as conn:
+            with conn.cursor() as cursor:
+                cursor.execute(
+                    "set local statement_timeout = %s",
+                    (self.statement_timeout_ms,),
+                )
+                for row_params in params:
+                    cursor.execute(sql, row_params)
+                    affected_rows += max(int(getattr(cursor, "rowcount", 1)), 0)
+        return affected_rows
+
 
 def to_pgvector_literal(embedding: list[float]) -> str:
     if not embedding:
         raise ValueError("query_embedding must not be empty.")
     return "[" + ",".join(f"{float(value):.10g}" for value in embedding) + "]"
+
+
+def legal_chunk_upsert_params(row: dict[str, Any]) -> dict[str, Any]:
+    embedding = row.get("embedding")
+    if not isinstance(embedding, list) or not embedding:
+        raise ValueError("legal chunk embedding must not be empty.")
+
+    metadata_json = row.get("metadata_json") or {}
+    return {
+        "law_id": row["law_id"],
+        "law_name": row["law_name"],
+        "article_no": row["article_no"],
+        "article_title": row["article_title"],
+        "effective_date": row.get("effective_date"),
+        "source_name": row["source_name"],
+        "source_url": row["source_url"],
+        "chunk_index": row["chunk_index"],
+        "content": row["content"],
+        "content_hash": row["content_hash"],
+        "embedding": to_pgvector_literal(embedding),
+        "metadata_json": json.dumps(metadata_json, ensure_ascii=False),
+    }

backend-ai/scripts/ingest_legal_docs.py

@@ -10,6 +10,8 @@
 from pathlib import Path
 from typing import Any, Sequence
 
+from app.clients.embedding_client import EmbeddingClient
+from app.clients.supabase_client import SupabaseVectorClient
 from app.rag.chunker import chunk_text
 
 
@@ -115,6 +117,31 @@ def prepare_legal_chunks(
     return rows
 
 
+def embed_legal_chunks(
+    rows: Sequence[dict[str, Any]],
+    embedding_client: Any,
+) -> list[dict[str, Any]]:
+    embedded_rows: list[dict[str, Any]] = []
+    for row in rows:
+        content = str(row.get("content", ""))
+        embedded_row = dict(row)
+        embedded_row["embedding"] = embedding_client.embed_query(content)
+        embedded_rows.append(embedded_row)
+    return embedded_rows
+
+
+def write_legal_chunks(
+    rows: Sequence[dict[str, Any]],
+    embedding_client: Any | None = None,
+    vector_client: Any | None = None,
+) -> dict[str, int]:
+    embedding_client = embedding_client or EmbeddingClient()
+    vector_client = vector_client or SupabaseVectorClient()
+    embedded_rows = embed_legal_chunks(rows, embedding_client)
+    upserted = vector_client.upsert_legal_document_chunks(embedded_rows)
+    return {"chunks": len(embedded_rows), "upserted": int(upserted)}
+
+
 def normalize_content(content: str) -> str:
     return "\n".join(line.strip() for line in content.splitlines() if line.strip())
 
@@ -156,15 +183,24 @@ def content_hash(document: LegalDocument, content: str, chunk_index: int) -> str
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(description="Prepare legal document chunks for pgvector ingestion.")
     parser.add_argument("--dry-run", action="store_true", help="Validate and summarize chunks without DB writes.")
+    parser.add_argument("--write", action="store_true", help="Embed chunks and upsert them into Supabase.")
     parser.add_argument("source", type=Path, help="Path to a legal document JSON array.")
     parser.add_argument("--chunk-size", type=int, default=800)
     parser.add_argument("--overlap", type=int, default=120)
     return parser
 
 
-def main(argv: Sequence[str] | None = None) -> int:
+def main(
+    argv: Sequence[str] | None = None,
+    embedding_client: Any | None = None,
+    vector_client: Any | None = None,
+) -> int:
     parser = build_parser()
     args = parser.parse_args(argv)
+    if args.dry_run and args.write:
+        parser.error("Use only one of --dry-run or --write.")
+    if not args.dry_run and not args.write:
+        parser.error("Use --dry-run to validate or --write to upsert legal chunks.")
     if args.chunk_size <= 0:
         parser.error("--chunk-size must be greater than 0.")
     if args.overlap < 0 or args.overlap >= args.chunk_size:
@@ -177,7 +213,12 @@ def main(argv: Sequence[str] | None = None) -> int:
         print(f"Legal ingestion dry-run: documents={len(documents)} chunks={len(rows)}")
         return 0
 
-    raise RuntimeError("Database writes are not enabled in Phase 2. Use --dry-run.")
+    summary = write_legal_chunks(rows, embedding_client, vector_client)
+    print(
+        "Legal ingestion write: "
+        f"documents={len(documents)} chunks={summary['chunks']} upserted={summary['upserted']}"
+    )
+    return 0
 
 
 if __name__ == "__main__":

backend-ai/tests/test_legal_ingestion_upsert.py

@@ -0,0 +1,193 @@
+import json
+
+import pytest
+
+from app.clients import supabase_client as supabase_module
+from app.clients.supabase_client import SupabaseVectorClient
+from scripts.ingest_legal_docs import (
+    embed_legal_chunks,
+    main,
+    write_legal_chunks,
+)
+
+
+class FakeEmbeddingClient:
+    def __init__(self) -> None:
+        self.queries: list[str] = []
+
+    def embed_query(self, query: str) -> list[float]:
+        self.queries.append(query)
+        return [float(len(self.queries)), 0.5]
+
+
+class FakeVectorClient:
+    def __init__(self) -> None:
+        self.rows: list[dict] = []
+
+    def upsert_legal_document_chunks(self, rows: list[dict]) -> int:
+        self.rows.extend(rows)
+        return len(rows)
+
+
+def write_source(tmp_path, documents):
+    source_path = tmp_path / "legal-documents.json"
+    source_path.write_text(json.dumps(documents), encoding="utf-8")
+    return source_path
+
+
+def sample_rows() -> list[dict]:
+    return [
+        {
+            "law_id": "housing-lease-protection-act",
+            "law_name": "Housing Lease Protection Act",
+            "article_no": "Article 3-2",
+            "article_title": "Recovery of Deposit",
+            "effective_date": "2025-01-01",
+            "source_name": "law.go.kr",
+            "source_url": "https://www.law.go.kr",
+            "chunk_index": 0,
+            "content": "A tenant may recover the deposit before junior creditors.",
+            "content_hash": "a" * 64,
+            "embedding": None,
+            "metadata_json": {"sourceUrl": "https://www.law.go.kr"},
+        }
+    ]
+
+
+def test_embed_legal_chunks_uses_embedding_client_without_mutating_source() -> None:
+    rows = sample_rows()
+    embedding_client = FakeEmbeddingClient()
+
+    embedded_rows = embed_legal_chunks(rows, embedding_client)
+
+    assert embedding_client.queries == [
+        "A tenant may recover the deposit before junior creditors."
+    ]
+    assert rows[0]["embedding"] is None
+    assert embedded_rows[0]["embedding"] == [1.0, 0.5]
+    assert embedded_rows[0]["content_hash"] == rows[0]["content_hash"]
+
+
+def test_write_legal_chunks_embeds_then_upserts_rows() -> None:
+    embedding_client = FakeEmbeddingClient()
+    vector_client = FakeVectorClient()
+
+    summary = write_legal_chunks(sample_rows(), embedding_client, vector_client)
+
+    assert summary == {"chunks": 1, "upserted": 1}
+    assert embedding_client.queries == [
+        "A tenant may recover the deposit before junior creditors."
+    ]
+    assert vector_client.rows[0]["embedding"] == [1.0, 0.5]
+
+
+def test_main_requires_explicit_mode_for_database_writes(tmp_path) -> None:
+    source_path = write_source(tmp_path, [])
+
+    with pytest.raises(SystemExit):
+        main([str(source_path)])
+
+
+def test_write_mode_outputs_summary_with_injected_clients(tmp_path, capsys) -> None:
+    source_path = write_source(
+        tmp_path,
+        [
+            {
+                "lawId": "housing-lease-protection-act",
+                "lawName": "Housing Lease Protection Act",
+                "articleNo": "Article 3-2",
+                "title": "Recovery of Deposit",
+                "content": "A tenant may recover the deposit before junior creditors.",
+                "sourceUrl": "https://www.law.go.kr",
+                "effectiveDate": "2025-01-01",
+            }
+        ],
+    )
+    embedding_client = FakeEmbeddingClient()
+    vector_client = FakeVectorClient()
+
+    exit_code = main(
+        ["--write", "--chunk-size", "200", str(source_path)],
+        embedding_client=embedding_client,
+        vector_client=vector_client,
+    )
+
+    output = capsys.readouterr().out
+    assert exit_code == 0
+    assert "Legal ingestion write" in output
+    assert "documents=1" in output
+    assert "chunks=1" in output
+    assert "upserted=1" in output
+    assert len(vector_client.rows) == 1
+
+
+def test_supabase_vector_client_upserts_legal_chunks(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    calls: dict[str, object] = {}
+
+    class FakeCursor:
+        rowcount = 1
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, traceback) -> None:
+            return None
+
+        def execute(self, sql, params=None) -> None:
+            calls.setdefault("executes", []).append((sql, params))
+
+    class FakeConnection:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, traceback) -> None:
+            return None
+
+        def cursor(self) -> FakeCursor:
+            return FakeCursor()
+
+    def fake_connect(database_url, **kwargs):
+        calls["database_url"] = database_url
+        calls["connect_kwargs"] = kwargs
+        return FakeConnection()
+
+    monkeypatch.setattr(supabase_module.psycopg, "connect", fake_connect)
+    client = SupabaseVectorClient(
+        connect_timeout_seconds=7,
+        statement_timeout_ms=3000,
+    )
+
+    upserted = client.upsert_legal_document_chunks(
+        [{**sample_rows()[0], "embedding": [0.1, 0.2]}]
+    )
+
+    executes = calls["executes"]
+    assert upserted == 1
+    assert executes[0] == ("set local statement_timeout = %s", (3000,))
+    assert "on conflict (content_hash)" in executes[1][0]
+    assert "where" in executes[1][0]
+    assert "existing.content is distinct from excluded.content" in executes[1][0]
+    assert executes[1][1]["embedding"] == "[0.1,0.2]"
+    assert executes[1][1]["metadata_json"] == '{"sourceUrl": "https://www.law.go.kr"}'
+
+
+def test_dry_run_mode_returns_success(tmp_path) -> None:
+    source_path = write_source(
+        tmp_path,
+        [
+            {
+                "lawId": "housing-lease-protection-act",
+                "lawName": "Housing Lease Protection Act",
+                "articleNo": "Article 3-2",
+                "title": "Recovery of Deposit",
+                "content": "A tenant may recover the deposit before junior creditors.",
+                "sourceUrl": "https://www.law.go.kr",
+            }
+        ],
+    )
+
+    exit_code = main(["--dry-run", str(source_path)])
+
+    assert exit_code == 0