diff --git a/.env b/.env index 45aa0e1..6a3fae5 100644 --- a/.env +++ b/.env @@ -102,10 +102,10 @@ DOCUMENT_PARSE_ARTIFACT_PREFIX=artifacts PARSER_FAILURE_MODE=fail # ===== Reranker 配置 ===== -RERANKER_ENABLED=true +RERANKER_ENABLED=false RERANKER_BASE_URL=http://6.86.80.4:30080/v1 RERANKER_MODEL=BAAI/bge-reranker-v2-m3 -RERANKER_API_KEY= +RERANKER_API_KEY=sk-fVr9KmDZNC4pGDBQj0EUWz9bDmFzNxjYC9EzZpe2bVDsxtz8 RERANKER_TOP_K=5 # ===== 会话持久化 ===== @@ -120,3 +120,10 @@ AUTH_ENABLED=true # ===== CORS ===== CORS_ALLOW_ORIGINS=http://localhost:5173 + +# ===== HyDE ???? ===== +HYDE_ENABLED=true +HYDE_MAX_TOKENS=200 +HYDE_LLM_PROVIDER=qwen +HYDE_LLM_MODEL=qwen3.5-flash + diff --git a/.env.example b/.env.example index 1bc8445..0aef281 100644 --- a/.env.example +++ b/.env.example @@ -138,6 +138,31 @@ AUTH_TOKEN_EXPIRE_MINUTES=480 # 设为 false 可跳过认证（仅限本地开发调试，生产必须 true） AUTH_ENABLED=true + +# ===== HyDE 查询增强 ===== +# HyDE (Hypothetical Document Embeddings): 在检索前让 LLM 生成一段"假设性回答"， +# 用该段落的 embedding 代替原始查询 embedding 进行向量检索。 +# 无需新模型，复用现有 LLM 和 Embedding 服务。降低此功能可减少每次查询的 LLM 调用次数。 +HYDE_ENABLED=true +HYDE_MAX_TOKENS=200 +# ?????? LLM;??????????????? +HYDE_LLM_PROVIDER=qwen +HYDE_LLM_MODEL=qwen3.5-flash + +# ===== Agentic RAG 配置 (P0-1) ===== +# 以下参数控制 /api/v1/agent/agentic/stream 多步推理管线 +# 意图分类: simple_qa / compare / multi_hop / ambiguous +# compare 和 multi_hop 触发查询分解，最多 AGENTIC_MAX_SUB_QUERIES 个子查询 +AGENTIC_MAX_SUB_QUERIES=4 +# 引文锚定 fast-path 阈值: avg_score > 此值 且 chunks >= 3 时跳过 LLM grounding check +# 降低此值可让更多查询触发 LLM 二次验证（更准确，但延迟+成本增加） +AGENTIC_GROUNDING_THRESHOLD=0.65 +# 各步骤 LLM 最大 token 数（越小越快，越大越准） +AGENTIC_INTENT_MAX_TOKENS=200 +AGENTIC_PLAN_MAX_TOKENS=400 +AGENTIC_GROUNDING_MAX_TOKENS=250 + # ===== CORS ===== # 逗号分隔的允许跨域来源列表，生产环境绝不能使用 * CORS_ALLOW_ORIGINS=http://localhost:5173 + diff --git a/.gitignore b/.gitignore index 9b157e5..c874fe5 100644 --- a/.gitignore +++ b/.gitignore @@ -61,4 +61,7 @@ Thumbs.db logs/ # codex -.agents \ No newline at end of file +.agents + +# personal local records (never commit) +local/ \ No newline at end of file diff --git a/QUICK_DEPLOY.md b/QUICK_DEPLOY.md index 3019b0f..7f7a570 100644 --- a/QUICK_DEPLOY.md +++ b/QUICK_DEPLOY.md @@ -390,12 +390,38 @@ Demo-glm/ | 下载文档 | `/api/v1/documents/download/{doc_id}` | GET | 下载原文PDF/DOCX | | 文档列表 | `/api/v1/documents/list` | GET | 列出已上传文档 | | 检索知识 | `/api/v1/knowledge/search` | POST | 向量检索 | -| 单次问答 | `/api/v1/agent/ask` | POST | 智能问答 | -| 多轮对话 | `/api/v1/agent/chat` | POST | 会话对话 | +| 单次问答 | `/api/v1/agent/ask` | POST | 标准单轮问答 | +| 多轮对话 | `/api/v1/agent/chat` | POST | 标准会话对话 | +| 流式对话 | `/api/v1/agent/chat/stream` | POST | 标准流式问答 (SSE) | +| **Agentic 流式对话** | **`/api/v1/agent/agentic/stream`** | **POST** | **P0-1 多步推理 (SSE)：意图分析→查询分解→迭代检索→引文锚定→生成** | | 会话信息 | `/api/v1/agent/session/{id}` | GET | 获取会话 | | 删除会话 | `/api/v1/agent/session/{id}` | DELETE | 删除会话 | -| Prompt模板 | `/api/v1/agent/templates` | GET | 模板列表 | -| 可用模型 | `/api/v1/agent/models` | GET | LLM模型列表 | +| 会话历史 | `/api/v1/agent/session/{id}/history` | GET | 获取历史记录 | +| 会话列表 | `/api/v1/agent/sessions` | GET | 列出所有会话 | + +### Agentic 流式接口说明 (`/api/v1/agent/agentic/stream`) + +**请求体** (同 `/agent/chat/stream`)： +```json +{ "query": "GB 18384 与 ECE R100 在电池安全上有哪些差异？", "session_id": null, "top_k": 5 } +``` + +**额外 SSE 事件** (`thinking`)： +``` +event: thinking +data: {"step": "intent_analysis", "status": "done", "intent_type": "compare", "requires_decomposition": true} + +event: thinking +data: {"step": "query_planning", "status": "done", "sub_queries": ["GB 18384 电池安全要求", "ECE R100 电池安全要求"]} + +event: thinking +data: {"step": "retrieving", "status": "done", "query": "GB 18384 电池安全要求", "index": 1, "total": 2, "found": 8} + +event: thinking +data: {"step": "grounding_check", "status": "done", "sufficient": true, "confidence": 0.82, "reason": "检索置信度充足"} +``` + +**意图类型**：`simple_qa`（单跳）/ `compare`（对比）/ `multi_hop`（多跳）/ `ambiguous`（模糊） --- diff --git a/backend/app/api/models/agent.py b/backend/app/api/models/agent.py index ac23486..53d624f 100644 --- a/backend/app/api/models/agent.py +++ b/backend/app/api/models/agent.py @@ -42,6 +42,11 @@ class ChatRequest(BaseModel): provider: Optional[str] = None model: Optional[str] = None top_k: Optional[int] = Field(default=None, ge=1, le=20) + # Optional document text uploaded by the user as conversation context. + # The text is injected directly into the LLM prompt so the model can + # answer questions about it without vector-store indexing. + context_text: Optional[str] = Field(default=None, max_length=12000) + context_filename: Optional[str] = Field(default=None, max_length=256) class ChatResponse(BaseModel): diff --git a/backend/app/api/routes/agent.py b/backend/app/api/routes/agent.py index 651aedf..e0b21c8 100644 --- a/backend/app/api/routes/agent.py +++ b/backend/app/api/routes/agent.py @@ -20,7 +20,11 @@ from app.api.models import ( ) from app.config.settings import settings from app.shared.async_utils import iter_in_thread -from app.shared.bootstrap import get_agent_conversation_service, get_agent_session_service +from app.shared.bootstrap import ( + get_agent_conversation_service, + get_agent_session_service, + get_agentic_conversation_service, +) # Keep route handlers close to their transport-layer wiring for easier auditing. @@ -182,3 +186,58 @@ async def submit_feedback(request: FeedbackRequest): return {"message": "反馈已提交", "session_id": result.session_id, "message_index": result.message_index} except ValueError as exc: raise HTTPException(status_code=404, detail=str(exc)) + + +# ── P0-1: Agentic RAG endpoint ──────────────────────────────────────────────── + +@router.post("/agentic/stream") +async def agentic_stream(request: ChatRequest): + """Stream an Agentic RAG response with live multi-step reasoning trace. + + Unlike the standard ``/chat/stream`` endpoint this route runs a full pipeline: + intent analysis → query planning → iterative retrieval → grounding check → + answer generation. + + Extra SSE event types beyond the standard ones: + + * ``thinking`` — reasoning sub-step progress; data is a JSON object with + ``step`` (intent_analysis / query_planning / retrieving / grounding_check), + ``status`` (running / done), and step-specific fields. + + The ``sources``, ``content``, and ``done`` events are identical to the standard + chat-stream contract so the existing frontend parser can handle them without + changes. + """ + async def generate_sse() -> AsyncGenerator[str, None]: + """Handle SSE generation for the agentic chat endpoint.""" + try: + session_id_, event_stream = get_agentic_conversation_service().stream_agentic_chat( + query=request.query, + session_id=request.session_id, + filters=request.filters, + provider=request.provider or settings.llm_provider, + model=request.model or settings.llm_model, + top_k=request.top_k or settings.rag_top_k, + context_text=request.context_text, + context_filename=request.context_filename, + ) + yield f"event: session\ndata: {json.dumps({'session_id': session_id_})}\n\n" + async for event_data in iter_in_thread(event_stream): + event_type = event_data.get("event", "content") + data = event_data.get("data", "") + if isinstance(data, (dict, list)): + yield f"event: {event_type}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n" + else: + yield f"event: {event_type}\ndata: {data}\n\n" + except Exception as exc: + yield f"event: error\ndata: {str(exc)}\n\n" + + return StreamingResponse( + generate_sse(), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", + }, + ) diff --git a/backend/app/api/routes/compliance.py b/backend/app/api/routes/compliance.py index d0be8c5..43d86ca 100644 --- a/backend/app/api/routes/compliance.py +++ b/backend/app/api/routes/compliance.py @@ -85,9 +85,10 @@ async def analyze_stream( Events: stage | source | finding | done | error """ from app.application.compliance.pipeline import ( + detect_cross_clause_conflicts, extract_text_from_doc_id, extract_text_from_file, - run_clauses_parallel, + run_clauses_streaming, split_into_clauses, synthesize_conclusion, ) @@ -135,23 +136,27 @@ async def analyze_stream( await asyncio.sleep(0) clauses: list[str] = await asyncio.to_thread(split_into_clauses, para_text, client) - # ── Stage 3: retrieve + gap check (parallel across all clauses) ──────────── + # ── Stage 3: progressive per-clause retrieve + gap check ────── findings: list[dict] = [] + total_clauses = len(clauses) yield _sse({ "type": "stage", "stage": "analyzing", - "label": f"Analyzing {len(clauses)} clauses in parallel…", + "label": f"Analyzing {total_clauses} clauses…", }) + # Emit initial progress so the frontend can show the total count + yield _sse({"type": "progress", "done": 0, "total": total_clauses}) await asyncio.sleep(0) - clause_results = await run_clauses_parallel( + done_count = 0 + # Stream results as each clause completes (not after all finish) + async for res in run_clauses_streaming( clauses, retrieval_service, client, top_k=5, domains=domains or None, - ) - - for res in clause_results: + ): + done_count += 1 i = res["index"] chunks = res["chunks"] finding = res["finding"] @@ -165,14 +170,25 @@ async def analyze_stream( "score": round(float(getattr(chunk, "score", 0)), 3), "status": "retrieved", "full_content": (getattr(chunk, "text", "") or "")[:300], + "clause_index": i, }) if finding: findings.append(finding) yield _sse({"type": "finding", **finding}) + # Real progress update after each clause completes + yield _sse({"type": "progress", "done": done_count, "total": total_clauses}) await asyncio.sleep(0) + # ── Stage 3b: cross-clause conflict detection ───────────────── + if findings: + conflicts = await asyncio.to_thread( + detect_cross_clause_conflicts, findings, client + ) + if conflicts: + yield _sse({"type": "conflicts", "items": conflicts}) + # ── Stage 4: synthesize conclusion ──────────────────────────── yield _sse({"type": "stage", "stage": "concluding", "label": "Generating conclusion…"}) await asyncio.sleep(0) diff --git a/backend/app/api/routes/documents.py b/backend/app/api/routes/documents.py index 91c1a22..d8caf14 100644 --- a/backend/app/api/routes/documents.py +++ b/backend/app/api/routes/documents.py @@ -241,6 +241,9 @@ async def get_document_management_list(): "updated_at": item.updated_at.isoformat(), "regulation_type": item.regulation_type, "version": item.version, + # True only when the original binary file is stored in MinIO. + # Milvus-only synthetic docs have no binary file — download is disabled. + "has_file": bool(item.object_name), } for item in documents ], diff --git a/backend/app/api/routes/rag.py b/backend/app/api/routes/rag.py index bf37d40..2ad9d5f 100644 --- a/backend/app/api/routes/rag.py +++ b/backend/app/api/routes/rag.py @@ -3,10 +3,14 @@ from __future__ import annotations import json -from typing import AsyncGenerator +import os +import re +import tempfile +from typing import AsyncGenerator, Optional -from fastapi import APIRouter, Depends +from fastapi import APIRouter, Depends, File, UploadFile from fastapi.responses import StreamingResponse +from loguru import logger from app.api.dependencies.auth import get_current_user from app.config.settings import settings @@ -15,6 +19,8 @@ from app.schemas.rag import RagChatRequest, QuickQuestionsResponse, QuickQuestio from app.shared.async_utils import iter_in_thread from app.shared.bootstrap import get_agent_conversation_service +# Maximum characters of document text injected as LLM context (≈ 6 000 tokens). +_MAX_CONTEXT_CHARS = 8_000 router = APIRouter(prefix="/rag", tags=["RAG问答"]) @@ -28,17 +34,90 @@ _DEFAULT_QUICK_QUESTIONS = [ ] +def _extract_text_from_bytes(content: bytes, filename: str) -> str: + """Extract plain text from an uploaded file using the document parser. + + Tries the configured parser first; falls back to raw UTF-8 decode for + plain-text formats (.txt, .md). Returns at most _MAX_CONTEXT_CHARS characters + so the text fits comfortably inside the LLM context window. + """ + suffix = os.path.splitext(filename or "doc.pdf")[1] or ".pdf" + # Fast path: plain-text files don't need a parser + if suffix.lower() in {".txt", ".md", ".csv"}: + try: + return content.decode("utf-8", errors="replace")[:_MAX_CONTEXT_CHARS] + except Exception: + pass + + tmp_path = "" + try: + with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp: + tmp.write(content) + tmp_path = tmp.name + from app.shared.bootstrap import get_document_command_service + svc = get_document_command_service() + parsed = svc.parser.parse(file_path=tmp_path, doc_id="ctx_extract", doc_name=filename) + if parsed.raw_text: + return parsed.raw_text[:_MAX_CONTEXT_CHARS] + # Fallback: join semantic blocks + return "\n".join( + b.get("text", "") for b in parsed.semantic_blocks if b.get("text") + )[:_MAX_CONTEXT_CHARS] + except Exception as exc: + logger.warning("Context text extraction failed for {}: {}", filename, exc) + return "" + finally: + if tmp_path: + try: + os.unlink(tmp_path) + except OSError: + pass + + +@router.post("/upload-context") +async def upload_context( + file: UploadFile = File(...), + current_user: UserClaims = Depends(get_current_user), +): + """Extract text from an uploaded document and return it as conversation context. + + The client stores the returned text and includes it in subsequent /rag/chat + requests via the context_text field — the LLM receives the document content + directly without requiring vector-store indexing. + """ + content = await file.read() + filename = file.filename or "document" + text = await __import__("asyncio").to_thread(_extract_text_from_bytes, content, filename) + if not text.strip(): + from fastapi import HTTPException + raise HTTPException(status_code=422, detail="Could not extract text from the uploaded file.") + return { + "filename": filename, + "text": text, + "char_count": len(text), + "truncated": len(text) >= _MAX_CONTEXT_CHARS, + } + + @router.post("/chat") async def rag_chat( request: RagChatRequest, current_user: UserClaims = Depends(get_current_user), ): - """Stream RAG Q&A using the real agent service.""" + """Stream RAG Q&A using the real agent service. + + When request.context_text is provided the document text is passed directly + to the answer generator as a dedicated document context section — RAG + retrieval still runs on the user's original question (not the document text) + so embedding quality is preserved for regulation chunk matching. + """ session_id, event_stream = get_agent_conversation_service().stream_chat( query=request.query, session_id=request.session_id, filters=request.filters, top_k=request.top_k or settings.rag_top_k, + context_text=request.context_text, + context_filename=request.context_filename, ) async def generate() -> AsyncGenerator[str, None]: diff --git a/backend/app/application/agent/__init__.py b/backend/app/application/agent/__init__.py index 7aaf8b6..9253766 100644 --- a/backend/app/application/agent/__init__.py +++ b/backend/app/application/agent/__init__.py @@ -1,7 +1,13 @@ """Initialize the app.application.agent package.""" from .services import AgentConversationService, AgentSessionFeedbackResult, AgentSessionService +from .agentic_service import AgenticConversationService # Keep package boundaries explicit so backend imports stay predictable. -__all__ = ["AgentConversationService", "AgentSessionFeedbackResult", "AgentSessionService"] +__all__ = [ + "AgentConversationService", + "AgentSessionFeedbackResult", + "AgentSessionService", + "AgenticConversationService", +] diff --git a/backend/app/application/agent/agentic_service.py b/backend/app/application/agent/agentic_service.py new file mode 100644 index 0000000..59e31f4 --- /dev/null +++ b/backend/app/application/agent/agentic_service.py @@ -0,0 +1,453 @@ +"""Implement the Agentic RAG pipeline for multi-step reasoning (P0-1). + +Architecture +------------ +The pipeline adds four explicit reasoning steps before answer generation: + + 1. Intent Analysis — classify query type (simple_qa / compare / multi_hop / ambiguous) + 2. Query Planning — for complex intents, decompose into focused sub-queries + 3. Iterative Retrieval — retrieve for each sub-query, merge with deduplication + 4. Grounding Check — verify retrieved context is sufficient; refine query when not + 5. Answer Generation — stream final answer with citations (reuses AnswerGenerator) + +Each step emits SSE ``thinking`` events so the frontend can render the live +reasoning trace. The pipeline is entirely synchronous and returns a generator so +it plugs into the same ``iter_in_thread`` pattern used by the existing chat routes. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from typing import Generator + +from loguru import logger + +from app.application.knowledge import KnowledgeRetrievalService +from app.application.agent.hyde_expander import HyDEExpander +from app.config.settings import settings +from app.domain.conversation import ConversationStore +from app.domain.retrieval import RetrievedChunk +from app.infrastructure.llm.openai_compatible_answer_generator import OpenAICompatibleAnswerGenerator +from app.services.llm.llm_factory import get_llm_client + +# ── Prompts ─────────────────────────────────────────────────────────────────── +# Each prompt is kept module-level for easy review and fine-tuning. + +_INTENT_SYSTEM = ( + "You are a query classifier for a Chinese regulatory compliance knowledge base.\n\n" + "Classify the query into exactly one of:\n" + '- "simple_qa" : Single-hop, factual question about one regulation or clause\n' + '- "compare" : Comparison between two or more regulations, standards, or versions\n' + '- "multi_hop" : Requires chaining facts across multiple regulations to answer\n' + '- "ambiguous" : Too vague or broad to retrieve effectively\n\n' + "Return ONLY valid JSON — no markdown, no extra text:\n" + '{"type": "...", "reason": "one sentence", "requires_decomposition": true/false}\n\n' + '"requires_decomposition" must be true for compare and multi_hop types.' +) + +_PLAN_SYSTEM = ( + "You are a query planner for a Chinese regulatory compliance knowledge base.\n\n" + "Decompose the query into 2-4 focused, self-contained sub-queries that together fully " + "address the original question. Each sub-query must target one specific regulation, " + "clause, or concept and be independently searchable.\n\n" + "Return ONLY a valid JSON array — no markdown, no extra text:\n" + '["sub-query 1", "sub-query 2", ...]' +) + +_GROUNDING_SYSTEM = ( + "You are a grounding verifier for a regulatory compliance QA system.\n\n" + "Given a query and retrieved regulation passages, decide whether the passages contain " + "sufficient, accurate information to answer the query.\n\n" + "Return ONLY valid JSON — no markdown, no extra text:\n" + '{"sufficient": true/false, "confidence": 0.0-1.0, "reason": "one sentence", ' + '"refined_query": "a more specific search query if not sufficient, else null"}' +) + + +# ── Result dataclasses ──────────────────────────────────────────────────────── + +@dataclass +class IntentResult: + """Capture the output of the intent-analysis step.""" + + type: str = "simple_qa" + reason: str = "" + requires_decomposition: bool = False + + +@dataclass +class GroundingResult: + """Capture the output of the grounding-check step.""" + + sufficient: bool = True + confidence: float = 1.0 + reason: str = "" + refined_query: str | None = None + + +# ── Service ─────────────────────────────────────────────────────────────────── + +class AgenticConversationService: + """Multi-step Agentic RAG pipeline with live reasoning trace via SSE. + + The service is intentionally synchronous so it can be wrapped in + ``iter_in_thread`` by the route layer without any async boilerplate. + """ + + def __init__( + self, + *, + retrieval_service: KnowledgeRetrievalService, + answer_generator: OpenAICompatibleAnswerGenerator, + conversation_store: ConversationStore, + ) -> None: + """Initialise with injected dependencies from the composition root.""" + self.retrieval_service = retrieval_service + self.answer_generator = answer_generator + self.conversation_store = conversation_store + # HyDE expander is stateless — one instance shared for all requests. + self._hyde = HyDEExpander() + + # ── Private helpers ─────────────────────────────────────────────────────── + + def _llm_json( + self, + system: str, + user: str, + provider: str | None, + model: str | None, + max_tokens: int = 300, + ) -> dict | list | None: + """Call the LLM with a JSON-only prompt and return the parsed result. + + Returns ``None`` on any API or parse failure so callers can degrade + gracefully without raising. + """ + client = get_llm_client( + provider=provider or settings.llm_provider, + model=model or settings.llm_model, + ) + resp = client.chat( + [{"role": "system", "content": system}, {"role": "user", "content": user}], + max_tokens=max_tokens, + temperature=0.1, + ) + if not resp.is_success: + logger.warning("AgenticService LLM call failed: {}", resp.error) + return None + try: + raw = resp.content.strip() + # Strip accidental markdown code fences the model may add. + if raw.startswith("```"): + parts = raw.split("```") + raw = parts[1] if len(parts) > 1 else raw + if raw.startswith("json"): + raw = raw[4:] + return json.loads(raw.strip()) + except (json.JSONDecodeError, IndexError) as exc: + logger.debug("AgenticService JSON parse failed: {} | raw={}", exc, resp.content[:200]) + return None + + def _analyze_intent( + self, query: str, provider: str | None, model: str | None + ) -> IntentResult: + """Classify query intent to select the appropriate retrieval strategy.""" + data = self._llm_json( + _INTENT_SYSTEM, + f"Query: {query}", + provider, + model, + max_tokens=settings.agentic_intent_max_tokens, + ) + if isinstance(data, dict): + return IntentResult( + type=str(data.get("type", "simple_qa")), + reason=str(data.get("reason", "")), + requires_decomposition=bool(data.get("requires_decomposition", False)), + ) + return IntentResult(type="simple_qa", reason="fallback — classifier returned no JSON", requires_decomposition=False) + + def _plan_queries( + self, query: str, intent_type: str, provider: str | None, model: str | None + ) -> list[str]: + """Decompose a complex query into focused, independently-retrievable sub-queries.""" + data = self._llm_json( + _PLAN_SYSTEM, + f"Original query ({intent_type}): {query}", + provider, + model, + max_tokens=settings.agentic_plan_max_tokens, + ) + if isinstance(data, list) and data: + # Cap at configured maximum to keep latency predictable. + return [str(q) for q in data[:settings.agentic_max_sub_queries] if q] + return [query] + + def _check_grounding( + self, + query: str, + chunks: list[RetrievedChunk], + provider: str | None, + model: str | None, + ) -> GroundingResult: + """Verify whether retrieved chunks are sufficient to ground an accurate answer. + + Uses a fast score-threshold heuristic first; falls back to an LLM call only + when scores are borderline so that the happy-path adds no extra latency. + """ + if not chunks: + return GroundingResult( + sufficient=False, + confidence=0.0, + reason="未检索到相关内容", + refined_query=None, + ) + + avg_score = sum(c.score for c in chunks) / len(chunks) + # Fast path: high-confidence retrieval → skip extra LLM call. + if avg_score > settings.agentic_grounding_threshold and len(chunks) >= 3: + return GroundingResult( + sufficient=True, + confidence=round(avg_score, 3), + reason="检索置信度充足，无需二次查询", + refined_query=None, + ) + + # LLM-based grounding check for borderline retrievals. + context_preview = "\n".join( + f"[{i + 1}] (score={c.score:.2f}) {c.text[:200]}" for i, c in enumerate(chunks[:5]) + ) + data = self._llm_json( + _GROUNDING_SYSTEM, + f"Query: {query}\n\nRetrieved passages:\n{context_preview}", + provider, + model, + max_tokens=settings.agentic_grounding_max_tokens, + ) + if isinstance(data, dict): + return GroundingResult( + sufficient=bool(data.get("sufficient", True)), + confidence=float(data.get("confidence", 0.5)), + reason=str(data.get("reason", "")), + refined_query=data.get("refined_query") or None, + ) + return GroundingResult(sufficient=True, confidence=0.5, reason="grounding check skipped (parse error)", refined_query=None) + + @staticmethod + def _intent_to_template(intent_type: str) -> str: + """Map an intent type to the best prompt template name for answer generation.""" + mapping = { + "compare": "comparison", + "multi_hop": "compliance_qa", + "simple_qa": "compliance_qa", + "ambiguous": "compliance_qa", + } + return mapping.get(intent_type, "compliance_qa") + + @staticmethod + def _deduplicate(chunks: list[RetrievedChunk], max_chunks: int) -> list[RetrievedChunk]: + """Remove duplicate chunk IDs, preserving first-occurrence order up to max_chunks.""" + seen: set[str] = set() + result: list[RetrievedChunk] = [] + for chunk in chunks: + if chunk.chunk_id not in seen: + seen.add(chunk.chunk_id) + result.append(chunk) + if len(result) >= max_chunks: + break + return result + + # ── Public interface ────────────────────────────────────────────────────── + + def stream_agentic_chat( + self, + *, + query: str, + session_id: str | None = None, + filters: str | None = None, + provider: str | None = None, + model: str | None = None, + top_k: int = 5, + context_text: str | None = None, + context_filename: str | None = None, + ) -> tuple[str, Generator[dict, None, None]]: + """Run the full Agentic RAG pipeline and return ``(session_id, event_generator)``. + + When context_text is provided (user-attached document) it is: + - Summarised and prepended to the intent-analysis prompt so the classifier + understands what kind of question is being asked. + - Treated as baseline grounding so the pipeline skips unnecessary retries + when the document itself is the primary source. + - Passed to the answer generator so the LLM sees the full document alongside + retrieved regulation chunks. + + The generator yields SSE event dicts compatible with the route's + ``iter_in_thread`` pattern. + """ + session = self.conversation_store.get_session(session_id) if session_id else None + if session is None: + session = self.conversation_store.create_session() + self.conversation_store.save_message(session.session_id, role="user", content=query) + history = [{"role": msg.role, "content": msg.content} for msg in session.messages[-10:]] + active_session_id = session.session_id + + # Build a brief document summary for classifier/planner prompts (avoid + # passing the full text which could overwhelm small-context LLMs). + _doc_summary: str = "" + if context_text and context_text.strip(): + _doc_label = context_filename or "document" + _preview = context_text.strip()[:400] + _doc_summary = f"[User has attached document: {_doc_label}]\nDocument preview: {_preview}…\n\n" + + def event_stream() -> Generator[dict, None, None]: + """Execute all pipeline steps and yield SSE events.""" + # ── Step 1: Intent Analysis ────────────────────────────────────── + yield {"event": "thinking", "data": {"step": "intent_analysis", "status": "running"}} + # Prepend doc summary so the classifier knows what the user is asking about + intent_user_msg = f"{_doc_summary}Query: {query}" if _doc_summary else f"Query: {query}" + data = self._llm_json( + _INTENT_SYSTEM, intent_user_msg, provider, model, + max_tokens=settings.agentic_intent_max_tokens, + ) + if isinstance(data, dict): + intent = IntentResult( + type=str(data.get("type", "simple_qa")), + reason=str(data.get("reason", "")), + requires_decomposition=bool(data.get("requires_decomposition", False)), + ) + else: + intent = IntentResult(type="simple_qa", reason="fallback", requires_decomposition=False) + logger.debug("Agentic intent: type={} decompose={}", intent.type, intent.requires_decomposition) + yield { + "event": "thinking", + "data": { + "step": "intent_analysis", + "status": "done", + "intent_type": intent.type, + "reason": intent.reason, + "requires_decomposition": intent.requires_decomposition, + }, + } + + # ── Step 2: Query Planning ─────────────────────────────────────── + sub_queries: list[str] = [query] + if intent.requires_decomposition: + yield {"event": "thinking", "data": {"step": "query_planning", "status": "running"}} + plan_user_msg = f"{_doc_summary}Original query ({intent.type}): {query}" if _doc_summary else f"Original query ({intent.type}): {query}" + data_plan = self._llm_json( + _PLAN_SYSTEM, plan_user_msg, provider, model, + max_tokens=settings.agentic_plan_max_tokens, + ) + if isinstance(data_plan, list) and data_plan: + sub_queries = [str(q) for q in data_plan[:settings.agentic_max_sub_queries] if q] + logger.debug("Agentic sub-queries ({}): {}", len(sub_queries), sub_queries) + yield { + "event": "thinking", + "data": {"step": "query_planning", "status": "done", "sub_queries": sub_queries}, + } + + # ── Step 3: Iterative Retrieval ────────────────────────────────── + # Always retrieve using the user's original question (NOT the document + # text) so embedding quality is preserved for regulation matching. + # HyDE enriches the retrieval query with a short hypothetical answer + # to close the vocabulary gap between terse queries and long documents. + candidate_k = max(top_k * 3, 15) + all_chunks: list[RetrievedChunk] = [] + + # For simple_qa with a single query, HyDE gives the biggest benefit + # (bridging vague/colloquial questions to formal document language). + # For compare/multi_hop, the planner already decomposed into precise + # sub-queries, so HyDE is less critical but still applied per sub-query. + for idx, sq in enumerate(sub_queries, start=1): + yield { + "event": "thinking", + "data": {"step": "retrieving", "status": "running", "query": sq, "index": idx, "total": len(sub_queries)}, + } + # HyDE expansion: generate hypothetical answer, embed it for retrieval. + # Falls back to original sub-query if LLM call fails. + retrieval_query = self._hyde.expand(sq) + chunks = self.retrieval_service.retrieve(query=retrieval_query, top_k=candidate_k, filters=filters) + all_chunks.extend(chunks) + yield { + "event": "thinking", + "data": {"step": "retrieving", "status": "done", "query": sq, "index": idx, "total": len(sub_queries), "found": len(chunks)}, + } + + unique_chunks = self._deduplicate(all_chunks, max_chunks=top_k * 4) + + # ── Step 4: Grounding Check ────────────────────────────────────── + yield {"event": "thinking", "data": {"step": "grounding_check", "status": "running"}} + + # When the user has attached a document, the document itself provides + # baseline grounding — skip the re-query loop to avoid the LLM asking + # "please provide the document text" as a refined query. + if context_text and context_text.strip(): + grounding = GroundingResult( + sufficient=True, + confidence=0.95, + reason="用户已附件上传文档，以文档内容为基础作答", + refined_query=None, + ) + else: + grounding = self._check_grounding(query, unique_chunks, provider, model) + + yield { + "event": "thinking", + "data": { + "step": "grounding_check", + "status": "done", + "sufficient": grounding.sufficient, + "confidence": grounding.confidence, + "reason": grounding.reason, + }, + } + + # Only retry from vector store when no document is attached and grounding failed + if not grounding.sufficient and grounding.refined_query and not context_text: + logger.info("Grounding insufficient — re-querying: {}", grounding.refined_query) + yield { + "event": "thinking", + "data": {"step": "retrieving", "status": "running", "query": grounding.refined_query, "index": 1, "total": 1, "retry": True}, + } + # Apply HyDE to the refined query as well for better retrieval. + refined_hyde_query = self._hyde.expand(grounding.refined_query) + refined_chunks = self.retrieval_service.retrieve(query=refined_hyde_query, top_k=candidate_k, filters=filters) + all_chunks.extend(refined_chunks) + unique_chunks = self._deduplicate(all_chunks, max_chunks=top_k * 4) + yield { + "event": "thinking", + "data": {"step": "retrieving", "status": "done", "query": grounding.refined_query, "index": 1, "total": 1, "found": len(refined_chunks), "retry": True}, + } + + final_chunks = unique_chunks[:top_k] + + # ── Step 5: Answer Generation ──────────────────────────────────── + sources_payload = [s.__dict__ for s in self.answer_generator._sources(final_chunks)] + yield {"event": "sources", "data": sources_payload} + + answer_parts: list[str] = [] + for event in self.answer_generator.stream_generate( + query=query, + retrieved_chunks=final_chunks, + history=history, + provider=provider, + model=model, + prompt_template=self._intent_to_template(intent.type), + context_text=context_text, + context_filename=context_filename, + ): + if event.get("event") == "content": + answer_parts.append(str(event.get("data", ""))) + yield event + + full_answer = "".join(answer_parts) + self.conversation_store.save_message( + active_session_id, + role="assistant", + content=full_answer, + sources=sources_payload, + ) + + return active_session_id, event_stream() + diff --git a/backend/app/application/agent/hyde_expander.py b/backend/app/application/agent/hyde_expander.py new file mode 100644 index 0000000..8b19e28 --- /dev/null +++ b/backend/app/application/agent/hyde_expander.py @@ -0,0 +1,105 @@ +"""Implement HyDE (Hypothetical Document Embeddings) query expansion. + +HyDE improves dense retrieval by addressing the vocabulary gap between +short user queries and longer document passages: + + User query → [LLM generates hypothetical answer] + ↓ + embed hypothetical answer (not original query) + ↓ + retrieve similar real passages from Milvus + +The hypothetical answer uses the same vocabulary and phrasing as documents, +so its embedding is much closer to relevant chunks than a terse query embedding. + +Usage: + expander = HyDEExpander() + retrieval_query = expander.expand(query, provider=..., model=...) + chunks = retrieval_service.retrieve(query=retrieval_query, ...) + +When the LLM call fails, expand() falls back to the original query so the +retrieval pipeline degrades gracefully. + +References: + Gao et al. (2022), "Precise Zero-Shot Dense Retrieval without Relevance Labels" + https://arxiv.org/abs/2212.10496 +""" + +from __future__ import annotations + +from loguru import logger + +from app.config.settings import settings +from app.services.llm.llm_factory import get_llm_client + +# Maximum chars to trim from the hypothetical answer to avoid token overrun. +_MAX_HYPOTHESIS_CHARS = 600 + +# System prompt that instructs the LLM to write a passage *as if* it were +# from a regulatory document, not a conversation answer. +_HYDE_SYSTEM = ( + "你是一位法规知识库专家。用户提出了一个问题，" + "请用50-120字写一段话，模拟如果相关法规文档中存在完美答案，" + "该段落会是什么内容。\n\n" + "要求：\n" + "- 使用与法规文档相同的正式书面语气\n" + "- 包含可能的条款编号、标准名称等关键术语\n" + "- 不要解释你在做什么，直接输出假设性段落\n" + "- 如问题过于模糊，写一段合理的通用法规说明" +) + + +class HyDEExpander: + """Generate a hypothetical document passage to improve dense retrieval. + + The expander is stateless — instantiate once and call expand() per query. + It requires no external dependencies beyond the project's existing LLM + client infrastructure. + """ + + def expand(self, query: str) -> str: + """Return a combined retrieval query: original query + hypothetical passage. + + The combination ensures: + - Dense retrieval uses the enriched hypothetical text (semantic match). + - BM25 retrieval still benefits from the original query keywords. + + The model used is ``settings.hyde_llm_model`` (dedicated lightweight model) + falling back to the main ``settings.llm_model`` when not configured. + + If the LLM call fails for any reason, returns the original query unchanged. + """ + if not settings.hyde_enabled: + return query + + # Use the dedicated HyDE model when configured; fall back to main LLM. + # A lightweight model (e.g. qwen3.5-flash) is sufficient for generating + # a short hypothetical passage and significantly reduces cost + latency. + provider = settings.hyde_llm_provider or settings.llm_provider + model = settings.hyde_llm_model or settings.llm_model + + try: + client = get_llm_client(provider=provider, model=model) + resp = client.chat( + messages=[ + {"role": "system", "content": _HYDE_SYSTEM}, + {"role": "user", "content": f"问题：{query}"}, + ], + max_tokens=settings.hyde_max_tokens, + # Low temperature: we want a plausible, deterministic passage. + temperature=0.3, + ) + if not resp.is_success or not resp.content: + logger.debug("HyDE LLM call failed or empty — using original query") + return query + + hypothesis = resp.content.strip()[:_MAX_HYPOTHESIS_CHARS] + logger.debug("HyDE expanded query ({}→{} chars)", len(query), len(hypothesis)) + + # Concatenate: the embedding model will see the full combined text, + # so the resulting vector leans toward the hypothetical document style. + return f"{query}\n\n{hypothesis}" + + except Exception as exc: # noqa: BLE001 — intentional broad catch for graceful fallback + logger.warning("HyDE expansion failed: {} — using original query", exc) + return query diff --git a/backend/app/application/agent/services.py b/backend/app/application/agent/services.py index d8d4d46..3034a9b 100644 --- a/backend/app/application/agent/services.py +++ b/backend/app/application/agent/services.py @@ -9,6 +9,7 @@ from app.domain.conversation import AnswerGenerator, AnswerResult, ConversationS from app.domain.retrieval import RetrievedChunk from app.application.knowledge import KnowledgeRetrievalService +from app.application.agent.hyde_expander import HyDEExpander # Keep orchestration logic centralized so use-case flow stays easy to trace. @@ -26,6 +27,8 @@ class AgentConversationService: self.retrieval_service = retrieval_service self.answer_generator = answer_generator self.conversation_store = conversation_store + # Shared HyDE expander — stateless, safe for reuse across requests. + self._hyde = HyDEExpander() def ask( self, @@ -108,14 +111,26 @@ class AgentConversationService: model: str | None = None, top_k: int = 5, prompt_template: str | None = None, + context_text: str | None = None, + context_filename: str | None = None, ) -> tuple[str, Generator[dict, None, None]]: - """Stream chat for the Agent Conversation Service instance.""" + """Stream chat for the Agent Conversation Service instance. + + When context_text is provided the user's document is passed directly to + the answer generator — RAG retrieval still runs on the user's question + (not the document text) to find relevant regulation passages. + """ session = self.conversation_store.get_session(session_id) if session_id else None if session is None: session = self.conversation_store.create_session() self.conversation_store.save_message(session.session_id, role="user", content=query) history = [{"role": msg.role, "content": msg.content} for msg in session.messages[-10:]] - retrieved = self.retrieval_service.retrieve(query=query, top_k=top_k, filters=filters) + # HyDE: expand the query with a hypothetical answer to improve dense retrieval. + # For document-context queries, skip HyDE since the document itself guides retrieval. + retrieval_query = self._hyde.expand(query) if not context_text else query + # Retrieve using the enriched query — NOT the document text — + # so embedding quality is preserved for regulation chunk matching. + retrieved = self.retrieval_service.retrieve(query=retrieval_query, top_k=top_k, filters=filters) def event_stream() -> Generator[dict, None, None]: """Handle event stream for the Agent Conversation Service instance.""" @@ -129,6 +144,8 @@ class AgentConversationService: provider=provider, model=model, prompt_template=prompt_template, + context_text=context_text, + context_filename=context_filename, ): if event.get("event") == "sources": sources_payload = event.get("data", []) @@ -189,3 +206,4 @@ class AgentSessionService: raise ValueError("消息索引不存在") # Preserve the existing API behavior until a persistent feedback store is introduced. return AgentSessionFeedbackResult(session_id=session_id, message_index=message_index) + diff --git a/backend/app/application/compliance/pipeline.py b/backend/app/application/compliance/pipeline.py index de348a4..563eef3 100644 --- a/backend/app/application/compliance/pipeline.py +++ b/backend/app/application/compliance/pipeline.py @@ -51,19 +51,36 @@ def _extract_json(text: str): def extract_text_from_doc_id(doc_id: str) -> str: + """Fetch the full text of a document by retrieving its chunks filtered by doc_id. + + Uses a high top_k and doc_id filter to reconstruct the document in chunk order, + avoiding the previous approach of semantic search by doc_name which could return + chunks from unrelated documents. + """ from app.shared.bootstrap import get_document_query_service, get_retrieval_service doc = get_document_query_service().get(doc_id) if not doc: raise ValueError(f"Document '{doc_id}' not found") service = get_retrieval_service() - chunks = service.retrieve(query=doc.doc_name, top_k=30) - doc_chunks = [c for c in chunks if c.doc_id == doc_id] + # Use doc_name as a broad query, filter strictly by doc_id so we only get + # this document's chunks; top_k=100 covers most real-world documents. + chunks = service.retrieve(query=doc.doc_name, top_k=100, filters=doc_id) + doc_chunks = [c for c in chunks if getattr(c, "doc_id", None) == doc_id] if not doc_chunks: - doc_chunks = chunks[:15] - return "\n\n".join(c.text for c in doc_chunks[:15]) + # Fallback: use top results even without doc_id match (e.g., legacy store) + doc_chunks = chunks[:30] + # Sort by chunk_index to preserve document reading order + doc_chunks.sort(key=lambda c: getattr(c, "chunk_index", 0)) + return "\n\n".join(c.text for c in doc_chunks[:40]) def extract_text_from_file(content: bytes, filename: str) -> str: + """Parse an uploaded file and return its full text content. + + Removed previous 4000-char cap so large specifications and standards are + fully analysed. The caller is responsible for splitting the text into + clause-sized chunks before passing to the LLM. + """ from app.shared.bootstrap import get_document_command_service suffix = os.path.splitext(filename or "doc.pdf")[1] or ".pdf" tmp_path = "" @@ -74,10 +91,11 @@ def extract_text_from_file(content: bytes, filename: str) -> str: service = get_document_command_service() parsed = service.parser.parse(file_path=tmp_path, doc_id="tmp_analysis", doc_name=filename) if parsed.raw_text: - return parsed.raw_text[:4000] + # Return full text — truncation happens in split_into_clauses() + return parsed.raw_text return "\n".join( - b.get("text", "") for b in parsed.semantic_blocks[:30] if b.get("text") - )[:4000] + b.get("text", "") for b in parsed.semantic_blocks if b.get("text") + ) except Exception as exc: logger.warning("File text extraction failed: {}", exc) return "" @@ -88,27 +106,68 @@ def extract_text_from_file(content: bytes, filename: str) -> str: def split_into_clauses(text: str, client: "BaseLLMClient") -> list[str]: - prompt = ( - "You are a compliance analysis expert. Split the following text into 3-8 " - "semantically complete compliance clauses. Each clause should be an independent " - "compliance requirement or technical statement.\n" - "Return as JSON array of strings, e.g.:\n" - '["Clause one...", "Clause two..."]\n' - "Return ONLY the JSON array.\n\n" - f"Text:\n{text[:2000]}" - ) - response = client.chat([{"role": "user", "content": prompt}], max_tokens=1000) - if response.is_success: - try: - result = _extract_json(response.content) - if isinstance(result, list): - clauses = [str(c).strip() for c in result if str(c).strip()] - if clauses: - return clauses[:8] - except (ValueError, TypeError): - logger.warning("Clause split JSON parse failed, using fallback") - sentences = re.split(r"[.?!;\n]+", text) - return [s.strip() for s in sentences if len(s.strip()) > 20][:6] + """Split a compliance document into semantically independent clauses. + + For long texts (> 2 000 chars) the document is processed in overlapping + 2 000-char windows so no content is missed. Each window produces up to 4 + clauses; results are deduplicated and capped at 12 total to keep analysis + latency reasonable. + """ + # Window size and step for sliding-window clause extraction + _WINDOW = 2000 + _STEP = 1800 # 200-char overlap to avoid cutting clauses at boundaries + _MAX_CLAUSES = 12 + + windows = [] + if len(text) <= _WINDOW: + windows = [text] + else: + pos = 0 + while pos < len(text): + windows.append(text[pos: pos + _WINDOW]) + pos += _STEP + + all_clauses: list[str] = [] + for window in windows: + prompt = ( + "You are a compliance analysis expert. Split the following text into " + "3-4 semantically complete compliance clauses. Each clause must be an " + "independent requirement or technical statement. Omit section headings, " + "definitions, and non-normative text.\n" + "Return as JSON array of strings, e.g.:\n" + '["Clause one...", "Clause two..."]\n' + "Return ONLY the JSON array.\n\n" + f"Text:\n{window}" + ) + response = client.chat([{"role": "user", "content": prompt}], max_tokens=800) + if response.is_success: + try: + result = _extract_json(response.content) + if isinstance(result, list): + clauses = [str(c).strip() for c in result if str(c).strip()] + all_clauses.extend(clauses[:4]) + except (ValueError, TypeError): + logger.warning("Clause split JSON parse failed for window, using sentence fallback") + sentences = re.split(r"[.?!;\n]+", window) + all_clauses.extend(s.strip() for s in sentences if len(s.strip()) > 20) + else: + # LLM unavailable — fall back to sentence splitting for this window + sentences = re.split(r"[.?!;\n]+", window) + all_clauses.extend(s.strip() for s in sentences if len(s.strip()) > 20) + + if len(all_clauses) >= _MAX_CLAUSES: + break + + # Deduplicate near-duplicates (same first 80 chars) that span window boundaries + seen: set[str] = set() + deduped: list[str] = [] + for c in all_clauses: + key = c[:80].lower() + if key not in seen: + seen.add(key) + deduped.append(c) + + return deduped[:_MAX_CLAUSES] def retrieve_for_clause( @@ -117,7 +176,33 @@ def retrieve_for_clause( top_k: int = 5, domains: str | None = None, ) -> list["RetrievedChunk"]: - return retrieval_service.retrieve(query=clause, top_k=top_k, filters=domains) + """Retrieve regulation chunks relevant to a clause. + + If the best retrieval score is below 0.55, rewrite the clause into a more + technical query and retry once to improve coverage. + """ + chunks = retrieval_service.retrieve(query=clause, top_k=top_k, filters=domains) + if not chunks: + return chunks + + best_score = max((getattr(c, "score", 0) for c in chunks), default=0) + if best_score < 0.55: + # Rewrite clause as technical keyword query and retry + keywords = " ".join( + w for w in re.split(r"\W+", clause) if len(w) > 3 + )[:200] + retry_chunks = retrieval_service.retrieve(query=keywords, top_k=top_k, filters=domains) + if retry_chunks: + # Merge: keep unique chunks, prefer higher-score version + seen_ids: set[str] = {getattr(c, "chunk_id", str(i)) for i, c in enumerate(chunks)} + for rc in retry_chunks: + rid = getattr(rc, "chunk_id", "") + if rid not in seen_ids: + chunks.append(rc) + seen_ids.add(rid) + chunks.sort(key=lambda c: getattr(c, "score", 0), reverse=True) + chunks = chunks[:top_k] + return chunks def process_single_clause( @@ -130,14 +215,75 @@ def process_single_clause( ) -> dict: """Process one clause: retrieve relevant regulations then check compliance. - Returns a dict with keys: index, chunks, finding (may be None on LLM failure). + Returns a dict with keys: + - index: clause position (for ordering) + - chunks: list of RetrievedChunk (for source events) + - finding: dict with title/desc/status/clause_ref/confidence (may be None on LLM failure) + Designed to run inside asyncio.to_thread() for parallel execution. + The finding now includes a 'source_refs' list linking back to the chunks + that informed the verdict, enabling the frontend to correlate sources with findings. """ chunks = retrieve_for_clause(clause, retrieval_service, top_k, domains) finding = check_clause_compliance(clause, chunks, client) + if finding is not None: + # Attach source references so the frontend can link finding ↔ sources + finding["source_refs"] = [ + { + "standard": getattr(c, "doc_title", "") or getattr(c, "doc_name", ""), + "clause": getattr(c, "section_title", "") or "", + "score": round(float(getattr(c, "score", 0)), 3), + } + for c in chunks[:3] + ] return {"index": index, "chunks": chunks, "finding": finding} +async def run_clauses_streaming( + clauses: list[str], + retrieval_service: "KnowledgeRetrievalService", + client: "BaseLLMClient", + top_k: int = 5, + domains: str | None = None, +): + """Process all clauses concurrently and yield each result as it completes. + + Unlike the old gather()-based approach, this uses asyncio.Queue so that + findings are emitted to the SSE stream immediately when each clause + finishes — the user sees results progressively rather than waiting for + the slowest clause before seeing any output. + + Yields dicts with keys: index, chunks, finding (same schema as + process_single_clause, plus a sentinel {"_done": True} at the end). + """ + queue: asyncio.Queue[dict] = asyncio.Queue() + total = len(clauses) + + async def _worker(clause: str, i: int) -> None: + """Run one clause in a thread and push the result into the queue.""" + try: + result = await asyncio.to_thread( + process_single_clause, + clause, i, retrieval_service, client, top_k, domains, + ) + except Exception as exc: + logger.warning("Clause {} processing failed: {}", i, exc) + result = {"index": i, "chunks": [], "finding": None} + await queue.put(result) + + # Launch all workers concurrently + tasks = [asyncio.create_task(_worker(clause, i)) for i, clause in enumerate(clauses)] + + received = 0 + while received < total: + result = await queue.get() + yield result + received += 1 + + # Wait for all tasks to complete (they should already be done by now) + await asyncio.gather(*tasks, return_exceptions=True) + + async def run_clauses_parallel( clauses: list[str], retrieval_service: "KnowledgeRetrievalService", @@ -145,31 +291,15 @@ async def run_clauses_parallel( top_k: int = 5, domains: str | None = None, ) -> list[dict]: - """Run all clauses through retrieve+gap-check in parallel. + """Legacy batch API kept for backward compatibility. - Results are returned in the original clause order even though processing - is concurrent. Exceptions in individual clauses are caught and returned as - dicts with finding=None so the stream continues for remaining clauses. - - Both retrieval_service and client must be thread-safe — they are shared - across all asyncio.to_thread() calls without locking. + Collects all streaming results and returns them sorted by clause index. + New code should use run_clauses_streaming() directly. """ - tasks = [ - asyncio.to_thread( - process_single_clause, - clause, i, retrieval_service, client, top_k, domains, - ) - for i, clause in enumerate(clauses) - ] - raw = await asyncio.gather(*tasks, return_exceptions=True) - results = [] - for i, r in enumerate(raw): - if isinstance(r, Exception): - logger.warning("Clause {} processing failed: {}", i, r) - results.append({"index": i, "chunks": [], "finding": None}) - else: - results.append(r) - return results + results: list[dict] = [] + async for result in run_clauses_streaming(clauses, retrieval_service, client, top_k, domains): + results.append(result) + return sorted(results, key=lambda r: r["index"]) def check_clause_compliance( @@ -177,6 +307,15 @@ def check_clause_compliance( chunks: list["RetrievedChunk"], client: "BaseLLMClient", ) -> dict | None: + """Check whether a business clause complies with the retrieved regulations. + + The prompt explicitly instructs the LLM to: + - extract clause_ref from the retrieved text (not invent it) + - include a confidence score (0-1) reflecting how well the retrieved + chunks cover the clause topic + + Returns None only when the LLM call fails after all retries. + """ reg_context = "\n".join( f"[{i+1}] {c.doc_title} {c.section_title or ''}: {c.text[:300]}" for i, c in enumerate(chunks[:5]) @@ -186,14 +325,17 @@ def check_clause_compliance( "complies with the retrieved regulations.\n\n" f"Business clause:\n{clause}\n\n" f"Retrieved regulations:\n{reg_context}\n\n" - "Return JSON:\n" + "Return JSON with these exact fields:\n" "{\n" ' "status": "ok" | "warn" | "risk",\n' ' "title": "Short finding title (max 30 chars)",\n' ' "desc": "Description (50-120 chars)",\n' - ' "clause_ref": "Regulation clause reference e.g. Art.9.1 or Sec.3.1"\n' + ' "clause_ref": "Exact clause/article reference copied from the retrieved text above, ' + 'e.g. Art.9.1 or Sec.3.1. Use null if no specific clause number appears in the retrieved text.",\n' + ' "confidence": 0.0-1.0 // how well the retrieved context covers this clause topic\n' "}\n" "status: ok=compliant, warn=gap exists, risk=critical/missing\n" + "IMPORTANT: copy clause_ref verbatim from the retrieved text; do NOT invent references.\n" "Return ONLY the JSON object." ) @@ -216,7 +358,10 @@ def check_clause_compliance( "title": str(result.get("title", "Compliance finding")), "desc": str(result.get("desc", "")), "status": result.get("status", "info"), - "clause_ref": result.get("clause_ref"), + # None if LLM correctly found no clause number in retrieved text + "clause_ref": result.get("clause_ref") or None, + # Confidence score helps frontend show retrieval quality indicator + "confidence": float(result.get("confidence", 0.5)), } except (ValueError, TypeError) as exc: logger.warning("Gap check JSON parse failed: {}", exc) @@ -368,3 +513,58 @@ def generate_suggestions( except (ValueError, TypeError) as exc: logger.warning("generate_suggestions JSON parse failed: {}", exc) return fallback + + +def detect_cross_clause_conflicts( + findings: list[dict], + client: "BaseLLMClient", +) -> list[dict]: + """Detect contradictions and missing cross-references across all findings. + + Runs a single LLM call after all per-clause findings are collected. + Returns a list of conflict dicts: {type, finding_a, finding_b, desc}. + Returns an empty list on LLM failure so the caller can proceed without it. + """ + if len(findings) < 2: + # Need at least 2 findings to compare + return [] + + findings_text = "\n".join( + f"[{i+1}] [{f['status'].upper()}] {f['title']}: {f['desc']}" + + (f" (Ref: {f['clause_ref']})" if f.get("clause_ref") else "") + for i, f in enumerate(findings) + ) + prompt = ( + "You are a compliance expert. Review the following compliance findings from the same document " + "and identify any cross-clause issues:\n\n" + f"Findings:\n{findings_text}\n\n" + "Return JSON array of conflicts (empty array [] if none found):\n" + "[\n" + " {\n" + ' "type": "contradiction" | "missing_ref" | "cumulative_risk",\n' + ' "finding_a": <1-based index>,\n' + ' "finding_b": <1-based index or null>,\n' + ' "desc": "Brief description of the cross-clause issue (max 100 chars)"\n' + " }\n" + "]\n" + "Return ONLY the JSON array." + ) + try: + response = client.chat([{"role": "user", "content": prompt}], max_tokens=600) + if not response.is_success: + return [] + result = _extract_json(response.content) + if isinstance(result, list): + return [ + { + "type": str(c.get("type", "contradiction")), + "finding_a": int(c.get("finding_a", 0)), + "finding_b": c.get("finding_b"), + "desc": str(c.get("desc", "")), + } + for c in result + if isinstance(c, dict) + ] + except Exception as exc: + logger.warning("detect_cross_clause_conflicts failed: {}", exc) + return [] diff --git a/backend/app/application/documents/services.py b/backend/app/application/documents/services.py index c39a57f..5efa6cf 100644 --- a/backend/app/application/documents/services.py +++ b/backend/app/application/documents/services.py @@ -526,10 +526,28 @@ class DocumentCommandService: logger.warning("临时文件清理失败: {}", temp_path) def delete(self, doc_id: str) -> bool: - """Delete document record, binary file, and vector chunks.""" + """Delete document record, binary file, and vector chunks. + + Handles two cases: + - Normal docs: have a metadata record in the document repository. + - Milvus-only (synthetic) docs: visible in management-list because they + have Milvus vectors but no JSON/PG metadata record. We still clean up + the Milvus chunks so the document disappears from the list. + """ document = self.document_repository.get(doc_id) if not document: + # No metadata record — might be a Milvus-only synthetic document. + # Attempt vector cleanup directly; treat as success if any chunks deleted. + try: + deleted_count = self.vector_index.delete_by_document(doc_id) + if deleted_count > 0: + logger.info("Deleted Milvus-only doc (no metadata record): doc_id={} chunks={}", doc_id, deleted_count) + return True + except Exception as exc: + logger.warning("Milvus-only delete failed for doc_id={}: {}", doc_id, exc) return False + + # Normal doc: clean up binary, vectors, artifacts, processing records, metadata. try: self.binary_store.delete(document.object_name) except Exception: @@ -627,13 +645,16 @@ class DocumentQueryService: result.append(doc) # Surface Milvus-only docs that have no metadata record at all. + # MinIO almost certainly has their binaries (they were uploaded), so + # set object_name to the sentinel "{doc_id}/" so the route marks + # has_file=True; the download endpoint will list MinIO to find the file. for doc_id, row in milvus_by_id.items(): if doc_id not in meta_by_id: synthetic = Document( doc_id=doc_id, doc_name=row.get("doc_title", doc_id), file_name=row.get("doc_title", doc_id), - object_name="", + object_name=f"{doc_id}/", # sentinel: MinIO prefix exists content_type="", size_bytes=0, status=DocumentStatus.INDEXED, @@ -646,9 +667,63 @@ class DocumentQueryService: result.sort(key=lambda d: d.updated_at, reverse=True) return result[:limit] if limit is not None else result - def download(self, doc_id: str) -> tuple[Document, bytes]: - """Handle download for the Document Query Service instance.""" + def download(self, doc_id: str) -> tuple["Document", bytes]: + """Return the document record and its binary content from MinIO. + + Fallback strategy for Milvus-only docs (no JSON/PG metadata record): + 1. Try metadata repository first (normal path). + 2. If metadata is missing, list MinIO objects with prefix ``{doc_id}/`` + and synthesise a minimal Document from the first object found. + This handles documents whose metadata records were lost but whose + binary files are still in object storage. + 3. If neither source has the file, raise FileNotFoundError. + """ + from app.domain.documents import Document, DocumentStatus + document = self.document_repository.get(doc_id) - if not document: - raise FileNotFoundError(f"文档不存在: {doc_id}") + + if document and document.object_name and not document.object_name.endswith("/"): + # Normal doc with a concrete object_name — read directly. + return document, self.binary_store.read(document.object_name) + + if document and not document.object_name: + raise FileNotFoundError(f"该文档无原始文件（仅含索引数据，无法下载）: {doc_id}") + + if not document or document.object_name.endswith("/"): + # Metadata missing — try to find the file in MinIO by doc_id prefix. + try: + objects = self.binary_store.list_objects(prefix=f"{doc_id}/") + # Filter out artifact JSON files; prefer the source document. + candidates = [o for o in objects if not o.endswith(".json")] + if not candidates: + candidates = objects # fall back to all objects if only JSON found + if not candidates: + raise FileNotFoundError(f"文档不存在（MinIO 和元数据均无记录）: {doc_id}") + object_name = candidates[0] + file_name = object_name.split("/", 1)[-1] if "/" in object_name else object_name + # Guess content type from extension. + ext = file_name.rsplit(".", 1)[-1].lower() if "." in file_name else "" + _ct_map = { + "pdf": "application/pdf", + "docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + "doc": "application/msword", + "txt": "text/plain", + } + content_type = _ct_map.get(ext, "application/octet-stream") + # Synthesise a minimal Document so the route can build the response. + document = Document( + doc_id=doc_id, + doc_name=file_name, + file_name=file_name, + object_name=object_name, + content_type=content_type, + size_bytes=0, + status=DocumentStatus.INDEXED, + ) + logger.info("MinIO fallback download: doc_id={} object={}", doc_id, object_name) + except FileNotFoundError: + raise + except Exception as exc: + raise FileNotFoundError(f"文档不存在: {doc_id}") from exc + return document, self.binary_store.read(document.object_name) diff --git a/backend/app/config/settings.py b/backend/app/config/settings.py index 917ab51..eb3dcef 100644 --- a/backend/app/config/settings.py +++ b/backend/app/config/settings.py @@ -133,6 +133,42 @@ class Settings(BaseSettings): reranker_api_key: str = Field(default="", description="Reranker API 密钥") reranker_top_k: int = Field(default=5, description="精排后保留的最终结果数量") + # ── HyDE (Hypothetical Document Embeddings) ────────────────────────────── + # When enabled, the agentic and standard RAG pipelines generate a short + # hypothetical answer before retrieval, then embed that text instead of the + # raw query. This closes the vocabulary gap between terse queries and longer + # document passages, typically improving recall by 15-30% on vague queries. + hyde_enabled: bool = Field(default=True, description="启用 HyDE 查询增强（假设文档嵌入）") + hyde_max_tokens: int = Field(default=200, description="HyDE 假设段落最大 token 数") + # Use a lightweight model for HyDE to reduce latency and cost. + # HyDE only needs a short plausible passage — a fast cheap model is sufficient. + # Leave empty to fall back to the main llm_provider / llm_model. + hyde_llm_provider: str = Field(default="", description="HyDE 专用 LLM 提供商（空则复用主 LLM）") + hyde_llm_model: str = Field(default="", description="HyDE 专用 LLM 模型（空则复用主 LLM）") + + # ── Agentic RAG (P0-1) ─────────────────────────────────────────────────── + # Controls the multi-step reasoning pipeline exposed at /agent/agentic/stream. + agentic_max_sub_queries: int = Field( + default=4, + description="Agentic 模式最大子查询分解数量（compare / multi_hop 意图触发）", + ) + agentic_grounding_threshold: float = Field( + default=0.65, + description=( + "引文锚定 fast-path 阈值：avg_score > 此值且 chunks ≥ 3 时跳过 LLM grounding check，" + "直接判定为充分；降低此值可让更多问题触发 LLM 二次验证。" + ), + ) + agentic_intent_max_tokens: int = Field( + default=200, description="意图分析步骤 LLM 最大 token 数" + ) + agentic_plan_max_tokens: int = Field( + default=400, description="查询分解步骤 LLM 最大 token 数" + ) + agentic_grounding_max_tokens: int = Field( + default=250, description="引文锚定步骤 LLM 最大 token 数" + ) + # Keep configuration setup explicit so runtime behavior is easy to reason about. milvus_index_type: str = Field(default="IVF_FLAT", description="Milvus索引类型") milvus_nlist: int = Field(default=128, description="Milvus nlist参数") diff --git a/backend/app/infrastructure/llm/openai_compatible_answer_generator.py b/backend/app/infrastructure/llm/openai_compatible_answer_generator.py index fc89f34..06e279d 100644 --- a/backend/app/infrastructure/llm/openai_compatible_answer_generator.py +++ b/backend/app/infrastructure/llm/openai_compatible_answer_generator.py @@ -9,10 +9,12 @@ from app.config.settings import settings from app.domain.conversation import AnswerGenerator, AnswerResult, AnswerSource from app.domain.retrieval import RetrievedChunk from app.services.llm.llm_factory import get_llm_client +from app.services.rag.prompt_templates import PromptTemplates # Keep adapter behavior explicit so integration details remain easy to audit. -PROMPT_TEMPLATES = { +# Fallback system prompts used when no rich template matches. +_FALLBACK_PROMPTS = { "default": "你是法规知识问答助手。请仅依据提供的上下文回答；如果上下文不足，明确说明。", "compliance_qa": "你是法规合规问答助手。优先引用给定法规原文，回答要准确、克制，并注明依据来源。", } @@ -38,33 +40,80 @@ class OpenAICompatibleAnswerGenerator(AnswerGenerator): retrieved_chunks: list[RetrievedChunk], history: list[dict[str, str]] | None, prompt_template: str | None, + context_text: str | None = None, + context_filename: str | None = None, ) -> tuple[list[dict[str, str]], int]: - """Handle build messages for this module for the Open A I Compatible Answer Generator instance.""" - system_prompt = PROMPT_TEMPLATES.get(prompt_template or "compliance_qa", PROMPT_TEMPLATES["default"]) + """Build the message list to send to the LLM. + + When context_text is provided the user's document is injected as a + dedicated section BEFORE the retrieved regulation chunks so the LLM + can reason about the document directly while still referencing regulations. + The retrieval step uses only the user's question, not the document text, + so embedding quality is preserved. + + System prompt selection priority: + 1. Rich template from PromptTemplates (compliance_qa / comparison / + compliance_check / clause_interpretation / …) + 2. Fallback hardcoded prompt when no rich template matches. + """ + # Look up the rich template first; fall back to simple hardcoded prompts. + tpl_name = prompt_template or "compliance_qa" + rich_tpl = PromptTemplates.get_template(tpl_name) + if rich_tpl: + system_prompt = rich_tpl.system_prompt + else: + system_prompt = _FALLBACK_PROMPTS.get(tpl_name, _FALLBACK_PROMPTS["default"]) context_blocks = [] context_tokens = 0 + + # ── User document context (if attached) ─────────────────────────────── + if context_text and context_text.strip(): + doc_label = f"附件文档：{context_filename}" if context_filename else "附件文档" + doc_block = f"[{doc_label}]\n{context_text.strip()}" + doc_tokens = self._estimate_tokens(doc_block) + # Reserve at most half the context budget for the user document + half_budget = settings.rag_max_context_tokens // 2 + if doc_tokens > half_budget: + # Truncate document to fit half the budget + ratio = half_budget / doc_tokens + doc_block = doc_block[: int(len(doc_block) * ratio)] + "\n…（文档已截断）" + doc_tokens = half_budget + context_blocks.append(doc_block) + context_tokens += doc_tokens + + # ── Retrieved regulation chunks ──────────────────────────────────────── + remaining_budget = settings.rag_max_context_tokens - context_tokens for idx, chunk in enumerate(retrieved_chunks, start=1): block = ( - f"[{idx}] 文档: {chunk.doc_title}\n" + f"[法规{idx}] 文档: {chunk.doc_title}\n" f"章节: {chunk.section_title or '未标注'}\n" f"页码: {chunk.page_start}" + (f"-{chunk.page_end}" if chunk.page_end and chunk.page_end != chunk.page_start else "") + "\n" f"内容: {chunk.text}" ) block_tokens = self._estimate_tokens(block) - if context_tokens + block_tokens > settings.rag_max_context_tokens: + if block_tokens > remaining_budget: break + remaining_budget -= block_tokens context_tokens += block_tokens context_blocks.append(block) + context = "\n\n".join(context_blocks) messages = [{"role": "system", "content": system_prompt}] for item in history or []: messages.append({"role": item["role"], "content": item["content"]}) - messages.append( - { - "role": "user", - "content": f"问题：{query}\n\n参考上下文：\n{context}\n\n请在回答后给出简要引用编号。", - } - ) + + # Craft the user turn differently when a document is attached + if context_text and context_text.strip(): + user_content = ( + f"问题：{query}\n\n" + f"请先基于上方附件文档内容进行分析，再结合法规参考上下文给出合规评估。" + f"\n\n参考上下文：\n{context}\n\n" + f"请在回答中注明引用来源编号（如适用）。" + ) + else: + user_content = f"问题：{query}\n\n参考上下文：\n{context}\n\n请在回答后给出简要引用编号。" + + messages.append({"role": "user", "content": user_content}) return messages, context_tokens def _is_context_truncated(self, *, retrieved_chunks: list[RetrievedChunk], context_tokens: int) -> bool: @@ -112,6 +161,8 @@ class OpenAICompatibleAnswerGenerator(AnswerGenerator): provider: str | None = None, model: str | None = None, prompt_template: str | None = None, + context_text: str | None = None, + context_filename: str | None = None, ) -> AnswerResult: """Handle generate for the Open A I Compatible Answer Generator instance.""" start = time.time() @@ -120,6 +171,8 @@ class OpenAICompatibleAnswerGenerator(AnswerGenerator): retrieved_chunks=retrieved_chunks, history=history, prompt_template=prompt_template, + context_text=context_text, + context_filename=context_filename, ) client = get_llm_client(provider=provider or settings.llm_provider, model=model or settings.llm_model) response = client.chat(messages) @@ -147,6 +200,8 @@ class OpenAICompatibleAnswerGenerator(AnswerGenerator): provider: str | None = None, model: str | None = None, prompt_template: str | None = None, + context_text: str | None = None, + context_filename: str | None = None, ) -> Generator[dict, None, AnswerResult]: """Stream generate for the Open A I Compatible Answer Generator instance.""" start = time.time() @@ -155,6 +210,8 @@ class OpenAICompatibleAnswerGenerator(AnswerGenerator): retrieved_chunks=retrieved_chunks, history=history, prompt_template=prompt_template, + context_text=context_text, + context_filename=context_filename, ) sources = [source.__dict__ for source in self._sources(retrieved_chunks)] yield {"event": "sources", "data": sources} diff --git a/backend/app/infrastructure/storage/minio_binary_store.py b/backend/app/infrastructure/storage/minio_binary_store.py index 034b46f..6bce3ff 100644 --- a/backend/app/infrastructure/storage/minio_binary_store.py +++ b/backend/app/infrastructure/storage/minio_binary_store.py @@ -41,6 +41,10 @@ class MinioDocumentBinaryStore(DocumentBinaryStore): raise FileNotFoundError(f"对象不存在: {object_name}") return data + def list_objects(self, prefix: str = "") -> list[str]: + """List object names in the bucket that start with the given prefix.""" + return self.client.list_objects(prefix=prefix) + def delete(self, object_name: str) -> None: """Handle delete for the Minio Document Binary Store instance.""" if not self.client.delete_object(object_name): diff --git a/backend/app/infrastructure/vectorstore/cross_encoder_reranker.py b/backend/app/infrastructure/vectorstore/cross_encoder_reranker.py index f3c0326..ab65db9 100644 --- a/backend/app/infrastructure/vectorstore/cross_encoder_reranker.py +++ b/backend/app/infrastructure/vectorstore/cross_encoder_reranker.py @@ -71,22 +71,48 @@ class OpenAICompatibleReranker(Reranker): return result def _call_reranker(self, query: str, texts: list[str]) -> list[float]: - """Call the reranker API and return a score per text.""" + """Call the reranker API and return a score per text. + + Tries TEI format first (POST /rerank with model+texts), then falls back + to Cohere/OpenAI format (POST /v1/rerank with model+documents). + Both formats now include the model name, which most gateways require. + """ headers = {"Content-Type": "application/json"} if self._api_key: headers["Authorization"] = f"Bearer {self._api_key}" - # Try TEI format first: POST /rerank - payload = {"query": query, "texts": texts, "raw_scores": False, "return_text": False} + # TEI format: POST /rerank — include model name (required by gateway proxies) + payload = { + "model": self._model, + "query": query, + "texts": texts, + "raw_scores": False, + "return_text": False, + } url = f"{self._base_url}/rerank" resp = requests.post(url, json=payload, headers=headers, timeout=self._timeout) - if resp.status_code == 404: - # Fall back to Cohere / OpenAI-style: POST /v1/rerank + if resp.status_code in (404, 400): + # Gateway returned an error — try Cohere/OpenAI-style format as fallback. + logger.debug( + "TEI rerank returned {} — trying Cohere format. Body: {}", + resp.status_code, + resp.text[:200], + ) payload_v1 = {"model": self._model, "query": query, "documents": texts} url = f"{self._base_url}/v1/rerank" resp = requests.post(url, json=payload_v1, headers=headers, timeout=self._timeout) + if not resp.ok: + # Surface a clear error message so callers can log it meaningfully. + try: + err_body = resp.json() + err_msg = err_body.get("error", {}).get("message", resp.text[:200]) + except Exception: + err_msg = resp.text[:200] + resp.raise_for_status() # raises HTTPError with status code + raise ValueError(err_msg) # unreachable but satisfies type checker + resp.raise_for_status() data = resp.json() diff --git a/backend/app/schemas/rag.py b/backend/app/schemas/rag.py index 3b72209..f6ab0c5 100644 --- a/backend/app/schemas/rag.py +++ b/backend/app/schemas/rag.py @@ -12,6 +12,11 @@ class RagChatRequest(BaseModel): top_k: int = 5 session_id: Optional[str] = None filters: Optional[str] = None + # Optional document text to inject directly as LLM conversation context. + # When provided the document content is prepended to the query so the LLM + # can answer questions about it without requiring vector-store indexing. + context_text: Optional[str] = None + context_filename: Optional[str] = None class RetrievedDoc(BaseModel): diff --git a/backend/app/services/llm/base_client.py b/backend/app/services/llm/base_client.py index 3070a97..9d768cf 100644 --- a/backend/app/services/llm/base_client.py +++ b/backend/app/services/llm/base_client.py @@ -1,9 +1,16 @@ -"""Provide service-layer logic for base client.""" +"""Provide service-layer logic for base client. + +P0-0: ``LLMResponse`` now carries an optional ``tool_calls`` list so that any +downstream code (agents, pipelines) can inspect and dispatch tool invocations +without touching the provider-specific adapter layer. +""" from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import List, Dict, Optional, Any from enum import Enum + +from app.services.llm.tool_types import Tool, ToolCall # noqa: F401 – re-exported for callers # Keep provider-specific behavior explicit so debugging stays straightforward. @@ -24,6 +31,8 @@ class LLMResponse: finish_reason: str = "stop" latency_ms: int = 0 error: Optional[str] = None + # P0-0: populated when the model returns tool-call(s) instead of plain text. + tool_calls: List[ToolCall] = field(default_factory=list) @property def is_success(self) -> bool: @@ -63,9 +72,19 @@ class BaseLLMClient(ABC): messages: List[Dict[str, str]], max_tokens: Optional[int] = None, temperature: Optional[float] = None, + tools: Optional[List["Tool"]] = None, **kwargs ) -> LLMResponse: - """Handle chat for the Base L L M Client instance.""" + """Handle chat for the Base L L M Client instance. + + Args: + messages: OpenAI-format message list. + max_tokens: Override config max_tokens when set. + temperature: Override config temperature when set. + tools: Optional list of Tool definitions to offer the model. + When provided, the model may respond with tool_calls in the + returned LLMResponse instead of (or in addition to) content. + """ pass def complete( diff --git a/backend/app/services/llm/deepseek_client.py b/backend/app/services/llm/deepseek_client.py index 22ed427..d0af5cb 100644 --- a/backend/app/services/llm/deepseek_client.py +++ b/backend/app/services/llm/deepseek_client.py @@ -1,4 +1,8 @@ -"""Provide service-layer logic for deepseek client.""" +"""Provide service-layer logic for deepseek client. + +P0-0: ``chat()`` now accepts an optional ``tools`` list and parses ``tool_calls`` +from the model response so that callers can dispatch tool invocations. +""" import time from typing import List, Dict, Optional @@ -6,6 +10,7 @@ from loguru import logger import httpx from .base_client import BaseLLMClient, LLMResponse, LLMConfig, LLMProvider +from .tool_types import Tool, ToolCall # Keep provider-specific behavior explicit so debugging stays straightforward. @@ -46,13 +51,20 @@ class DeepSeekClient(BaseLLMClient): messages: List[Dict[str, str]], max_tokens: Optional[int] = None, temperature: Optional[float] = None, + tools: Optional[List[Tool]] = None, **kwargs ) -> LLMResponse: - """Handle chat for the Deep Seek Client instance.""" + """Handle chat for the Deep Seek Client instance. + + When ``tools`` is provided the request includes the tool definitions and + ``tool_choice="auto"``; any tool_calls returned by the model are parsed + into ``LLMResponse.tool_calls``. + """ + import json start_time = time.time() try: - payload = { + payload: Dict = { "model": self.config.model, "messages": messages, "max_tokens": max_tokens or self.config.max_tokens, @@ -61,6 +73,11 @@ class DeepSeekClient(BaseLLMClient): "stream": False } + # P0-0: inject tool definitions when provided. + if tools: + payload["tools"] = [t.to_openai_format() for t in tools] + payload["tool_choice"] = "auto" + response = self._client.post("/chat/completions", json=payload) response.raise_for_status() @@ -71,12 +88,24 @@ class DeepSeekClient(BaseLLMClient): choices = data.get("choices", [{}]) message = choices[0].get("message", {}) + # P0-0: parse tool_calls returned by the model. + raw_tool_calls = message.get("tool_calls") or [] + parsed_tool_calls: List[ToolCall] = [] + for tc in raw_tool_calls: + fn = tc.get("function", {}) + try: + args = json.loads(fn.get("arguments", "{}")) + except json.JSONDecodeError: + args = {} + parsed_tool_calls.append(ToolCall(id=tc.get("id", ""), name=fn.get("name", ""), arguments=args)) + return LLMResponse( - content=message.get("content", ""), + content=message.get("content", "") or "", model=data.get("model", self.config.model), usage=data.get("usage", {}), finish_reason=choices[0].get("finish_reason", "stop"), - latency_ms=latency_ms + latency_ms=latency_ms, + tool_calls=parsed_tool_calls, ) except httpx.HTTPStatusError as e: diff --git a/backend/app/services/llm/qwen_client.py b/backend/app/services/llm/qwen_client.py index b3bd5d6..39d179e 100644 --- a/backend/app/services/llm/qwen_client.py +++ b/backend/app/services/llm/qwen_client.py @@ -1,4 +1,8 @@ -"""Provide service-layer logic for qwen client.""" +"""Provide service-layer logic for qwen client. + +P0-0: ``chat()`` now accepts an optional ``tools`` list and parses ``tool_calls`` +from the model response so that callers can dispatch tool invocations. +""" import time import json @@ -7,6 +11,7 @@ from loguru import logger import httpx from .base_client import BaseLLMClient, LLMResponse, LLMConfig, LLMProvider +from .tool_types import Tool, ToolCall # Keep provider-specific behavior explicit so debugging stays straightforward. @@ -54,14 +59,20 @@ class QwenClient(BaseLLMClient): messages: List[Dict[str, str]], max_tokens: Optional[int] = None, temperature: Optional[float] = None, + tools: Optional[List[Tool]] = None, **kwargs ) -> LLMResponse: - """Handle chat for the Qwen Client instance.""" + """Handle chat for the Qwen Client instance. + + When ``tools`` is provided the request includes the tool definitions and + ``tool_choice="auto"``; any tool_calls returned by the model are parsed + into ``LLMResponse.tool_calls``. + """ start_time = time.time() try: # Keep provider-specific behavior explicit so debugging stays straightforward. - payload = { + payload: Dict = { "model": self.config.model, "messages": messages, "max_tokens": max_tokens or self.config.max_tokens, @@ -70,6 +81,11 @@ class QwenClient(BaseLLMClient): "stream": False } + # P0-0: inject tool definitions when provided. + if tools: + payload["tools"] = [t.to_openai_format() for t in tools] + payload["tool_choice"] = "auto" + # Keep provider-specific behavior explicit so debugging stays straightforward. response = self._client.post("/chat/completions", json=payload) response.raise_for_status() @@ -82,12 +98,24 @@ class QwenClient(BaseLLMClient): choices = data.get("choices", [{}]) message = choices[0].get("message", {}) + # P0-0: parse tool_calls returned by the model. + raw_tool_calls = message.get("tool_calls") or [] + parsed_tool_calls: List[ToolCall] = [] + for tc in raw_tool_calls: + fn = tc.get("function", {}) + try: + args = json.loads(fn.get("arguments", "{}")) + except json.JSONDecodeError: + args = {} + parsed_tool_calls.append(ToolCall(id=tc.get("id", ""), name=fn.get("name", ""), arguments=args)) + return LLMResponse( - content=message.get("content", ""), + content=message.get("content", "") or "", model=data.get("model", self.config.model), usage=data.get("usage", {}), finish_reason=choices[0].get("finish_reason", "stop"), - latency_ms=latency_ms + latency_ms=latency_ms, + tool_calls=parsed_tool_calls, ) except httpx.HTTPStatusError as e: diff --git a/backend/app/services/llm/tool_types.py b/backend/app/services/llm/tool_types.py new file mode 100644 index 0000000..1eaa76c --- /dev/null +++ b/backend/app/services/llm/tool_types.py @@ -0,0 +1,80 @@ +"""Shared tool and tool-call type definitions for LLM function calling (P0-0). + +These types implement the OpenAI-compatible tool/function-calling interface so that +any provider whose gateway supports the spec (DeepSeek, Qwen, etc.) can expose +tools to the LLM and receive structured tool invocations in return. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class ToolCall: + """Represent a single tool invocation returned by the LLM. + + The model fills in ``id``, ``name``, and ``arguments`` when it decides to call + a tool instead of (or in addition to) producing a text response. + """ + + # Unique identifier assigned by the model for this call. + id: str + # Name of the tool to invoke, matching the name registered in Tool. + name: str + # Parsed JSON arguments ready for direct use by the tool handler. + arguments: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class ToolParameter: + """JSON-Schema–compatible parameter block for a tool definition.""" + + # Top-level schema type — always "object" for OpenAI-compatible tools. + type: str = "object" + # Map of parameter name → JSON-Schema property descriptor. + properties: dict[str, Any] = field(default_factory=dict) + # List of required parameter names. + required: list[str] = field(default_factory=list) + + +@dataclass +class Tool: + """Describe a callable tool that can be offered to the LLM. + + Example usage:: + + search_tool = Tool( + name="search_regulations", + description="Search the compliance knowledge base for relevant regulation clauses.", + parameters=ToolParameter( + properties={"query": {"type": "string", "description": "Search query"}}, + required=["query"], + ), + ) + response = client.chat(messages, tools=[search_tool]) + """ + + name: str + description: str + parameters: ToolParameter = field(default_factory=ToolParameter) + + def to_openai_format(self) -> dict[str, Any]: + """Serialise this tool to the OpenAI-compatible function-calling schema. + + The returned dict can be placed directly in the ``tools`` list of a chat + completions request without any further transformation. + """ + return { + "type": "function", + "function": { + "name": self.name, + "description": self.description, + "parameters": { + "type": self.parameters.type, + "properties": self.parameters.properties, + "required": self.parameters.required, + }, + }, + } diff --git a/backend/app/shared/bootstrap.py b/backend/app/shared/bootstrap.py index b016913..69fa1ba 100644 --- a/backend/app/shared/bootstrap.py +++ b/backend/app/shared/bootstrap.py @@ -6,6 +6,7 @@ from functools import lru_cache from typing import Callable from app.application.agent import AgentConversationService, AgentSessionService +from app.application.agent.agentic_service import AgenticConversationService from app.application.documents import DocumentCommandService, DocumentQueryService from app.application.knowledge import KnowledgeRetrievalService from app.application.perception.services import PerceptionService @@ -365,6 +366,20 @@ def get_agent_session_service() -> AgentSessionService: return AgentSessionService(conversation_store=get_conversation_store()) +@lru_cache +def get_agentic_conversation_service() -> AgenticConversationService: + """Return the Agentic RAG service (P0-1). + + Uses the same retrieval, generation, and session infrastructure as the + standard chat service so no additional dependencies are required. + """ + return AgenticConversationService( + retrieval_service=get_retrieval_service(), + answer_generator=OpenAICompatibleAnswerGenerator(), + conversation_store=get_conversation_store(), + ) + + @lru_cache def get_celery_app(): """Return the shared Celery application instance. diff --git a/frontend/src/api/index.ts b/frontend/src/api/index.ts index 40205cf..de97334 100644 --- a/frontend/src/api/index.ts +++ b/frontend/src/api/index.ts @@ -73,6 +73,22 @@ export interface SSEMessage { text?: string; docs?: RetrievedDoc[]; session_id?: string; + // ── P0-1 Agentic-mode thinking-step fields ──────────────────────────────── + // Populated when type === 'thinking'; maps to the backend IntentResult / + // GroundingResult / retrieval step payloads emitted by AgenticConversationService. + step?: string; // intent_analysis | query_planning | retrieving | grounding_check + status?: string; // running | done + intent_type?: string; // simple_qa | compare | multi_hop | ambiguous + requires_decomposition?: boolean; + reason?: string; + sub_queries?: string[]; + query?: string; // sub-query being retrieved + index?: number; // 1-based sub-query index + total?: number; // total sub-query count + found?: number; // chunks found for this sub-query + retry?: boolean; // true when this is a grounding-failure re-query + sufficient?: boolean; // grounding check result + confidence?: number; // grounding confidence 0–1 } export async function streamSSE( diff --git a/frontend/src/api/rag.ts b/frontend/src/api/rag.ts index c182268..105de11 100644 --- a/frontend/src/api/rag.ts +++ b/frontend/src/api/rag.ts @@ -76,6 +76,27 @@ function parseSSEChunk(raw: string, onMessage: (data: SSEMessage) => void) { onMessage({ type: 'error', text: joined }); } else if (eventName === 'status') { onMessage({ type: 'status', text: joined }); + } else if (eventName === 'thinking') { + // P0-1: Agentic reasoning step events from /agent/agentic/stream + try { + const payload = JSON.parse(joined) as Record; + onMessage({ + type: 'thinking', + step: payload.step as string | undefined, + status: payload.status as string | undefined, + intent_type: payload.intent_type as string | undefined, + requires_decomposition: payload.requires_decomposition as boolean | undefined, + reason: payload.reason as string | undefined, + sub_queries: payload.sub_queries as string[] | undefined, + query: payload.query as string | undefined, + index: payload.index as number | undefined, + total: payload.total as number | undefined, + found: payload.found as number | undefined, + retry: payload.retry as boolean | undefined, + sufficient: payload.sufficient as boolean | undefined, + confidence: payload.confidence as number | undefined, + }); + } catch { /* ignore */ } } else if (eventName === 'message') { // /rag/chat format: event:message + JSON body with type field try { @@ -147,3 +168,73 @@ export async function ragChat( } export type { QuickQuestionsResponse, SSEMessage }; + +/** + * P0-1 Agentic RAG chat — calls /agent/agentic/stream which runs the full + * intent-analysis → query-planning → retrieval → grounding-check → answer pipeline. + * + * The onMessage callback receives the same event types as ragChat plus + * ``type: 'thinking'`` events that carry live reasoning-step progress. + */ +export async function agenticChat( + query: string, + topK: number = 5, + onMessage: (data: SSEMessage) => void, + onError?: (error: Error) => void, + onComplete?: () => void, + filters?: string, + sessionId?: string, + signal?: AbortSignal, + contextText?: string, + contextFilename?: string, +): Promise { + try { + const response = await fetch(`${AGENT_API_BASE}/agent/agentic/stream`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + Accept: 'text/event-stream', + ...(getToken() ? { Authorization: `Bearer ${getToken()}` } : {}), + }, + body: JSON.stringify({ + query, + top_k: topK, + ...(filters ? { filters } : {}), + ...(sessionId ? { session_id: sessionId } : {}), + ...(contextText ? { context_text: contextText, context_filename: contextFilename ?? '' } : {}), + }), + signal, + }); + + if (!response.ok || !response.body) { + throw new Error(`HTTP error! status: ${response.status}`); + } + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + + buffer += decoder.decode(value, { stream: true }); + const parts = buffer.split('\n\n'); + buffer = parts.pop() || ''; + parseSSEChunk(parts.join('\n\n'), onMessage); + } + + if (buffer.trim()) { + parseSSEChunk(buffer, onMessage); + } + + if (onComplete) { + onComplete(); + } + } catch (error) { + if (error instanceof DOMException && error.name === 'AbortError') return; + if (onError) { + onError(error instanceof Error ? error : new Error(String(error))); + } + } +} diff --git a/frontend/src/contexts/PageStateContext.tsx b/frontend/src/contexts/PageStateContext.tsx index 0dfad19..b9127ba 100644 --- a/frontend/src/contexts/PageStateContext.tsx +++ b/frontend/src/contexts/PageStateContext.tsx @@ -59,6 +59,8 @@ export interface ComplianceSourceEvent { score: number; status: string; full_content: string; + /** Index of the clause this source was retrieved for (for source↔finding linking) */ + clause_index?: number; } export interface ComplianceFindingEvent { @@ -66,6 +68,17 @@ export interface ComplianceFindingEvent { desc: string; status: 'ok' | 'warn' | 'risk'; clause_ref?: string; + /** LLM confidence that retrieved context covers the clause topic (0–1) */ + confidence?: number; + /** Top-3 regulation chunks that informed this finding */ + source_refs?: Array<{ standard: string; clause: string; score: number }>; +} + +export interface ComplianceConflict { + type: 'contradiction' | 'missing_ref' | 'cumulative_risk'; + finding_a: number; + finding_b: number | null; + desc: string; } export interface ComplianceActionItem { @@ -103,6 +116,10 @@ export interface ComplianceState { analysisId: string | null; isReadOnly: boolean; activeFindingId: string | null; + /** Real-time per-clause progress {done, total} */ + progress: { done: number; total: number } | null; + /** Cross-clause conflicts detected after all findings complete */ + conflicts: ComplianceConflict[]; } const COMPLIANCE_INIT: ComplianceState = { @@ -117,6 +134,8 @@ const COMPLIANCE_INIT: ComplianceState = { analysisId: null, isReadOnly: false, activeFindingId: null, + progress: null, + conflicts: [], }; // ── Perception types ────────────────────────────────────────────────────────── diff --git a/frontend/src/contexts/index.ts b/frontend/src/contexts/index.ts index bc3a85b..89a9b59 100644 --- a/frontend/src/contexts/index.ts +++ b/frontend/src/contexts/index.ts @@ -12,6 +12,7 @@ export type { ComplianceStatus, ComplianceSourceEvent, ComplianceFindingEvent, + ComplianceConflict, ComplianceDonePayload, ComplianceMeta, ComplianceActionItem, diff --git a/frontend/src/locales/en.ts b/frontend/src/locales/en.ts index f69ef43..50cba4a 100644 --- a/frontend/src/locales/en.ts +++ b/frontend/src/locales/en.ts @@ -238,6 +238,36 @@ export interface Translations { citationsHeader: string; citationsEmpty: string; apiError: string; + // ── Agentic mode ───────────────────────────────────────────────────────── + agenticMode: string; + agenticModeHint: string; + agentThinking: string; + agentDone: string; + stepSuffix: string; + stepIntentAnalysis: string; + stepQueryPlanning: string; + stepRetrieving: string; + stepGrounding: string; + intentSimpleQa: string; + intentCompare: string; + intentMultiHop: string; + intentAmbiguous: string; + intentNeedsDecomposition: string; + subQueriesCountSuffix: string; + chunksFoundSuffix: string; + retryLabel: string; + groundingSufficient: string; + groundingInsufficient: string; + // ── Document attachment in interface ───────────────────────────────────── + attachBtn: string; + attachExtracting: string; + attachReady: string; + attachError: string; + attachClearLabel: string; + attachContextBadge: string; + attachAccept: string; + attachTruncated: string; + attachErrorMsg: string; }; } @@ -480,5 +510,35 @@ export const en: Translations = { citationsHeader: 'Sources', citationsEmpty: 'Citations will appear here after a response is generated.', apiError: 'Could not reach the RAG API. Please check the backend.', + // ── Agentic mode ───────────────────────────────────────────────────────── + agenticMode: 'Agentic mode', + agenticModeHint: 'Intent · Planning · Retrieval · Grounding', + agentThinking: 'Agent reasoning…', + agentDone: 'Reasoning complete', + stepSuffix: 'steps', + stepIntentAnalysis: 'Intent analysis', + stepQueryPlanning: 'Query planning', + stepRetrieving: 'Knowledge retrieval', + stepGrounding: 'Citation grounding', + intentSimpleQa: 'Simple Q&A', + intentCompare: 'Comparison', + intentMultiHop: 'Multi-hop', + intentAmbiguous: 'Ambiguous', + intentNeedsDecomposition: 'Decomposed', + subQueriesCountSuffix: 'sub-queries', + chunksFoundSuffix: 'chunks', + retryLabel: '(retry) ', + groundingSufficient: '✓ Sufficient', + groundingInsufficient: '⚠ Re-queried', + // ── Document context attachment ─────────────────────────────────────────── + attachBtn: 'Attach document as context', + attachExtracting: 'Extracting text…', + attachReady: 'Context loaded', + attachError: 'Extraction failed', + attachClearLabel: 'Clear', + attachContextBadge: 'Doc context', + attachAccept: '.pdf,.docx,.doc,.txt,.md', + attachTruncated: '(truncated to 8 000 chars)', + attachErrorMsg: 'Could not extract text from this file.', }, }; diff --git a/frontend/src/locales/zh.ts b/frontend/src/locales/zh.ts index f501d85..68e9532 100644 --- a/frontend/src/locales/zh.ts +++ b/frontend/src/locales/zh.ts @@ -239,5 +239,35 @@ export const zh: Translations = { citationsHeader: '引用来源', citationsEmpty: '生成回答后，引用来源将显示在此处。', apiError: '无法连接到 RAG API，请检查后端服务。', + // ── Agentic mode ───────────────────────────────────────────────────────── + agenticMode: 'Agentic 模式', + agenticModeHint: '意图分析 · 查询分解 · 迭代检索 · 引文锚定', + agentThinking: 'Agent 推理中…', + agentDone: '推理完成', + stepSuffix: '步', + stepIntentAnalysis: '意图分析', + stepQueryPlanning: '查询分解', + stepRetrieving: '知识检索', + stepGrounding: '引文锚定', + intentSimpleQa: '单跳问答', + intentCompare: '对比分析', + intentMultiHop: '多跳推理', + intentAmbiguous: '模糊查询', + intentNeedsDecomposition: '需分解', + subQueriesCountSuffix: '个子查询', + chunksFoundSuffix: '条', + retryLabel: '(补充) ', + groundingSufficient: '✓ 充分', + groundingInsufficient: '⚠ 补充检索', + // ── Document context attachment ─────────────────────────────────────────── + attachBtn: '上传文档作为对话上下文', + attachExtracting: '正在提取文本…', + attachReady: '上下文已加载', + attachError: '提取失败', + attachClearLabel: '清除', + attachContextBadge: '文档上下文', + attachAccept: '.pdf,.docx,.doc,.txt,.md', + attachTruncated: '（已截断至 8000 字符）', + attachErrorMsg: '无法从该文件提取文本，请检查文件格式。', }, }; diff --git a/frontend/src/pages/Compliance/CompliancePage.tsx b/frontend/src/pages/Compliance/CompliancePage.tsx index 81f84f6..dc6a717 100644 --- a/frontend/src/pages/Compliance/CompliancePage.tsx +++ b/frontend/src/pages/Compliance/CompliancePage.tsx @@ -1,6 +1,6 @@ import { useState, useRef, useEffect } from 'react'; import { useLanguage } from '../../contexts/LanguageContext'; -import { Search, Plus, AlertTriangle, Download, MessageSquare, ChevronDown } from 'lucide-react'; +import { Search, Plus, Download, MessageSquare, ChevronDown, AlertTriangle } from 'lucide-react'; import { Topbar } from '../../components/layout/Topbar'; import { NewAnalysisModal } from './NewAnalysisModal'; import { useComplianceAnalysis } from './useComplianceAnalysis'; @@ -39,81 +39,8 @@ function formatTs(iso: string) { } catch { return iso; } } -// ── Chat state for a single finding ───────────────────────────────────────── -interface ChatMsg { id: number; role: 'user' | 'assistant'; content: string } - -function useFindingChat() { - const [open, setOpen] = useState(false); - const [findingIdx, setFindingIdx] = useState(null); - const [messages, setMessages] = useState([]); - const [input, setInput] = useState(''); - const [loading, setLoading] = useState(false); - const abortRef = useRef(null); - - function openFor(idx: number, finding: FindingEvent) { - setFindingIdx(idx); - setOpen(true); - setMessages([{ - id: 0, - role: 'assistant', - content: `I'm reviewing finding: **${finding.title}**\n\n${finding.desc}${finding.clause_ref ? `\n\nRef: ${finding.clause_ref}` : ''}\n\nHow can I help?`, - }]); - setInput(''); - } - - function close() { setOpen(false); abortRef.current?.abort(); } - - async function send(segmentContext: string) { - if (!input.trim() || loading) return; - const q = input.trim(); - setInput(''); - const userMsg: ChatMsg = { id: Date.now(), role: 'user', content: q }; - const assistantId = Date.now() + 1; - setMessages(m => [...m, userMsg, { id: assistantId, role: 'assistant', content: '' }]); - setLoading(true); - - const ctrl = new AbortController(); - abortRef.current = ctrl; - - try { - const res = await fetch(`/api/v1/compliance/chat/${findingIdx ?? 0}`, { - method: 'POST', - headers: { 'Content-Type': 'application/json', ...authHeader() }, - body: JSON.stringify({ query: q, segment_context: segmentContext }), - signal: ctrl.signal, - }); - if (!res.body) { setLoading(false); return; } - const reader = res.body.getReader(); - const dec = new TextDecoder(); - let buf = ''; - while (true) { - const { done, value } = await reader.read(); - if (done) break; - buf += dec.decode(value, { stream: true }); - const blocks = buf.split('\n\n'); - buf = blocks.pop() ?? ''; - for (const block of blocks) { - const dl = block.split('\n').find(l => l.startsWith('data: ')); - if (!dl) continue; - try { - const j = JSON.parse(dl.slice(6)); - if (j.type === 'chunk' && j.text) { - setMessages(m => m.map(msg => msg.id === assistantId ? { ...msg, content: msg.content + j.text } : msg)); - } - } catch { /* skip */ } - } - } - } catch (e: unknown) { - if (e instanceof Error && e.name === 'AbortError') return; - } finally { - setLoading(false); - } - } - - return { open, findingIdx, messages, input, setInput, loading, openFor, close, send }; -} - -function _FindingChatDrawerWrapper({ +/** Wrapper that resolves findingIndex → findingId from the saved analysis, then renders FindingChatDrawer. */ +function FindingChatDrawerWrapper({ analysisId, findingIndex, finding, @@ -128,7 +55,7 @@ function _FindingChatDrawerWrapper({ useEffect(() => { fetch(`/api/v1/compliance/history/${analysisId}`, { - headers: { Authorization: `Bearer ${localStorage.getItem('auth_token') ?? ''}` }, + headers: authHeader(), }) .then(r => r.json()) .then((data: { findings?: Array<{ seq: number; id: string }> }) => { @@ -153,8 +80,8 @@ export function CompliancePage() { const [showModal, setShowModal] = useState(false); const [showExportMenu, setShowExportMenu] = useState(false); const { state, run, reset } = useComplianceAnalysis(); - const chat = useFindingChat(); - const [drawerFindingIdx, setDrawerFindingIdx] = useState(null); + // drawerFinding holds {index, finding} for the currently-open FindingChatDrawer + const [drawerFinding, setDrawerFinding] = useState<{ idx: number; finding: FindingEvent } | null>(null); const { setComplianceState } = usePageState(); const { t } = useLanguage(); @@ -198,6 +125,8 @@ export function CompliancePage() { analysisId: data.id, isReadOnly: true, activeFindingId: null, + progress: null, + conflicts: [], }); } @@ -258,12 +187,6 @@ export function CompliancePage() { setShowExportMenu(false); } - // ── Chat context (finding desc + clause_ref as segment context) ────────── - const activeFinding = chat.findingIdx !== null ? state.findings[chat.findingIdx] : null; - const chatContext = activeFinding - ? `Finding: ${activeFinding.title}\n${activeFinding.desc}${activeFinding.clause_ref ? `\nRef: ${activeFinding.clause_ref}` : ''}` - : ''; - return (
Findings {state.findings.length > 0 && `(${state.findings.length})`} + {/* Real per-clause progress bar during streaming */} + {isStreaming && state.progress && state.progress.total > 0 && ( + + + + + {state.progress.done}/{state.progress.total} + + )}
{state.findings.length === 0 && isStreaming && ( @@ -472,30 +414,85 @@ export function CompliancePage() { {STATUS_LABEL[f.status] ?? f.status}

{f.desc}

+ + {/* Source refs: which retrieved chunks informed this finding */} + {f.source_refs && f.source_refs.length > 0 && ( +
+ {f.source_refs.map((sr, si) => ( + + 📄 {sr.standard ? sr.standard.slice(0, 20) : '—'} + {sr.score > 0 && ` · ${Math.round(sr.score * 100)}%`} + + ))} +
+ )} +
- {f.clause_ref && ( -
Ref: {f.clause_ref}
- )} - - {state.analysisId && ( +
+ {f.clause_ref && ( +
Ref: {f.clause_ref}
+ )} + {/* Confidence dot: green ≥0.7, amber 0.4–0.7, red <0.4 */} + {f.confidence !== undefined && ( + + = 0.7 ? '#22c55e' : f.confidence >= 0.4 ? '#f59e0b' : '#ef4444', + }} /> + {Math.round(f.confidence * 100)}% + + )} +
+ {/* Single consolidated chat button — only when analysis is saved */} + {state.analysisId ? ( + ) : ( + /* Fallback for unsaved analyses: show disabled chat hint */ + + {t.compliance.askAIBtn} + )}
))} + {/* Cross-clause conflicts panel */} + {state.conflicts && state.conflicts.length > 0 && ( +
+
+ + Cross-Clause Issues ({state.conflicts.length}) +
+ {state.conflicts.map((c, ci) => ( +
+ + [{c.type.replace('_', ' ')}] + + {' '}Finding #{c.finding_a}{c.finding_b ? ` ↔ #${c.finding_b}` : ''}: {c.desc} +
+ ))} +
+ )} + {/* Conclusion */} {isDone && state.done && (
@@ -540,92 +537,18 @@ export function CompliancePage() {
- {/* ── Finding Chat Side Panel ────────────────────────────────── */} - {chat.open && ( -
- {/* Header */} -
-
-
{t.compliance.chatSidebarHeader}
-
- Finding #{(chat.findingIdx ?? 0) + 1} · {activeFinding?.title} -
-
- -
- - {/* Messages */} -
- {chat.messages.map(msg => ( -
- {msg.role === 'assistant' && ( -
AI
- )} -
{msg.content}
-
- ))} - {chat.loading && ( -
-
AI
-
- {t.compliance.chatThinking} -
-
- )} -
- - {/* Quick questions */} -
- {[t.compliance.quickQ1, t.compliance.quickQ2, t.compliance.quickQ3].map(q => ( - - ))} -
- - {/* Input */} -
- chat.setInput(e.target.value)} - onKeyDown={e => { if (e.key === 'Enter' && !e.shiftKey) { e.preventDefault(); chat.send(chatContext); } }} - placeholder={t.compliance.chatPlaceholder} - style={{ flex: 1, padding: '9px 12px', fontSize: 13, background: 'var(--bg)', border: '1px solid var(--border)', borderRadius: 8, color: 'var(--fg)', outline: 'none' }} - /> - -
-
- )} - {drawerFindingIdx !== null && state.analysisId && ( - <_FindingChatDrawerWrapper + {/* ── Finding Chat Drawer (single consolidated UI) ───────────── */} + {drawerFinding !== null && state.analysisId && ( + setDrawerFindingIdx(null)} + onClose={() => setDrawerFinding(null)} /> )} diff --git a/frontend/src/pages/Compliance/useComplianceAnalysis.ts b/frontend/src/pages/Compliance/useComplianceAnalysis.ts index c9f77a5..4a0c110 100644 --- a/frontend/src/pages/Compliance/useComplianceAnalysis.ts +++ b/frontend/src/pages/Compliance/useComplianceAnalysis.ts @@ -14,9 +14,10 @@ import type { ComplianceSourceEvent, ComplianceFindingEvent, ComplianceDonePayload, + ComplianceConflict, } from '../../contexts'; -export type { ComplianceMeta, ComplianceState, ComplianceSourceEvent as SourceEvent, ComplianceFindingEvent as FindingEvent, ComplianceDonePayload as DonePayload }; +export type { ComplianceMeta, ComplianceState, ComplianceSourceEvent as SourceEvent, ComplianceFindingEvent as FindingEvent, ComplianceDonePayload as DonePayload, ComplianceConflict }; export type { ComplianceActionItem as ActionItem } from '../../contexts'; export type AnalysisStatus = import('../../contexts').ComplianceStatus; export type AnalysisMeta = ComplianceMeta; @@ -38,6 +39,8 @@ const INITIAL_STATE: ComplianceState = { errorText: '', analysisId: null, isReadOnly: false, + progress: null, + conflicts: [], }; export function useComplianceAnalysis() { @@ -92,6 +95,9 @@ export function useComplianceAnalysis() { if (j.type === 'stage') { setState(s => ({ ...s, stageLabel: j.label ?? '', stageKey: j.stage ?? '' })); + } else if (j.type === 'progress') { + // Real per-clause progress update from backend + setState(s => ({ ...s, progress: { done: j.done ?? 0, total: j.total ?? 0 } })); } else if (j.type === 'source') { const src: ComplianceSourceEvent = { standard: j.standard ?? '', @@ -99,6 +105,7 @@ export function useComplianceAnalysis() { score: j.score ?? 0, status: j.status ?? 'retrieved', full_content: j.full_content ?? '', + clause_index: j.clause_index, }; setState(s => ({ ...s, sources: [...s.sources, src] })); } else if (j.type === 'finding') { @@ -107,8 +114,13 @@ export function useComplianceAnalysis() { desc: j.desc ?? '', status: j.status ?? 'info', clause_ref: j.clause_ref, + confidence: j.confidence, + source_refs: j.source_refs, }; setState(s => ({ ...s, findings: [...s.findings, finding] })); + } else if (j.type === 'conflicts') { + // Cross-clause conflicts detected after all findings finish + setState(s => ({ ...s, conflicts: j.items ?? [] })); } else if (j.type === 'done') { const payload: ComplianceDonePayload = { conclusion: j.conclusion ?? '', diff --git a/frontend/src/pages/Docs/DocsPage.tsx b/frontend/src/pages/Docs/DocsPage.tsx index 45d75a7..d15ecce 100644 --- a/frontend/src/pages/Docs/DocsPage.tsx +++ b/frontend/src/pages/Docs/DocsPage.tsx @@ -20,6 +20,7 @@ interface Doc { sizeBytes: number; summary?: string; version?: string; + hasFile: boolean; } const STATUS_FILTERS = ['All', 'Ready', 'Processing', 'Failed', 'Pending']; @@ -102,6 +103,7 @@ export function DocsPage() { sizeBytes: (item.size_bytes as number) ?? 0, summary: item.summary as string | undefined, version: item.version as string | undefined, + hasFile: item.has_file !== false, }))); setLoading(false); }) @@ -130,11 +132,21 @@ export function DocsPage() { } // ── Download ───────────────────────────────────────────────────────────── - function downloadDoc(id: string, name: string) { - const a = document.createElement('a'); - a.href = `/api/v1/documents/download/${id}`; - a.download = name; - a.click(); + async function downloadDoc(id: string, name: string) { + try { + const resp = await fetch(`/api/v1/documents/download/${id}`, { headers: authHeader() }); + if (!resp.ok) throw new Error(`下载失败: ${resp.status}`); + const blob = await resp.blob(); + const url = URL.createObjectURL(blob); + const a = document.createElement('a'); + a.href = url; + a.download = name; + a.click(); + URL.revokeObjectURL(url); + } catch (err) { + console.error('Download failed', err); + alert(String(err)); + } } // ── Retry (re-process failed doc) ──────────────────────────────────────── @@ -289,11 +301,13 @@ export function DocsPage() { {formatSize(d.sizeBytes)} {d.type} - {/* Download */} + {/* Download — disabled for Milvus-only docs that have no binary file */} diff --git a/frontend/src/pages/Docs/UploadModal.tsx b/frontend/src/pages/Docs/UploadModal.tsx index 7d9159f..5ac7f96 100644 --- a/frontend/src/pages/Docs/UploadModal.tsx +++ b/frontend/src/pages/Docs/UploadModal.tsx @@ -222,7 +222,7 @@ export function UploadModal({ onClose, onComplete }: Props) { {/* ── Left panel: upload form ── */} -
+
Upload documents
Stage files for parsing and indexing.

PDF, DOCX, TXT — one per API call, processed sequentially.

@@ -254,7 +254,7 @@ export function UploadModal({ onClose, onComplete }: Props) {
{files.length > 0 && ( -
+
{files.map((f, i) => { const isDone = doneCount > i; const isActive = submitting && currentFileIdx === i; diff --git a/frontend/src/pages/RagChat/RagChatPage.tsx b/frontend/src/pages/RagChat/RagChatPage.tsx index afd2273..44ad8cb 100644 --- a/frontend/src/pages/RagChat/RagChatPage.tsx +++ b/frontend/src/pages/RagChat/RagChatPage.tsx @@ -1,9 +1,11 @@ import { useRef, useEffect, useCallback, useState } from 'react'; import { Topbar } from '../../components/layout/Topbar'; -import { Send, Download } from 'lucide-react'; +import { Send, Download, Zap, Paperclip, X, FileText, AlertCircle } from 'lucide-react'; import { usePageState } from '../../contexts'; import type { RagCitation } from '../../contexts'; import { useLanguage } from '../../contexts/LanguageContext'; +import { agenticChat } from '../../api/rag'; +import type { SSEMessage } from '../../api/index'; const TOKEN_KEY = 'auth_token'; function authHeader(): Record { @@ -11,6 +13,46 @@ function authHeader(): Record { return t ? { Authorization: `Bearer ${t}` } : {}; } +// ── Document context state ───────────────────────────────────────────────────── + +interface DocContext { + filename: string; + text: string; + charCount: number; + truncated: boolean; + /** 'extracting' while the backend is parsing; 'ready' when text is available; 'error' on failure */ + status: 'extracting' | 'ready' | 'error'; + errorMsg?: string; +} + +// ── Agentic-mode types ──────────────────────────────────────────────────────── + +interface ThinkingStep { + id: string; + step: string; + status: 'running' | 'done'; + intent_type?: string; + reason?: string; + requires_decomposition?: boolean; + sub_queries?: string[]; + query?: string; + index?: number; + total?: number; + found?: number; + sufficient?: boolean; + confidence?: number; + retry?: boolean; +} + +const STEP_ICONS: Record = { + intent_analysis: '🔍', + query_planning: '📋', + retrieving: '📚', + grounding_check: '🔗', +}; + +// ── Helpers ─────────────────────────────────────────────────────────────────── + // Map a raw source doc from the backend "retrieved" event to our Citation shape. function mapSource(s: Record, idx: number): RagCitation { const rawScore = typeof s.score === 'number' ? s.score : 0; @@ -69,10 +111,72 @@ export function RagChatPage() { const [streaming, setStreaming] = useState(ragStreamingRef.current); const [quickPrompts, setQuickPrompts] = useState(MOCK_QUICK); + // P0-1 Agentic mode state + const [agenticMode, setAgenticMode] = useState(false); + const [thinkingSteps, setThinkingSteps] = useState([]); + const [thinkingExpanded, setThinkingExpanded] = useState(true); + + // ── Document context state ───────────────────────────────────────────────── + // Holds the extracted text from the attached file; sent to the backend as + // conversation context on every message while it is set. + const [docContext, setDocContext] = useState(null); + const fileInputRef = useRef(null); + const bottomRef = useRef(null); const citRailRef = useRef(null); const citItemRefs = useRef>({}); + // ── Document context helpers ─────────────────────────────────────────────── + + /** Upload file to /rag/upload-context, extract its text, store as context. */ + async function handleFileAttach(file: File) { + setDocContext({ filename: file.name, text: '', charCount: 0, truncated: false, status: 'extracting' }); + + const fd = new FormData(); + fd.append('file', file); + + try { + const res = await fetch('/api/v1/rag/upload-context', { + method: 'POST', + headers: authHeader(), + body: fd, + }); + if (!res.ok) { + const errText = await res.text().catch(() => t.ragchat.attachErrorMsg); + setDocContext(prev => prev ? { ...prev, status: 'error', errorMsg: errText.slice(0, 120) } : null); + return; + } + const data = await res.json(); + setDocContext({ + filename: data.filename ?? file.name, + text: data.text ?? '', + charCount: data.char_count ?? 0, + truncated: data.truncated ?? false, + status: 'ready', + }); + } catch (err) { + setDocContext(prev => prev + ? { ...prev, status: 'error', errorMsg: String(err).slice(0, 120) } + : null + ); + } + } + + function handleFileInputChange(e: React.ChangeEvent) { + const file = e.target.files?.[0]; + if (file) void handleFileAttach(file); + // Reset so the same file can be re-selected + e.target.value = ''; + } + + function handleFileDrop(e: React.DragEvent) { + e.preventDefault(); + const file = Array.from(e.dataTransfer.files).find(f => + /\.(pdf|docx?|txt|md)$/i.test(f.name) + ); + if (file) void handleFileAttach(file); + } + // Fetch quick questions from backend on mount (only once per session) useEffect(() => { fetch('/api/v1/rag/quick-questions', { headers: authHeader() }) @@ -102,9 +206,17 @@ export function RagChatPage() { async function send(text?: string) { const q = (text ?? inputDraft).trim(); - if (!q || ragStreamingRef.current) return; + // Block send while a document is still being extracted + if (!q || ragStreamingRef.current || docContext?.status === 'extracting') return; + setRagState(s => ({ ...s, inputDraft: '' })); + // Show document context badge in user message bubble when active + const docPrefix = docContext?.status === 'ready' + ? `📄 ${docContext.filename}\n` + : ''; + const displayQuery = docPrefix + q; + const userMsgId = Date.now().toString(); const assistantId = (Date.now() + 1).toString(); @@ -112,7 +224,7 @@ export function RagChatPage() { ...s, messages: [ ...s.messages, - { id: userMsgId, role: 'user', text: q }, + { id: userMsgId, role: 'user', text: displayQuery }, { id: assistantId, role: 'assistant', text: '' }, ], citations: [], @@ -122,100 +234,211 @@ export function RagChatPage() { setStreaming(true); setHighlightedCit(null); + // P0-1: reset thinking panel for new query + if (agenticMode) { + setThinkingSteps([]); + setThinkingExpanded(true); + } + const ctrl = new AbortController(); ragAbortRef.current = ctrl; - try { - const body: Record = { query: q, top_k: 5 }; - if (sessionId) body.session_id = sessionId; - - const res = await fetch('/api/v1/rag/chat', { - method: 'POST', - headers: { 'Content-Type': 'application/json', ...authHeader() }, - body: JSON.stringify(body), - signal: ctrl.signal, - }); - - if (!res.body) throw new Error('No stream'); - const reader = res.body.getReader(); - const dec = new TextDecoder(); - let buffer = ''; + if (agenticMode) { + // ── Agentic path ──────────────────────────────────────────────────── const newCitations: RagCitation[] = []; - while (true) { - const { done, value } = await reader.read(); - if (done) break; - buffer += dec.decode(value, { stream: true }); + const handleMessage = (msg: SSEMessage) => { + if (msg.type === 'session') { + if (msg.session_id) setRagState(s => ({ ...s, sessionId: msg.session_id! })); - const blocks = buffer.split('\n\n'); - buffer = blocks.pop() ?? ''; - - for (const block of blocks) { - const dataLine = block.split('\n').find(l => l.startsWith('data: ')); - if (!dataLine) continue; - const raw = dataLine.slice(6).trim(); - if (!raw) continue; - try { - const j = JSON.parse(raw); - - if (j.type === 'session') { - if (j.session_id) setRagState(s => ({ ...s, sessionId: j.session_id })); - - } else if (j.type === 'retrieved' && Array.isArray(j.docs)) { - const mapped = j.docs.map((d: Record, i: number) => mapSource(d, i + 1)); - newCitations.push(...mapped); - setRagState(s => ({ ...s, citations: [...mapped] })); - - } else if (j.type === 'chunk' && j.text) { - setRagState(s => ({ - ...s, - messages: s.messages.map(msg => - msg.id === assistantId - ? { ...msg, text: msg.text + (j.text as string) } - : msg - ), - })); - - } else if (j.type === 'done') { - setRagState(s => ({ - ...s, - messages: s.messages.map(msg => { - if (msg.id !== assistantId) return msg; - const refs = [...new Set( - [...msg.text.matchAll(/\[(\d+)\]/g)].map(r => parseInt(r[1], 10)) - )].filter(n => n >= 1 && n <= newCitations.length); - return { ...msg, citationRefs: refs }; - }), - })); - break; - - } else if (j.type === 'error') { - setRagState(s => ({ - ...s, - messages: s.messages.map(msg => - msg.id === assistantId - ? { ...msg, text: `Error: ${j.text ?? 'Unknown error'}` } - : msg - ), - })); + } else if (msg.type === 'thinking') { + // Build a stable step id so we can upsert running→done transitions. + const stepId = `${msg.step}-${msg.retry ? 'retry' : (msg.index ?? 0)}`; + setThinkingSteps(prev => { + const idx = prev.findIndex(s => s.id === stepId); + const stepObj: ThinkingStep = { + id: stepId, + step: msg.step ?? '', + status: (msg.status as 'running' | 'done') ?? 'running', + intent_type: msg.intent_type, + reason: msg.reason, + sub_queries: msg.sub_queries, + query: msg.query, + index: msg.index, + total: msg.total, + found: msg.found, + sufficient: msg.sufficient, + confidence: msg.confidence, + retry: msg.retry, + }; + if (idx >= 0) { + const updated = [...prev]; + updated[idx] = stepObj; + return updated; } - } catch { /* malformed JSON chunk, skip */ } + return [...prev, stepObj]; + }); + + } else if (msg.type === 'retrieved' && Array.isArray(msg.docs)) { + const mapped = (msg.docs as unknown as Record[]).map((d, i) => mapSource(d, i + 1)); + newCitations.push(...mapped); + setRagState(s => ({ ...s, citations: [...mapped] })); + + } else if (msg.type === 'chunk' && msg.text) { + setRagState(s => ({ + ...s, + messages: s.messages.map(m => + m.id === assistantId ? { ...m, text: m.text + msg.text! } : m + ), + })); + + } else if (msg.type === 'done') { + setThinkingExpanded(false); + setRagState(s => ({ + ...s, + messages: s.messages.map(m => { + if (m.id !== assistantId) return m; + const refs = [...new Set( + [...m.text.matchAll(/\[(\d+)\]/g)].map(r => parseInt(r[1], 10)) + )].filter(n => n >= 1 && n <= newCitations.length); + return { ...m, citationRefs: refs }; + }), + })); + + } else if (msg.type === 'error') { + setRagState(s => ({ + ...s, + messages: s.messages.map(m => + m.id === assistantId ? { ...m, text: `Error: ${msg.text ?? 'Unknown error'}` } : m + ), + })); } + }; + + try { + await agenticChat( + q, 5, handleMessage, + (err) => { + setRagState(s => ({ + ...s, + messages: s.messages.map(m => + m.id === assistantId ? { ...m, text: t.ragchat.apiError } : m + ), + })); + console.error('agenticChat error:', err); + }, + undefined, + undefined, + sessionId ?? undefined, + ctrl.signal, + // Pass document context to agentic pipeline + docContext?.status === 'ready' ? docContext.text : undefined, + docContext?.status === 'ready' ? docContext.filename : undefined, + ); + } finally { + ragStreamingRef.current = false; + setStreaming(false); } - } catch (e: unknown) { - if (e instanceof Error && e.name !== 'AbortError') { - setRagState(s => ({ - ...s, - messages: s.messages.map(msg => - msg.id === assistantId - ? { ...msg, text: t.ragchat.apiError } - : msg - ), - })); + + } else { + // ── Standard RAG path (unchanged) ─────────────────────────────────── + try { + const body: Record = { query: q, top_k: 5 }; + if (sessionId) body.session_id = sessionId; + // Inject document text as conversation context when a file is attached + if (docContext?.status === 'ready') { + body.context_text = docContext.text; + body.context_filename = docContext.filename; + } + + const res = await fetch('/api/v1/rag/chat', { + method: 'POST', + headers: { 'Content-Type': 'application/json', ...authHeader() }, + body: JSON.stringify(body), + signal: ctrl.signal, + }); + + if (!res.body) throw new Error('No stream'); + const reader = res.body.getReader(); + const dec = new TextDecoder(); + let buffer = ''; + const newCitations: RagCitation[] = []; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + buffer += dec.decode(value, { stream: true }); + + const blocks = buffer.split('\n\n'); + buffer = blocks.pop() ?? ''; + + for (const block of blocks) { + const dataLine = block.split('\n').find(l => l.startsWith('data: ')); + if (!dataLine) continue; + const raw = dataLine.slice(6).trim(); + if (!raw) continue; + try { + const j = JSON.parse(raw); + + if (j.type === 'session') { + if (j.session_id) setRagState(s => ({ ...s, sessionId: j.session_id })); + + } else if (j.type === 'retrieved' && Array.isArray(j.docs)) { + const mapped = j.docs.map((d: Record, i: number) => mapSource(d, i + 1)); + newCitations.push(...mapped); + setRagState(s => ({ ...s, citations: [...mapped] })); + + } else if (j.type === 'chunk' && j.text) { + setRagState(s => ({ + ...s, + messages: s.messages.map(msg => + msg.id === assistantId + ? { ...msg, text: msg.text + (j.text as string) } + : msg + ), + })); + + } else if (j.type === 'done') { + setRagState(s => ({ + ...s, + messages: s.messages.map(msg => { + if (msg.id !== assistantId) return msg; + const refs = [...new Set( + [...msg.text.matchAll(/\[(\d+)\]/g)].map(r => parseInt(r[1], 10)) + )].filter(n => n >= 1 && n <= newCitations.length); + return { ...msg, citationRefs: refs }; + }), + })); + break; + + } else if (j.type === 'error') { + setRagState(s => ({ + ...s, + messages: s.messages.map(msg => + msg.id === assistantId + ? { ...msg, text: `Error: ${j.text ?? 'Unknown error'}` } + : msg + ), + })); + } + } catch { /* malformed JSON chunk, skip */ } + } + } + } catch (e: unknown) { + if (e instanceof Error && e.name !== 'AbortError') { + setRagState(s => ({ + ...s, + messages: s.messages.map(msg => + msg.id === assistantId + ? { ...msg, text: t.ragchat.apiError } + : msg + ), + })); + } + } finally { + ragStreamingRef.current = false; + setStreaming(false); } - } finally { - ragStreamingRef.current = false; - setStreaming(false); } } @@ -254,7 +477,126 @@ export function RagChatPage() { {/* ── Chat main ── */}
-
+ {/* P0-1: Agentic Thinking Panel — shown when agentic mode is active */} + {agenticMode && thinkingSteps.length > 0 && ( +
+ {/* Panel header — clickable to collapse/expand */} + + {/* Step list */} + {thinkingExpanded && ( +
+ {thinkingSteps.map(step => { + const stepLabels: Record = { + intent_analysis: t.ragchat.stepIntentAnalysis, + query_planning: t.ragchat.stepQueryPlanning, + retrieving: t.ragchat.stepRetrieving, + grounding_check: t.ragchat.stepGrounding, + }; + const intentLabels: Record = { + simple_qa: t.ragchat.intentSimpleQa, + compare: t.ragchat.intentCompare, + multi_hop: t.ragchat.intentMultiHop, + ambiguous: t.ragchat.intentAmbiguous, + }; + return ( +
+ + {step.status === 'running' + ? ⟳ + : (STEP_ICONS[step.step] ?? '·') + } + + + {stepLabels[step.step] ?? step.step} + {/* Intent analysis detail */} + {step.step === 'intent_analysis' && step.status === 'done' && step.intent_type && ( + + → {intentLabels[step.intent_type] ?? step.intent_type} + {step.requires_decomposition && ` · ${t.ragchat.intentNeedsDecomposition}`} + + )} + {/* Query planning detail */} + {step.step === 'query_planning' && step.status === 'done' && step.sub_queries && ( + + → {step.sub_queries.length} {t.ragchat.subQueriesCountSuffix} + + )} + {/* Retrieval detail */} + {step.step === 'retrieving' && ( + + {step.total && step.total > 1 && `[${step.index}/${step.total}] `} + {step.retry && t.ragchat.retryLabel} + {step.query && step.query.length > 50 + ? step.query.slice(0, 50) + '…' + : step.query} + {step.status === 'done' && step.found !== undefined && ( + 0 ? 'var(--success-fg, #16a34a)' : 'var(--warning, #ca8a04)' }}> + {' '}· {step.found} {t.ragchat.chunksFoundSuffix} + + )} + + )} + {/* Grounding check detail */} + {step.step === 'grounding_check' && step.status === 'done' && ( + + → {step.sufficient ? t.ragchat.groundingSufficient : t.ragchat.groundingInsufficient} + {step.confidence !== undefined && ` (${Math.round(step.confidence * 100)}%)`} + + )} + +
+ ); + })} +
+ )} +
+ )} + + {/* Messages area — accepts drag-and-drop document context attachment */} +
{ e.preventDefault(); e.dataTransfer.dropEffect = 'copy'; }} + onDrop={handleFileDrop} + > {messages.map(msg => (
{msg.role === 'assistant' &&
AI
} @@ -281,19 +623,130 @@ export function RagChatPage() { ))}
+ {/* P0-1: Agentic mode toggle */} +
+ + {agenticMode && ( + + {t.ragchat.agenticModeHint} + + )} +
+ + {/* ── Document context badge ── */} + {docContext && ( +
+ {docContext.status === 'extracting' && ( + ⟳ + )} + {docContext.status === 'ready' && } + {docContext.status === 'error' && } + + + {t.ragchat.attachContextBadge} + + + + {docContext.filename} + + + {docContext.status === 'ready' && ( + + {(docContext.charCount / 1000).toFixed(1)}k chars + {docContext.truncated ? ` · ${t.ragchat.attachTruncated}` : ''} + + )} + {docContext.status === 'extracting' && ( + + {t.ragchat.attachExtracting} + + )} + {docContext.status === 'error' && ( + + {t.ragchat.attachError} + + )} + + {/* Clear button */} + +
+ )} + + {/* Hidden file input */} + +
+ {/* Paperclip button — replaces attached doc when clicked again */} +