Langflow 설계 안내
Project
목표
- 감사/회계 질의에 대해 근거 기반 답변(환각 억제) + 유사사례/처분 근거를 안정적으로 출력
- Baseline(단일 질의) 대비 MultiQuery(질의 확장)로 recall 강화 :contentReference[oaicite:1]{index=1}
데이터/저장소
- Neo4j: 검색 후보(Chunk 노드) 탐색
- Vector Index:
chunk_embedding_idx - Fulltext Index:
chunk_text_ft
- Vector Index:
- Postgres: 원문 chunk_text/metadata 제공
- View/Table:
public.audit_flat_chunks(chunk_id, chunk_text, metadata, doc_code, sub_code)
- View/Table:
[Version 1] audit_rag_Baseline (Neo4j+Postgres)
1. 플로우 전체 구조 개요
[1] Chat Input
↓ (사용자 질문)
[2] OpenAI Embeddings
↓ (질문 임베딩)
[3] Neo4j Hybrid Retriever (Custom)
↓ (Candidates DataFrame: chunk_id/doc_code/sub_code/score...)
[4] Postgres Chunk Fetcher (Custom)
↓ (Docs DataFrame: chunk_id/chunk_text/metadata...)
[5] Parser (DataFrame → 문자열 context)
↓
[6] Prompt Template (context + question)
↓
[7] OpenAI (LLM)
↓
[8] Chat Output2. 각 컴포넌트별 상세 설명
(1) Chat Input
- 사용자 질문 입력 노드
- Retriever의
user_question에 연결
(2) OpenAI Embeddings
- 질문을 벡터화하여 Neo4j Vector Search에 사용
- Model 예:
text-embedding-3-large또는text-embedding-3-small
(3) Neo4j Hybrid Retriever (Custom)
-
역할
- Vector / Fulltext / Hybrid 중 선택해 후보 chunk를 탐색
- 결과를
Candidates(DataFrame)으로 반환 (chunk_id 포함)
-
핵심 입력
user_questionembeddings_modelneo4j_uri / user / password / databasevector_index_name / fulltext_index_namesearch_mode(hybrid/vector_only/fulltext_only)k_vec / k_ft / topk,weight_vec / weight_txt
-
핵심 출력
candidates: DataFrame (최소chunk_id컬럼 포함)
-
Code
import math from typing import Any, Dict, List, Optional from langflow.custom import Component from langflow.io import ( Output, MessageTextInput, StrInput, IntInput, FloatInput, SecretStrInput, DropdownInput, ) from langflow.schema.dataframe import DataFrame try: from langflow.io import HandleInput except Exception: from langflow.inputs.inputs import HandleInput from neo4j import GraphDatabase class Neo4jHybridRetriever(Component): display_name = "Neo4j Hybrid Retriever" description = "Vector/Fulltext/Hybrid 검색 후 merge/dedupe + 정규화 가중합으로 Candidates(DataFrame) 반환" icon = "database" name = "Neo4jHybridRetriever" inputs = [ # Chat Input(Message) -> 연결 가능 (MessageTextInput이 텍스트를 추출) MessageTextInput( name="user_question", display_name="User Question", required=True, ), # OpenAI Embeddings 노드 출력(Embeddings) -> 연결 가능 HandleInput( name="embeddings_model", display_name="Embeddings Model", input_types=["embeddings", "Embeddings", "Any"], required=True, ), # Neo4j connection StrInput(name="neo4j_uri", display_name="Neo4j URI", value="bolt://host.docker.internal:7687", required=True), StrInput(name="neo4j_user", display_name="Neo4j User", value="neo4j", required=True), SecretStrInput(name="neo4j_password", display_name="Neo4j Password", required=True), StrInput(name="neo4j_database", display_name="Neo4j DB", value="auditdb2", required=True), # Index names StrInput(name="vector_index_name", display_name="Vector Index", value="chunk_embedding_idx", required=True), StrInput(name="fulltext_index_name", display_name="Fulltext Index", value="chunk_text_ft", required=True), # Mode dropdown DropdownInput( name="search_mode", display_name="Search Mode", options=["hybrid", "vector_only", "fulltext_only"], value="hybrid", required=True, ), # Pools and final topk IntInput(name="k_vec", display_name="Vector TopK (pool)", value=40, required=True), IntInput(name="k_ft", display_name="Fulltext TopK (pool)", value=40, required=True), IntInput(name="topk", display_name="Final TopK", value=10, required=True), # Hybrid weights (정규화된 점수에 가중합) FloatInput(name="weight_vec", display_name="Weight (vector)", value=0.7, required=True), FloatInput(name="weight_txt", display_name="Weight (fulltext)", value=0.3, required=True), ] outputs = [ Output( name="candidates", display_name="Candidates (DataFrame)", method="build", ) ] # -------- helpers -------- @staticmethod def _as_str_question(q: Any) -> str: # MessageTextInput은 보통 str을 주지만, 혹시 모를 케이스 방어 if q is None: return "" return str(q).strip() @staticmethod def _safe_float(x: Any, default: float = 0.0) -> float: try: if x is None: return default return float(x) except Exception: return default @staticmethod def _make_langflow_df(records: List[Dict[str, Any]]) -> DataFrame: """ LangFlow DataFrame 생성 방식이 버전별로 미묘하게 달라서 몇 가지 패턴을 순차 시도합니다. """ # 1) 가장 흔한 형태 try: return DataFrame(data=records) # type: ignore except TypeError: pass # 2) value로 받는 버전 try: return DataFrame(value=records) # type: ignore except TypeError: pass # 3) 마지막 수단: 빈 DF라도 반환 return DataFrame() # type: ignore def _embed_query(self, embeddings_model: Any, question: str) -> List[float]: """ OpenAI Embeddings(=langchain embeddings) 계열은 보통 embed_query 지원. """ if embeddings_model is None: raise ValueError("Embeddings Model is required.") # langchain embeddings if hasattr(embeddings_model, "embed_query"): vec = embeddings_model.embed_query(question) # 혹시 embed_documents만 있는 경우 elif hasattr(embeddings_model, "embed_documents"): vecs = embeddings_model.embed_documents([question]) vec = vecs[0] if vecs else [] else: raise TypeError("Unsupported embeddings model: need embed_query or embed_documents.") # numpy/tuple 방어 return [float(v) for v in list(vec)] def _neo4j_run(self, query: str, params: Dict[str, Any]) -> List[Dict[str, Any]]: driver = GraphDatabase.driver( self.neo4j_uri, auth=(self.neo4j_user, self.neo4j_password), ) try: with driver.session(database=self.neo4j_database) as s: res = s.run(query, params) return [dict(r) for r in res] finally: driver.close() # -------- main -------- def build(self) -> DataFrame: question = self._as_str_question(self.user_question) if not question: return self._make_langflow_df([]) mode = (self.search_mode or "hybrid").strip() k_vec = int(self.k_vec) k_ft = int(self.k_ft) topk = int(self.topk) w_vec = float(self.weight_vec) w_txt = float(self.weight_txt) # 가중치 안정화(합이 0이면 기본값) if (w_vec + w_txt) <= 0: w_vec, w_txt = 0.7, 0.3 # 1) qvec 생성 qvec = self._embed_query(self.embeddings_model, question) # 2) vector 결과 vec_rows: List[Dict[str, Any]] = [] if mode in ("hybrid", "vector_only"): q_vec = """ CALL db.index.vector.queryNodes($vector_index, $k, $qvec) YIELD node, score RETURN node.chunk_id AS chunk_id, node.doc_code AS doc_code, node.sub_code AS sub_code, score AS vec_score ORDER BY vec_score DESC LIMIT $k """ vec_rows = self._neo4j_run( q_vec, { "vector_index": self.vector_index_name, "k": k_vec, "qvec": qvec, }, ) # 3) fulltext 결과 txt_rows: List[Dict[str, Any]] = [] if mode in ("hybrid", "fulltext_only"): q_txt = """ CALL db.index.fulltext.queryNodes($fulltext_index, $q) YIELD node, score RETURN node.chunk_id AS chunk_id, node.doc_code AS doc_code, node.sub_code AS sub_code, score AS txt_score ORDER BY txt_score DESC LIMIT $k """ txt_rows = self._neo4j_run( q_txt, { "fulltext_index": self.fulltext_index_name, "q": question, "k": k_ft, }, ) # 4) rank 부여 + merge/dedupe # key = chunk_id merged: Dict[str, Dict[str, Any]] = {} # vector ranks for i, r in enumerate(vec_rows, start=1): cid = str(r.get("chunk_id", "")).strip() if not cid: continue merged.setdefault(cid, {}) merged[cid]["chunk_id"] = cid merged[cid]["doc_code"] = r.get("doc_code") merged[cid]["sub_code"] = r.get("sub_code") merged[cid]["vec_score"] = self._safe_float(r.get("vec_score")) merged[cid]["vec_rank"] = i # text ranks for i, r in enumerate(txt_rows, start=1): cid = str(r.get("chunk_id", "")).strip() if not cid: continue merged.setdefault(cid, {}) merged[cid]["chunk_id"] = cid # doc/sub는 둘 중 하나라도 있으면 채움 merged[cid]["doc_code"] = merged[cid].get("doc_code") or r.get("doc_code") merged[cid]["sub_code"] = merged[cid].get("sub_code") or r.get("sub_code") merged[cid]["txt_score"] = self._safe_float(r.get("txt_score")) merged[cid]["txt_rank"] = i # 5) 점수 정규화(스케일 차이 방지) + 최종 score max_vec = max([self._safe_float(v.get("vec_score")) for v in merged.values()] + [0.0]) max_txt = max([self._safe_float(v.get("txt_score")) for v in merged.values()] + [0.0]) out: List[Dict[str, Any]] = [] for cid, v in merged.items(): vec_score = self._safe_float(v.get("vec_score")) txt_score = self._safe_float(v.get("txt_score")) vec_rank = v.get("vec_rank") txt_rank = v.get("txt_rank") vec_norm = (vec_score / max_vec) if max_vec > 0 else 0.0 txt_norm = (txt_score / max_txt) if max_txt > 0 else 0.0 if mode == "vector_only": score = vec_norm source = "vec" elif mode == "fulltext_only": score = txt_norm source = "txt" else: score = (w_vec * vec_norm) + (w_txt * txt_norm) if vec_score > 0 and txt_score > 0: source = "vec+txt" elif vec_score > 0: source = "vec" else: source = "txt" out.append( { "chunk_id": cid, "score": float(score), "vec_score": float(vec_score) if vec_score else 0.0, "txt_score": float(txt_score) if txt_score else 0.0, "vec_rank": int(vec_rank) if vec_rank else None, "txt_rank": int(txt_rank) if txt_rank else None, "doc_code": v.get("doc_code"), "sub_code": v.get("sub_code"), "source": source, } ) out.sort(key=lambda x: x["score"], reverse=True) out = out[:topk] return self._make_langflow_df(out)
(4) Postgres Chunk Fetcher (Custom)
-
역할
- Candidates의
chunk_id리스트로 Postgresaudit_flat_chunks에서chunk_text,metadata,doc_code,sub_code를 조회 - 조회 결과를
Docs(DataFrame)으로 반환
- Candidates의
-
Code
from __future__ import annotations from typing import Any, Dict, List, Optional, Tuple import psycopg2 from psycopg2.extras import RealDictCursor from langflow.custom import Component from langflow.io import ( DataFrameInput, Output, StrInput, IntInput, SecretStrInput, ) from langflow.schema.dataframe import DataFrame class PostgresChunkFetcher(Component): display_name = "Postgres Chunk Fetcher" description = "Candidates(chunk_id 리스트)를 받아 Postgres 뷰/테이블에서 chunk_text/metadata 등을 가져옵니다." icon = "database" name = "PostgresChunkFetcher" inputs = [ DataFrameInput(name="candidates", display_name="Candidates (DataFrame)", required=True), StrInput(name="pg_host", display_name="PG Host", required=True), IntInput(name="pg_port", display_name="PG Port", value=5432, required=True), StrInput(name="pg_db", display_name="PG DB", required=True), StrInput(name="pg_user", display_name="PG User", required=True), SecretStrInput(name="pg_password", display_name="PG Password", required=True), StrInput(name="view_name", display_name="View/Table", value="public.audit_flat_chunks", required=True), IntInput(name="max_chunks", display_name="Max chunks to fetch", value=10, required=True), ] outputs = [ Output(name="docs", display_name="Docs (DataFrame)", method="build"), ] # ---------- helpers ---------- def _df_to_rows(self, df: Any) -> List[Dict[str, Any]]: """ LangFlow DataFrame은 환경/버전에 따라 내부 구조가 다를 수 있어 방어적으로 파싱합니다. 기대 형태: list[dict] (각 dict가 row) """ if df is None: return [] # 1) langflow.schema.dataframe.DataFrame if isinstance(df, DataFrame): data = getattr(df, "data", None) if isinstance(data, list): out: List[Dict[str, Any]] = [] for r in data: if isinstance(r, dict): out.append(r) elif hasattr(r, "data") and isinstance(getattr(r, "data"), dict): out.append(getattr(r, "data")) elif hasattr(r, "model_dump"): out.append(r.model_dump()) else: out.append({"value": r}) return out # 혹시 model_dump에 data가 있을 때 if hasattr(df, "model_dump"): dumped = df.model_dump() if isinstance(dumped, dict) and isinstance(dumped.get("data"), list): return [r if isinstance(r, dict) else {"value": r} for r in dumped["data"]] # 2) dict 형태 if isinstance(df, dict) and isinstance(df.get("data"), list): return [r if isinstance(r, dict) else {"value": r} for r in df["data"]] # 3) 이미 list인 경우 if isinstance(df, list): return [r if isinstance(r, dict) else {"value": r} for r in df] # 4) pandas DataFrame 들어오는 경우(드뭄) try: import pandas as pd # type: ignore if isinstance(df, pd.DataFrame): return df.to_dict(orient="records") except Exception: pass return [] def _extract_chunk_ids(self, rows: List[Dict[str, Any]]) -> List[str]: """ Candidates에서 chunk_id 추출. 자주 나오는 키들을 모두 커버합니다. """ id_keys = ["chunk_id", "id", "chunkId", "node.chunk_id"] ids: List[str] = [] for r in rows: cid = None # 1) 평면 키 for k in id_keys: if k in r and r[k]: cid = r[k] break # 2) 중첩 구조(node: {chunk_id: ...}) 케이스 if cid is None: node = r.get("node") if isinstance(node, dict) and node.get("chunk_id"): cid = node.get("chunk_id") if cid is not None: ids.append(str(cid)) # 중복 제거(순서 유지) seen = set() uniq = [] for x in ids: if x not in seen: seen.add(x) uniq.append(x) return uniq # ---------- main ---------- def build(self) -> DataFrame: cand_rows = self._df_to_rows(self.candidates) chunk_ids = self._extract_chunk_ids(cand_rows) if not chunk_ids: # 여기서 조용히 빈 결과로 보내면 원인 파악이 어려워서 "명시적으로" 실패 처리 sample_keys = list(cand_rows[0].keys()) if cand_rows else [] raise ValueError( f"[PostgresChunkFetcher] No chunk_ids extracted from candidates. " f"rows={len(cand_rows)}, sample_keys={sample_keys}" ) chunk_ids = chunk_ids[: int(self.max_chunks)] placeholders = ",".join(["%s"] * len(chunk_ids)) sql = f""" SELECT a.chunk_id::text AS chunk_id, a.doc_code, a.sub_code, a.chunk_text, a.metadata FROM {self.view_name} a WHERE a.chunk_id::text IN ({placeholders}); """.strip() conn = psycopg2.connect( host=self.pg_host, port=int(self.pg_port), dbname=self.pg_db, user=self.pg_user, password=self.pg_password, ) try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute(sql, tuple(chunk_ids)) rows = cur.fetchall() # list[dict] # Postgres 결과를 후보 순서대로 정렬(있으면) by_id = {str(r.get("chunk_id")): r for r in rows} ordered = [by_id[cid] for cid in chunk_ids if cid in by_id] # 혹시 추가로 온 rows가 있으면 뒤에 붙임 for r in rows: cid = str(r.get("chunk_id")) if cid not in set(chunk_ids): ordered.append(r) return DataFrame(data=ordered) finally: conn.close()
(5) Parser
- DataFrame(Docs)을 LLM이 읽을 수 있는 context 텍스트로 변환
- (권장) TopN/PreviewChars로 토큰 폭주 방지
(6) Prompt Template
- “근거팩(retrieved_docs)”만 사용하도록 강제
- 출력 포맷 강제(결론/근거/추가확인)
아래 템플릿을 사용:
당신은 회계·감사 분야에 전문성을 가진 상급 감사 전문가입니다.
당신의 답변은 반드시 아래 근거팩(retrieved_docs)에 포함된 내용에만 기반해야 합니다.
근거팩에 없는 사실·규정·처분·기관명·시점·금액·절차는 추정/창작 금지이며, 필요한 경우 “근거 부족”으로 표시하십시오.
[사용자 질문]
{question}
[근거팩(retrieved_docs)]
{retrieved_docs}
────────────────────────────────
작성 원칙(엄격)
1) 근거 우선: 결론은 근거팩에서 확인되는 사실/조치 유형/처분수위 범위 내에서만 제시한다.
2) 식별자 필수: 근거를 제시할 때는 반드시 (doc_code/sub_code/chunk_id)를 붙인다.
3) 최대 3개 근거: 근거는 가장 직접적인 3개만 선택한다(중복 내용은 1개로 합친다).
4) 불확실성 표기: 근거팩에 처분/조치 수준이 명시되지 않으면 단정하지 말고 “근거 부족”으로 처리한다.
5) 용어 제한: 근거팩에 실제로 등장하는 용어(예: 주의/통보/경고/문책/징계요구/환수/수사의뢰 등)만 사용한다.
6) 사실·의견 분리: 사실(근거에서 확인)과 권고(추가 확인 필요 시 제안)는 분리해 작성한다.
7) 과잉 일반화 금지: 특정 기관/사례의 조치를 전체 규정처럼 일반화하지 않는다.
────────────────────────────────
출력 형식(반드시 준수)
1) 결론(한 줄)
- (질문에 대한 최종 판단을 1문장으로 작성)
- 단, 근거팩에 처분/조치 수준이 직접 없으면: “근거 부족: 처분수위를 확정하기 어렵습니다.” 형태로 작성
2) 근거(최대 3개)
- (doc_code/sub_code/chunk_id) 근거 요지: 근거팩에서 확인되는 핵심 사실·조치·판단 기준을 1~2문장으로 요약
- (doc_code/sub_code/chunk_id) 근거 요지: ...
- (doc_code/sub_code/chunk_id) 근거 요지: ...
3) 처분/조치 판단 로직(근거 기반)
- 근거팩에 명시된 조치 유형/처분수위가 있다면: “근거에 따르면 ○○(조치 유형/처분) 수준이 적용됨”을 1문장으로 정리
- 명시가 없다면: “근거팩에 처분/조치 수준 직접 언급 없음 → 근거 부족”이라고 명시
4) 추가 확인 필요(있을 때만)
- 부족한 정보: (예: 위반 사실 확정 여부, 내부 규정/복무규정 조항, 승인 절차 여부, 고의·반복성, 금전 피해 등)
- 확인 이유: (왜 처분수위 판단에 필수인지 1문장)
- 필요한 추가 데이터 예: (필요한 문서/필드/기간 등)
────────────────────────────────
답변 스타일
- 간결하고 단정하되, 근거가 없으면 단정하지 않는다.
- 사용자가 “추천”을 요청하더라도 근거팩에 처분 관련 선례/규정이 없으면 추천을 확정하지 말고 “추가 확인 필요”로 돌린다.3. 기대 효과 / 한계
- 기대 효과: 단일 질의 기반으로도 “근거 기반 답변”은 안정화
- 한계: 사용자 표현이 애매하거나 용어가 다양한 경우 recall 부족 가능
[Version 2] audit_rag_MultiQuery
목표: Baseline(단일 질의) 대비 MultiQuery(질의 확장)로 recall을 올리고,
Neo4j(Vector+Fulltext)에서 더 넓게 후보를 모은 뒤 Postgres(audit_flat_chunks)로 원문을 붙여 근거 기반 답변을 생성한다.
1. 전체 플로우
[1] Chat Input
↓ (사용자 질문)
[2] OpenAI(ChatModel) ← MultiQuery 전용 (저렴한 모델 권장)
↓ (LLM handle)
[3] MultiQuery Generator (Custom)
↓ (Queries DataFrame: query[])
[4] OpenAI Embeddings
↓ (Embeddings handle)
[5] Neo4j Hybrid Retriever (MQ, Custom)
↓ (Candidates DataFrame: chunk_id/doc_code/sub_code/score...)
[6] Postgres Chunk Fetcher (Custom)
↓ (Docs DataFrame: chunk_id/chunk_text/metadata/doc_code/sub_code)
[7] Parser (Docs DF → retrieved_docs 텍스트)
↓
[8] Prompt Template
↓
[9] OpenAI (LLM)
↓
[10] Chat Output2) Defualt 설정값(수정 가능)
(A) MultiQuery Generator
n_queries→ 4- 사용자 질문을 몇 개의 “검색용 확장 질의”로 만들지(= Multi-Query 개수)
include_original→ True- MultiQuery에 “원 질문”을 무조건 포함할지 여부.
(B) Neo4j Hybrid Retriever (MQ)
search_mode→hybrid- vector_only: 임베딩 기반 벡터 검색만
- fulltext_only: 키워드/BM25 계열 풀텍스트 검색만
- hybrid: 둘 다 수행 후 합쳐서 최종 후보 생성
k_vec→ 25- 벡터 검색에서 “후보 풀”로 몇 개를 뽑을지
- MultiQuery라 호출이 늘어나므로 pool은 살짝 줄이는 편 권장
k_ft→ 25- 풀텍스트 검색에서 후보 풀로 몇 개를 뽑을지
topk→ 10- 최종적으로 Retriever가 반환하는 후보 chunk 개수
weight_vec/weight_txt→ 0.5- hybrid 모드에서 벡터/풀텍스트 결과를 합칠 때의 상대적 비중
rrf_k→ 60- RRF(Reciprocal Rank Fusion)에서 사용하는 상수 k
- 1 / (rrf_k + rank) 형태로 계산
- rrf_k가 작을수록: 상위 몇 개 rank가 강하게 우세해짐
- rrf_k가 클수록: 상위 rank의 “쏠림”이 완화되고, 여러 질의/여러 소스의 후보가 비교적 고르게 섞임
(C) Postgres Chunk Fetcher
view_name:public.audit_flat_chunks- chunk 원문/메타데이터를 조회할 테이블 또는 뷰 이름
- 이후 total_vdb 로 변경 예정
max_chunks: 10- Postgres에서 실제로 가져올 chunk 개수 상한
2. Custom Component 코드
2-1) MultiQuery Generator (Custom)
역할: 사용자 질문을 여러 관점으로 확장하여
Queries(DataFrame)을 만든다.
출력 DataFrame은 반드시query컬럼을 가진다.
import json
import re
from typing import Any, Dict, List
from langflow.custom import Component
from langflow.io import Output, MessageTextInput, IntInput, BoolInput
from langflow.schema.dataframe import DataFrame
from langflow.schema.message import Message
try:
from langflow.io import HandleInput
except Exception:
from langflow.inputs.inputs import HandleInput
def _make_df(records: List[Dict[str, Any]]) -> DataFrame:
try:
return DataFrame(data=records) # type: ignore
except TypeError:
pass
try:
return DataFrame(value=records) # type: ignore
except TypeError:
pass
return DataFrame() # type: ignore
def _msg_text(x: Any) -> str:
if x is None:
return ""
if isinstance(x, str):
return x
if hasattr(x, "content"):
try:
return str(x.content or "")
except Exception:
pass
if hasattr(x, "text"):
try:
return str(x.text or "")
except Exception:
pass
return str(x)
def _extract_json_obj(text: str) -> Dict[str, Any]:
text = text.strip()
try:
obj = json.loads(text)
if isinstance(obj, dict):
return obj
except Exception:
pass
m = re.search(r"\{[\s\S]*\}", text)
if not m:
return {}
try:
obj = json.loads(m.group(0))
return obj if isinstance(obj, dict) else {}
except Exception:
return {}
class MultiQueryGenerator(Component):
display_name = "MultiQuery Generator"
description = "사용자 질문을 여러 검색 질의로 확장하여 Queries(DataFrame)를 생성"
icon = "sparkles"
name = "MultiQueryGenerator"
inputs = [
MessageTextInput(name="user_question", display_name="User Question", required=True),
HandleInput(
name="llm",
display_name="LLM (ChatModel Handle)",
input_types=["LanguageModel", "LLM", "ChatModel", "Any"],
required=True,
),
IntInput(name="n_queries", display_name="N Queries", value=4, required=True),
BoolInput(name="include_original", display_name="Include Original", value=True, required=True),
]
outputs = [
Output(name="queries_df", display_name="Queries (DataFrame)", method="build_queries_df"),
Output(name="debug", display_name="Debug (Message)", method="build_debug"),
]
def build_queries_df(self) -> DataFrame:
self._debug_lines = []
q = (_msg_text(self.user_question)).strip()
if not q:
self._debug_lines.append("[mq] empty user_question -> fallback empty df")
return _make_df([])
n = int(self.n_queries)
n = 1 if n < 1 else n
n = 8 if n > 8 else n # 과도한 비용 방지
prompt = f"""
너는 감사/회계 문서 검색을 위한 Query Generator다.
사용자 질문을 바탕으로 검색 질의 {n}개를 생성하라.
규칙:
- 의미가 같은 '원 질문' 1개는 반드시 포함한다.
- 나머지는 관점 확장: (1) 행위/사실관계 (2) 규정/절차/내부통제 (3) 처분/조치/선례 중심으로 만든다.
- 각 질의는 너무 길지 않게 작성한다.
- 결과는 반드시 JSON만 출력한다. 형식: {{"queries":["...","..."]}}
사용자 질문: {q}
""".strip()
llm = self.llm
if hasattr(llm, "invoke"):
resp = llm.invoke(prompt)
elif hasattr(llm, "predict"):
resp = llm.predict(prompt)
elif callable(llm):
resp = llm(prompt)
else:
raise TypeError("Unsupported LLM handle (need invoke/predict/callable).")
text = _msg_text(resp).strip()
self._debug_lines.append(f"[mq] raw_len={len(text)}")
obj = _extract_json_obj(text)
queries = obj.get("queries", [])
if not isinstance(queries, list):
queries = []
out: List[str] = []
for s in queries:
s = str(s).strip()
if s:
out.append(s)
if self.include_original and q not in out:
out.insert(0, q)
seen = set()
uniq = []
for s in out:
key = re.sub(r"\s+", " ", s).strip()
if key and key not in seen:
seen.add(key)
uniq.append(s)
self._debug_lines.append(f"[mq] queries={len(uniq)} sample={uniq[:3]}")
records = [{"query": s, "i": idx} for idx, s in enumerate(uniq)]
return _make_df(records)
def build_debug(self) -> Message:
return Message(text="\n".join(getattr(self, "_debug_lines", ["(no debug)"])))2-2) Neo4j Hybrid Retriever (MQ, Custom)
역할: Queries(DataFrame)의 각 query로 Neo4j 검색(Vector/Fulltext)을 수행하고,
후보를 RRF(Reciprocal Rank Fusion)로 합쳐 Candidates(DataFrame)을 반환한다.
출력 DataFrame은 반드시 chunk_id 컬럼을 가진다(→ Postgres Fetcher가 조회 가능).
from __future__ import annotations
import re
from typing import Any, Dict, List
from neo4j import GraphDatabase
from langflow.custom import Component
from langflow.io import (
Output, MessageTextInput, StrInput, IntInput, FloatInput, SecretStrInput, DropdownInput, DataFrameInput
)
from langflow.schema.dataframe import DataFrame
from langflow.schema.message import Message
try:
from langflow.io import HandleInput
except Exception:
from langflow.inputs.inputs import HandleInput
def _msg_text(x: Any) -> str:
if x is None:
return ""
if isinstance(x, str):
return x
if hasattr(x, "content"):
try:
return str(x.content or "")
except Exception:
pass
if hasattr(x, "text"):
try:
return str(x.text or "")
except Exception:
pass
return str(x)
def _make_df(records: List[Dict[str, Any]]) -> DataFrame:
try:
return DataFrame(data=records) # type: ignore
except TypeError:
pass
try:
return DataFrame(value=records) # type: ignore
except TypeError:
pass
return DataFrame() # type: ignore
def _df_to_rows(df: Any) -> List[Dict[str, Any]]:
if df is None:
return []
if isinstance(df, DataFrame):
data = getattr(df, "data", None)
if isinstance(data, list):
return [r if isinstance(r, dict) else {"value": r} for r in data]
if hasattr(df, "model_dump"):
dumped = df.model_dump()
if isinstance(dumped, dict) and isinstance(dumped.get("data"), list):
return [r if isinstance(r, dict) else {"value": r} for r in dumped["data"]]
if isinstance(df, dict) and isinstance(df.get("data"), list):
return [r if isinstance(r, dict) else {"value": r} for r in df["data"]]
if isinstance(df, list):
return [r if isinstance(r, dict) else {"value": r} for r in df]
return []
def _embed_query(embeddings_model: Any, text: str) -> List[float]:
if embeddings_model is None:
raise ValueError("Embeddings Model is required.")
if hasattr(embeddings_model, "embed_query"):
vec = embeddings_model.embed_query(text)
return [float(v) for v in list(vec)]
if hasattr(embeddings_model, "embed_documents"):
vecs = embeddings_model.embed_documents([text])
vec = vecs[0] if vecs else []
return [float(v) for v in list(vec)]
raise TypeError("Unsupported embeddings model: need embed_query or embed_documents.")
class Neo4jHybridRetrieverMQ(Component):
display_name = "Neo4j Hybrid Retriever (MQ)"
description = "Queries(DataFrame)를 받아 Vector/Fulltext/Hybrid 검색 후 RRF로 merge/dedupe하여 Candidates(DataFrame) 반환"
icon = "database"
name = "Neo4jHybridRetrieverMQ"
inputs = [
MessageTextInput(name="user_question", display_name="User Question (fallback)", required=True),
DataFrameInput(name="queries_df", display_name="Queries (DataFrame, column=query)", required=False),
HandleInput(
name="embeddings_model",
display_name="Embeddings Model",
input_types=["embeddings", "Embeddings", "Any"],
required=True,
),
StrInput(name="neo4j_uri", display_name="Neo4j URI", value="bolt://host.docker.internal:7687", required=True),
StrInput(name="neo4j_user", display_name="Neo4j User", value="neo4j", required=True),
SecretStrInput(name="neo4j_password", display_name="Neo4j Password", required=True),
StrInput(name="neo4j_database", display_name="Neo4j DB", value="auditdb2", required=True),
StrInput(name="vector_index_name", display_name="Vector Index", value="chunk_embedding_idx", required=True),
StrInput(name="fulltext_index_name", display_name="Fulltext Index", value="chunk_text_ft", required=True),
DropdownInput(
name="search_mode",
display_name="Search Mode",
options=["hybrid", "vector_only", "fulltext_only"],
value="hybrid",
required=True,
),
IntInput(name="k_vec", display_name="Vector TopK (pool)", value=25, required=True),
IntInput(name="k_ft", display_name="Fulltext TopK (pool)", value=25, required=True),
IntInput(name="topk", display_name="Final TopK", value=10, required=True),
FloatInput(name="weight_vec", display_name="Weight (vector)", value=0.5, required=True),
FloatInput(name="weight_txt", display_name="Weight (fulltext)", value=0.5, required=True),
IntInput(name="rrf_k", display_name="RRF k (default 60)", value=60, required=True),
]
outputs = [
Output(name="candidates_df", display_name="Candidates (DataFrame)", method="build"),
Output(name="debug", display_name="Debug (Message)", method="build_debug"),
]
VECTOR_Q = """
CALL db.index.vector.queryNodes($vector_index, $k, $qvec)
YIELD node, score
RETURN
node.chunk_id AS chunk_id,
node.doc_code AS doc_code,
node.sub_code AS sub_code,
score AS vec_score
ORDER BY vec_score DESC
LIMIT $k
"""
FULLTEXT_Q_OPT = """
CALL db.index.fulltext.queryNodes($fulltext_index, $q, {limit: $k})
YIELD node, score
RETURN
node.chunk_id AS chunk_id,
node.doc_code AS doc_code,
node.sub_code AS sub_code,
score AS txt_score
"""
FULLTEXT_Q_FALLBACK = """
CALL db.index.fulltext.queryNodes($fulltext_index, $q)
YIELD node, score
RETURN
node.chunk_id AS chunk_id,
node.doc_code AS doc_code,
node.sub_code AS sub_code,
score AS txt_score
ORDER BY txt_score DESC
LIMIT $k
"""
def build_debug(self) -> Message:
return Message(text="\n".join(getattr(self, "_debug_lines", ["(no debug)"])))
def _neo4j_run(self, query: str, params: Dict[str, Any]) -> List[Dict[str, Any]]:
driver = GraphDatabase.driver(self.neo4j_uri, auth=(self.neo4j_user, self.neo4j_password))
try:
with driver.session(database=self.neo4j_database) as s:
res = s.run(query, params)
return [dict(r) for r in res]
finally:
driver.close()
@staticmethod
def _norm(s: str) -> str:
return re.sub(r"\s+", " ", (s or "")).strip()
def _collect_queries(self) -> List[str]:
q0 = self._norm(_msg_text(self.user_question))
rows = _df_to_rows(getattr(self, "queries_df", None))
qs: List[str] = []
for r in rows:
v = r.get("query") or r.get("normalized_query") or r.get("q")
if v:
qs.append(self._norm(str(v)))
if not qs and q0:
qs = [q0]
elif q0 and q0 not in qs:
qs.insert(0, q0)
seen = set()
uniq = []
for s in qs:
if s and s not in seen:
seen.add(s)
uniq.append(s)
return uniq[:8]
def build(self) -> DataFrame:
self._debug_lines = []
mode = (self.search_mode or "hybrid").strip()
k_vec = int(self.k_vec)
k_ft = int(self.k_ft)
topk = int(self.topk)
wv = float(self.weight_vec)
wt = float(self.weight_txt)
if (wv + wt) <= 0:
wv, wt = 0.5, 0.5
rrf_k = int(self.rrf_k) if int(self.rrf_k) > 0 else 60
queries = self._collect_queries()
self._debug_lines.append(f"[mq] queries={len(queries)} sample={queries[:3]}")
if not queries:
return _make_df([])
fused: Dict[str, Dict[str, Any]] = {}
def add_rrf(cid: str, add: float):
fused.setdefault(cid, {
"chunk_id": cid,
"doc_code": None,
"sub_code": None,
"score": 0.0,
"vec_score": 0.0,
"txt_score": 0.0,
"source": "",
"best_query": None,
})
fused[cid]["score"] = float(fused[cid]["score"]) + float(add)
for qi, q in enumerate(queries):
qvec = _embed_query(self.embeddings_model, q)
vec_rows: List[Dict[str, Any]] = []
txt_rows: List[Dict[str, Any]] = []
if mode in ("hybrid", "vector_only"):
vec_rows = self._neo4j_run(self.VECTOR_Q, {
"vector_index": self.vector_index_name,
"k": k_vec,
"qvec": qvec,
})
if mode in ("hybrid", "fulltext_only"):
try:
txt_rows = self._neo4j_run(self.FULLTEXT_Q_OPT, {
"fulltext_index": self.fulltext_index_name,
"q": q,
"k": k_ft,
})
except Exception:
txt_rows = self._neo4j_run(self.FULLTEXT_Q_FALLBACK, {
"fulltext_index": self.fulltext_index_name,
"q": q,
"k": k_ft,
})
self._debug_lines.append(f"[q{qi}] vec={len(vec_rows)} txt={len(txt_rows)} q='{q[:40]}'")
for rank, r in enumerate(vec_rows, start=1):
cid = self._norm(str(r.get("chunk_id") or ""))
if not cid:
continue
add_rrf(cid, wv / float(rrf_k + rank))
fused[cid]["doc_code"] = fused[cid]["doc_code"] or r.get("doc_code")
fused[cid]["sub_code"] = fused[cid]["sub_code"] or r.get("sub_code")
fused[cid]["vec_score"] = max(float(fused[cid]["vec_score"]), float(r.get("vec_score") or 0.0))
fused[cid]["best_query"] = fused[cid]["best_query"] or q
src = set((fused[cid]["source"] or "").split("+")) if fused[cid]["source"] else set()
src.add("vec")
fused[cid]["source"] = "+".join(sorted([s for s in src if s]))
for rank, r in enumerate(txt_rows, start=1):
cid = self._norm(str(r.get("chunk_id") or ""))
if not cid:
continue
add_rrf(cid, wt / float(rrf_k + rank))
fused[cid]["doc_code"] = fused[cid]["doc_code"] or r.get("doc_code")
fused[cid]["sub_code"] = fused[cid]["sub_code"] or r.get("sub_code")
fused[cid]["txt_score"] = max(float(fused[cid]["txt_score"]), float(r.get("txt_score") or 0.0))
fused[cid]["best_query"] = fused[cid]["best_query"] or q
src = set((fused[cid]["source"] or "").split("+")) if fused[cid]["source"] else set()
src.add("txt")
fused[cid]["source"] = "+".join(sorted([s for s in src if s]))
out = list(fused.values())
out.sort(key=lambda x: float(x.get("score") or 0.0), reverse=True)
out = out[:topk]
self._debug_lines.append(f"[final] unique={len(fused)} topk={len(out)}")
records = []
for r in out:
records.append({
"chunk_id": r.get("chunk_id"),
"score": float(r.get("score") or 0.0),
"vec_score": float(r.get("vec_score") or 0.0),
"txt_score": float(r.get("txt_score") or 0.0),
"doc_code": r.get("doc_code"),
"sub_code": r.get("sub_code"),
"source": r.get("source"),
"best_query": r.get("best_query"),
})
return _make_df(records)3. 기대 효과 / 비용 트레이드오프
-
기대 효과
- 단일 질문에서 놓칠 수 있는 표현/용어/관점을 질의로 확장하여 recall 상승
- Neo4j vector/fulltext 후보를 더 넓게 모아 유사사례 매칭 확률 증가
-
트레이드오프
- 질의 개수(n_queries)만큼 Neo4j 검색 호출(벡터+FT)이 늘어 지연/비용 증가
- 보수적으로 n_queries=4부터 시작 권장
[Version 3] audit_rag_Rerank — OpenAI 기반 Listwise ReRanker 추가
목표: 베이스(Version 1) / 멀티쿼리(Version 2)까지 안정화한 뒤,
후보 문서 중 “답변에 실제로 쓸 근거”를 더 정확히 고르기 위해 ReRanker(재정렬) 단계 추가
1. 전체 플로우
[1] Chat Input
↓ (사용자 질문)
[2] OpenAI(ChatModel) ← MultiQuery 전용 (저렴한 모델 권장)
↓ (LLM handle)
[3] MultiQuery Generator (Custom)
↓ (Queries DataFrame: query[])
[4] OpenAI Embeddings
↓ (Embeddings handle)
[5] Neo4j Hybrid Retriever (MQ, Custom)
↓ (Candidates DataFrame: chunk_id/doc_code/sub_code/score...)
[6] Postgres Chunk Fetcher (Custom)
↓ (Docs DataFrame: chunk_id/chunk_text/metadata/doc_code/sub_code)
[7] OpenAI Listwise ReRanker
↓
[8] Parser (Docs DF → retrieved_docs 텍스트)
↓
[9] Prompt Template
↓
[10] OpenAI (LLM)
↓
[11] Chat Output2. 설정값 (수정 가능)
(A) Postgres Chunk Fetcher
view_name:public.audit_flat_chunksmax_chunks: 30- ReRanker로 “선별”할 수 있게 후보 풀을 넓혀둠
(B) OpenAI Listwise ReRanker (NEW)
max_docs: 30 (리랭크 입력 후보 상한)top_n: 10 (리랭크 출력 Top-N)max_chars_per_doc: 1500~2000temperature: 0.0 (재현성/일관성)max_completion_tokens: 800debug: 처음엔 True, 안정화 후 False
3. Rerank → Listwise
LLM Rerank의 3가지 방식
A. Pointwise (문서별 점수화)
- 문서 i 하나씩 놓고 “관련성 점수(0~1)”를 매김
- 장점: 구현 쉬움, 디버깅 쉬움
- 단점: N개면 N번 판단(호출/토큰 증가), 문서 간 상대 비교가 약함
B. Pairwise (두 문서 비교 반복)
- “A vs B 중 더 관련 있는 것?”을 여러 번 반복해 순위 구성
- 장점: 비교가 명확하면 정확
- 단점: 조합이 많아지면 폭발(N²), 구현 번거로움
C. Listwise (한 번에 목록 정렬)
- 후보 목록을 한 번에 보여주고 “Top K와 순서”를 뽑게 함
- 장점: 호출 1회로 끝나는 경우가 많아 효율적, “상대 비교”에 강함
- 단점: 입력 토큰이 커지며, 문서가 너무 길면 비용/지연 증가
전용 Rerank API
A. Jina Rerank
B. Voyage Rerank
C. Cohere Rerank
OpenAI Listwise ReRanker
from __future__ import annotations
import json
from typing import Any, Dict, List, Optional
import requests
from langflow.custom import Component
from langflow.io import (
DataFrameInput,
Output,
MessageTextInput,
StrInput,
IntInput,
FloatInput,
SecretStrInput,
BoolInput,
)
from langflow.schema.dataframe import DataFrame
class OpenAIListwiseReRanker(Component):
display_name = "OpenAI Listwise ReRanker"
description = "Docs(DataFrame)를 OpenAI LLM으로 listwise rerank하여 상위 N개를 반환 (Structured Outputs 지원)"
icon = "sort"
name = "OpenAIListwiseReRanker"
inputs = [
DataFrameInput(name="docs", display_name="Docs (DataFrame)", required=True),
MessageTextInput(name="user_question", display_name="User Question", required=True),
SecretStrInput(name="openai_api_key", display_name="OpenAI API Key", required=True),
StrInput(name="model", display_name="Model", value="gpt-4o-mini", required=True),
IntInput(name="top_n", display_name="Top N", value=10, required=True),
IntInput(name="max_docs", display_name="Max docs (input cap)", value=30, required=True),
IntInput(name="max_chars_per_doc", display_name="Max chars per doc", value=2000, required=True),
FloatInput(name="temperature", display_name="Temperature", value=0.0, required=True),
IntInput(name="max_completion_tokens", display_name="Max completion tokens", value=800, required=True),
BoolInput(name="strict_fail", display_name="Strict Fail (raise on error)", value=False, required=True),
BoolInput(name="debug", display_name="Debug (attach debug fields)", value=False, required=True),
]
outputs = [
Output(name="reranked_docs", display_name="ReRanked Docs (DataFrame)", method="build"),
]
# ----------------------------
# Helpers: LangFlow DataFrame -> rows(list[dict])
# ----------------------------
def _df_to_rows(self, df: Any) -> List[Dict[str, Any]]:
if df is None:
return []
# 1) langflow.schema.dataframe.DataFrame
if isinstance(df, DataFrame):
data = getattr(df, "data", None)
if isinstance(data, list):
out: List[Dict[str, Any]] = []
for r in data:
if isinstance(r, dict):
out.append(r)
elif hasattr(r, "data") and isinstance(getattr(r, "data"), dict):
out.append(getattr(r, "data"))
elif hasattr(r, "model_dump"):
out.append(r.model_dump())
else:
out.append({"value": r})
return out
# 1-b) dataframe 속성(pandas)인 경우
pdf = getattr(df, "dataframe", None)
if pdf is not None:
try:
return pdf.to_dict(orient="records")
except Exception:
pass
if hasattr(df, "model_dump"):
dumped = df.model_dump()
if isinstance(dumped, dict) and isinstance(dumped.get("data"), list):
return [r if isinstance(r, dict) else {"value": r} for r in dumped["data"]]
# 2) dict 형태
if isinstance(df, dict) and isinstance(df.get("data"), list):
return [r if isinstance(r, dict) else {"value": r} for r in df["data"]]
# 3) 이미 list인 경우
if isinstance(df, list):
return [r if isinstance(r, dict) else {"value": r} for r in df]
# 4) pandas DataFrame
try:
import pandas as pd # type: ignore
if isinstance(df, pd.DataFrame):
return df.to_dict(orient="records")
except Exception:
pass
return []
def _make_lf_df(self, records: List[Dict[str, Any]]) -> DataFrame:
# LangFlow 버전별 생성 시그니처 방어
try:
return DataFrame(data=records) # type: ignore
except TypeError:
pass
try:
return DataFrame(value=records) # type: ignore
except TypeError:
pass
return DataFrame() # type: ignore
def _clip(self, s: Any, n: int) -> str:
if s is None:
return ""
t = str(s)
return t[:n]
# ----------------------------
# OpenAI call (Chat Completions)
# ----------------------------
def _openai_chat(self, api_key: str, payload: Dict[str, Any]) -> Dict[str, Any]:
url = "https://api.openai.com/v1/chat/completions"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
}
r = requests.post(url, headers=headers, json=payload, timeout=90)
if r.status_code >= 400:
raise RuntimeError(f"OpenAI API error {r.status_code}: {r.text}")
return r.json()
def build(self) -> DataFrame:
question = (self.user_question or "").strip()
if not question:
raise ValueError("user_question is empty")
rows = self._df_to_rows(self.docs)
if not rows:
# 입력이 비어있으면 그대로 반환
return self._make_lf_df([])
# rerank 입력 제한
max_docs = max(1, int(self.max_docs))
cand = rows[:max_docs]
# 텍스트 필드 추정 (당신 스키마: chunk_text)
# 없으면 chunk / text / content 순으로 fallback
def get_text(r: Dict[str, Any]) -> str:
for k in ("chunk_text", "chunk", "text", "content"):
if k in r and r[k]:
return str(r[k])
return ""
# 식별자 추정
def get_id(r: Dict[str, Any]) -> str:
for k in ("chunk_id", "id"):
if k in r and r[k]:
return str(r[k])
return ""
# LLM에 넘길 문서 목록 구성
max_chars = max(200, int(self.max_chars_per_doc))
items = []
for i, r in enumerate(cand):
items.append(
{
"i": i,
"chunk_id": get_id(r),
"doc_code": r.get("doc_code"),
"sub_code": r.get("sub_code"),
"text": self._clip(get_text(r), max_chars),
}
)
system = (
"당신은 감사/준법 RAG 시스템을 위한 '엄격한 리랭킹 함수'입니다.\n"
"사용자 질문(question)과 후보 문서 청크 목록(items)을 입력으로 받아, 후보들의 순위를 반환하세요.\n"
"우선순위 기준:\n"
"1) 질문에 대한 직접 관련성\n"
"2) 근거로서의 유용성(증거성)\n"
"3) 구체성(사실/절차/처분/규정 조항 등 구체 정보 포함)\n"
"4) 중복 최소화(서로 같은 내용이면 하나만 상위로)\n"
"절대 질문에 답변하지 마세요. 오직 '순위'만 반환하세요."
)
user = {
"question": question,
"items": items,
"task": (
"아래 형식의 JSON 객체만 반환하세요:\n"
"- order: 항목 인덱스 배열(가장 적합한 것부터 최하위까지)\n"
"- scores: 0..100 점수 배열(order와 같은 순서로 정렬) (선호, 선택)\n"
"- reasons: 짧은 사유 문자열 배열(order와 같은 순서로 정렬) (선호, 선택)\n"
"제약:\n"
"- order에는 유효한 인덱스만 포함하고, 중복 인덱스는 금지합니다.\n"
"- JSON 외의 설명/문장/코드블록은 출력하지 마세요."
),
}
schema = {
"name": "rerank_result",
"schema": {
"type": "object",
"additionalProperties": False,
"properties": {
"order": {"type": "array", "items": {"type": "integer"}},
"scores": {"type": "array", "items": {"type": "number"}},
"reasons": {"type": "array", "items": {"type": "string"}},
},
"required": ["order"],
},
}
# 1) json_schema (Structured Outputs)
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": json.dumps(user, ensure_ascii=False)},
],
"temperature": float(self.temperature),
"max_completion_tokens": int(self.max_completion_tokens),
"response_format": {"type": "json_schema", "json_schema": schema},
}
api_key = str(self.openai_api_key)
try:
resp = self._openai_chat(api_key, payload)
content = resp["choices"][0]["message"]["content"]
data = json.loads(content) if isinstance(content, str) else content
except Exception as e:
# 2) fallback: json_object (older JSON mode)
try:
payload2 = dict(payload)
payload2["response_format"] = {"type": "json_object"}
resp2 = self._openai_chat(api_key, payload2)
content2 = resp2["choices"][0]["message"]["content"]
data = json.loads(content2) if isinstance(content2, str) else content2
except Exception as e2:
if bool(self.strict_fail):
raise
# 실패 시 원본 유지
if bool(self.debug):
out = []
for r in rows[: int(self.top_n)]:
rr = dict(r)
rr["_rerank_error"] = f"{type(e).__name__}: {e} / fallback: {type(e2).__name__}: {e2}"
out.append(rr)
return self._make_lf_df(out)
return self._make_lf_df(rows[: int(self.top_n)])
order = data.get("order", [])
scores = data.get("scores", [])
reasons = data.get("reasons", [])
# order 정제: 범위 밖/중복 제거
seen = set()
cleaned: List[int] = []
for idx in order:
try:
j = int(idx)
except Exception:
continue
if 0 <= j < len(cand) and j not in seen:
seen.add(j)
cleaned.append(j)
# fallback: order가 비면 그대로
if not cleaned:
cleaned = list(range(len(cand)))
top_n = max(1, int(self.top_n))
cleaned = cleaned[:top_n]
# 결과 조립 (원본 row를 재정렬)
out_rows: List[Dict[str, Any]] = []
for rank, j in enumerate(cleaned, start=1):
base = dict(cand[j])
# score/reason 매핑
sc = None
rs = None
if isinstance(scores, list) and (rank - 1) < len(scores):
try:
sc = float(scores[rank - 1])
except Exception:
sc = None
if isinstance(reasons, list) and (rank - 1) < len(reasons):
rs = str(reasons[rank - 1])
base["rerank_rank"] = rank
if sc is not None:
base["rerank_score"] = sc
if rs:
base["rerank_reason"] = rs
if bool(self.debug):
base["_rerank_raw"] = data
out_rows.append(base)
return self._make_lf_df(out_rows)4. 기대효과
- 검색 품질 개선
- Retriever(Neo4j Hybrid) : 유사 표현/키워드 변화에도 관련 사례 유입률(Recall@K) 상승
- ReRanker(OpenAI Listwise) : 최종 근거팩의 정밀도 상승, 중복/배경성 문서 감소
[Version 4] audit_rag_Evidence
Version 3에서 OpenAI Listwise ReRanker로 후보 문서의 순서를 재정렬했다면,
Version 4에서는 Evidence Pack Builder를 추가해서 LLM에 주는 근거팩(retrieved_docs)을 일관된 포맷으로 만들고, 중복 제거/길이 제한/인용 키 강제까지 한 번에 처리한다.
0. Evidence Pack Builder 추가 이유
ReRanker 결과(DataFrame)를 그대로 Prompt에 넣었을 때 생길 수 있는 문제 방지
- 토큰 폭발: chunk_text가 길고, 중복 chunk가 섞이면 비용/지연이 급격히 증가
- 인용 포맷 불안정: LLM이 (doc_code/sub_code/chunk_id)를 놓치거나 형태가 흔들림
- 중복 근거: 같은 sub_code(같은 사례)에서 여러 chunk가 떠도, 답변에 같은 내용이 반복
- 근거팩 구조 불명확: LLM이 근거를 어떻게 써야 하는지 파악하기 어려움
→ 상위 N개만 뽑고, item당 최대 글자수로 자르고, sub_code 기준 dedupe해서 중복 제거하고, (doc_code/sub_code/chunk_id) 인용 키를 강제한 근거팩 문자열(Message)로 변환
1. 전체 플로우
[1] Chat Input
↓ (사용자 질문)
[2] OpenAI(ChatModel) ← MultiQuery 전용 (저렴한 모델 권장)
↓ (LLM handle)
[3] MultiQuery Generator (Custom)
↓ (Queries DataFrame: query[])
[4] OpenAI Embeddings
↓ (Embeddings handle)
[5] Neo4j Hybrid Retriever (MQ, Custom)
↓ (Candidates DataFrame: chunk_id/doc_code/sub_code/score...)
[6] Postgres Chunk Fetcher (Custom)
↓ (Docs DataFrame: chunk_id/chunk_text/metadata/doc_code/sub_code)
[7] OpenAI Listwise ReRanker
↓
[8] Evidence Pack Builder
↓
[9] Prompt Template
↓
[10] OpenAI (LLM)
↓
[11] Chat Output2. Evidence Pack Builder 권장 파라미터
-
Max items in evidence pack(=10): 근거팩에 포함할 최대 문서 조각(chunk) 개수
-
Max chars per item(=900): 각 chunk_text를 최대 N 글자까지만 잘라서 근거팩에 넣습니다
-
Dedupe by sub_code(=ON): 같은 sub_code에서 나온 chunk가 여러 개 있어도 대표 1개만 남기고 제거
-
Include loc.lines if exists(=ON): metadata에 loc.lines.from/to가 있으면 근거 항목에 같이 표기
- 근거의 “문서 내 위치”를 같이 주면 LLM이 인용을 더 정확하게 하거나 사람이 검증할 때 원문 추적이 쉬워짐
-
Format style(=markdown): 근거팩을 마크다운 스타일로 포맷해서 Prompt에
-
(옵션) Strict fail when empty(=ON): (평가/디버깅 때 사용, 운영에서는 OFF도 가능)
- 입력 docs가 비거나 파싱 결과가 0행이면 즉시 에러를 발생시킵니다(Flow를 실패 처리)
3. 기대효과
- 비용·지연(토큰) 안정화
- max_items, max_chars_per_item로 컨텍스트 상한을 강제
- 결과적으로 질문 난이도/문서 길이 편차가 커도 응답 시간과 비용이 예측 가능
- 특히 보고서 생성(서술형)에서는 “한두 개 chunk가 지나치게 길어” 프롬프트가 커지는 문제가 자주 있는데, 이를 구조적으로 차단
- 중복 근거 제거 → 답변 반복/편향 감소
- LLM이 같은 내용을 반복 인용하는 빈도 감소
- 한 사례에 과도하게 편향되는 현상 완화
- 근거 다양성(coverage)이 늘어 답변이 더 균형잡히는 경향
- 근거 인용 포맷 준수율 상승(감사/검증에 유리)
- 근거팩에 (doc_code/sub_code/chunk_id)를 항목마다 강제로 포함시키기 때문에 Prompt Template에서 근거(최대 3개) 섹션 작성 시 모델이 정확히 인용 키를 가져다 쓰기 쉬워짐
