"""Khung hỏi-đáp trên đồ thị GraphRAG — R08-T14. Người dùng hỏi một câu về mã nguồn; agent trả lời dựa trên đồ thị vừa quét, rồi câu trả lời được gắn liên kết tới đúng file và làm nổi các node liên quan. ``_ask`` dài (119 dòng) vì nó là một lượt chạy hoàn chỉnh: dựng ngữ cảnh từ đồ thị, gọi provider ở luồng nền, nhận sự kiện phát dần, rồi dựng lại câu trả lời có liên kết. Cắt nhỏ ra thì phải chuyền qua lại chừng chục biến trạng thái, đọc còn khó hơn. Cùng kiểu mixin như shell và Co4E: các phương thức này đọc/ghi state của ``StructureGraphView`` (đồ thị đang hiển thị, thư mục giải nén tạm, panel agent). Xem ghi chú ở đầu ``presentation/shell/nav_rail.py``. """ from __future__ import annotations from .graph_scene import _Node # Import muộn trong hàm: structure_graph_view.py trộn chính mixin này vào lớp # của nó, nên import ở mức module là vòng. import re import sys from pathlib import Path from PySide6.QtCore import Qt, QUrl from ...core.worker import AgentWorker from ...i18n import tr from ...ui.osutil import open_folder, open_location from ...ui.widgets import CollapseStrip class GraphQaMixin: """Hỏi-đáp trên đồ thị. Trộn vào StructureGraphView.""" def _toggle_messages(self) -> None: """Kept for callers that still ask for a flip (e.g. keyboard paths).""" showing = self._stack.currentWidget() is self._msgs_view self.view_tabs.setCurrentIndex(0 if showing else 1) def _reload_messages(self) -> None: """Build the tree: day → conversation. Click a conversation to see its messages as JSON. Scoped to the current project (its history folder).""" from collections import OrderedDict from PySide6.QtCore import Qt from PySide6.QtWidgets import QTreeWidgetItem from ...core.history import list_conversations self._msgs_view.clear() pid = self._active_project_id or "" by_day: "OrderedDict[str, list]" = OrderedDict() try: convs = list_conversations(self.ctx.config.history_dir()) except Exception: # noqa: BLE001 convs = [] for conv in convs: if pid and conv.get("project_id", "default") != pid: continue day = (conv.get("created") or "")[:10] or "—" by_day.setdefault(day, []).append(conv) if not by_day: self._msgs_view.addTopLevelItem(QTreeWidgetItem([tr("structure.msgs_none")])) return for day in sorted(by_day, reverse=True): convs_d = by_day[day] day_item = QTreeWidgetItem([f"{day} ({len(convs_d)})"]) for conv in convs_d: it = QTreeWidgetItem([conv.get("title", "(untitled)")]) it.setData(0, Qt.UserRole, str(conv.get("path", ""))) day_item.addChild(it) self._msgs_view.addTopLevelItem(day_item) day_item.setExpanded(True) def _show_msg_json(self, item, _col: int = 0) -> None: import html import json from PySide6.QtCore import Qt from ...core.history import load_conversation path = item.data(0, Qt.UserRole) if not path: return try: conv = load_conversation(path) payload = {"title": conv.get("title", ""), "created": conv.get("created", ""), "kind": conv.get("kind", ""), "project_id": conv.get("project_id", ""), "messages": conv.get("messages", [])} text = json.dumps(payload, ensure_ascii=False, indent=2) except Exception as exc: # noqa: BLE001 text = f"(could not read: {exc})" self.detail.setHtml( f'
{html.escape(text)}
') def _preserve_answer(self) -> None: if self._detail_mode == "answer" and self._answer.strip(): self._render_answer() def _set_agent_collapsed(self, collapsed: bool) -> None: strip_w = CollapseStrip.WIDTH + 2 self._agent_panel.setVisible(not collapsed) self._agent_strip.setVisible(collapsed) if collapsed: self._agent_pane.setMaximumWidth(strip_w) sizes = self._split.sizes() if len(sizes) == 2: self._split.setSizes([max(1, sum(sizes) - strip_w), strip_w]) else: self._agent_pane.setMaximumWidth(16777215) self._split.setSizes([840, 320]) def _matched_sources(self, text: str): if self._graph is None or not text: return [] found: dict[str, tuple[str, str, str]] = {} for n in self._graph.nodes: if not n.path: continue label = n.label.rstrip("()") if len(label) < 3: continue if n.path not in found and re.search(rf"\b{re.escape(label)}\b", text): found[n.path] = (n.kind, n.label, n.detail or n.path) return sorted(found.items(), key=lambda kv: kv[1][1].lower())[:12] def _linkify_files(self, text: str, sources) -> str: """Turn file/entity NAMES mentioned in the answer into clickable links that open the file — so the user can click a name in the answer to view it.""" for path, (kind, label, rel) in sources: href = QUrl.fromLocalFile(path).toString(QUrl.ComponentFormattingOption.FullyEncoded) tokens = [] base = Path(path).name if base and len(base) >= 3: tokens.append(base) lab = (label or "").rstrip("()").strip() if lab and lab != base and len(lab) >= 3: tokens.append(lab) for tok in tokens: esc = re.escape(tok) # `tok` (code span) → keep the code style but make it a link text = re.sub(rf"`{esc}`", f"[`{tok}`]({href})", text) # bare tok, not already inside a link / path / code span text = re.sub(rf"(? None: text = self._answer sources = self._matched_sources(text) if sources: # 1) Make the file/entity names IN THE ANSWER clickable (open on click). text = self._linkify_files(text, sources) # 2) Append a clickable "Related sources" section listing each file. lines = [text, "", "---", f"**{tr('structure.related_sources')}**"] for path, (kind, label, rel) in sources: href = QUrl.fromLocalFile(path).toString(QUrl.ComponentFormattingOption.FullyEncoded) # kind badge for context (file/function/section/json_key) kind_badge = f" [{kind.upper()}]" if kind not in ("file",) else "" lines.append(f"- **[{label}⧉]({href})**{kind_badge} — `{rel}`") text = "\n".join(lines) self.detail.setMarkdown(text) def _on_detail_link(self, url: QUrl) -> None: if url.isLocalFile(): p = url.toLocalFile() # Open the FILE itself for viewing (fall back to its folder for a dir). if Path(p).is_file(): open_location(p) else: open_folder(p) def _ask(self) -> None: question = self.ask_edit.text().strip() if not question: return from ...core.skills import parse_skill_command skill_prefix, question, info = parse_skill_command(question) if info is not None: self.detail.setMarkdown(info) self._detail_mode = "answer" self.ask_edit.clear() return if self._graph is None: self.status_message.emit(tr("structure.scan_first")) return context = self._graph_context(self._graph) # Real file CONTENT to answer from (extracted temporarily in the worker): file_paths = self._candidate_file_paths() extract_cache = dict(self._extract_cache) extract_dir = str(self._extract_tmp_dir()) self._answer = "" self._detail_mode = "answer" self.detail.setPlainText("…") self.ask_edit.clear() active_project_id = self._active_project_id # Collect selected node context for auto-filtering selected_nodes = [item.data for item in self.scene.selectedItems() if isinstance(item, _Node)] selected_context = "" if selected_nodes: node_lines = [] for nd in selected_nodes: node_lines.append(f"- {nd.label} (kind: {nd.kind}, path: {getattr(nd, 'path', '')})") if nd.detail: node_lines.append(f" detail: {nd.detail}") # Also gather connected nodes connected_ids = set() for nd in selected_nodes: for edge in self._graph.edges: if edge.source == nd.id: connected_ids.add(edge.target) elif edge.target == nd.id: connected_ids.add(edge.source) connected_nodes = [n for n in self._graph.nodes if n.id in connected_ids] if connected_nodes: node_lines.append("\nConnected nodes:") for cn in connected_nodes: node_lines.append(f"- {cn.label} (kind: {cn.kind})") selected_context = "\n".join(node_lines) def job(worker: AgentWorker): provider = self.ctx.build_active_provider() system = ("You answer questions about a code/document knowledge graph. Use the provided " "graph context AND the extracted file contents to retrieve, synthesize and " "explain the answer. Be concise. Answer ONLY from what is provided (graph " "context + extracted contents) — never invent files, functions, or facts that " "aren't in it.\n\n" "EACH answer MUST include source citations so the user can verify where " "information came from. For every factual claim, file reference, or code " "element you mention, add a citation using this format:\n\n" " [source: filename.ext, line/section: XXX]\n\n" "Rules for citations:\n" " 1. Cite the EXACT file path from the graph context (use the path field).\n" " 2. For Python files: cite the function/class name and approximate line " " if available, or the module name.\n" " 3. For document files (.md, .txt): cite the section heading.\n" " 4. For JSON files: cite the key path (e.g. settings > database > host).\n" " 5. Place citations inline after the relevant sentence or fact.\n" " 6. At the end of your answer, add a '---' separator followed by a " " numbered **Sources cited:** section listing each unique source with " " its full path so the user can click to open it.\n\n" "Example citation format in text:\n" " The `process_data()` function handles CSV parsing " "[source: src/utils/parser.py, function: process_data].\n\n" "Example end-of-answer source list:\n" " ---\n" " **Sources cited:**\n" " 1. `src/utils/parser.py` — process_data function\n" " 2. `docs/api.md` — Section: Authentication\n") if skill_prefix: system += "\n\nFollow this skill:\n" + skill_prefix if active_project_id: from ...core.projects import load_project, project_context_text proj_ctx = project_context_text(load_project(active_project_id)) if proj_ctx: system += "\n\n" + proj_ctx user_content = f"Graph context:\n{context}" if selected_context: user_content += f"\n\nSelected node(s) context (focus your answer on these):\n{selected_context}" # Auto-extract the actual file contents (temporary) so the answer is # synthesized from real content, not just the graph structure. from .structure_graph_view import _extract_file_contents content_block, new_cache = _extract_file_contents(file_paths, extract_cache, extract_dir) if content_block: user_content += ("\n\nExtracted file contents (read these to answer about file " "details/data; cite the file path):\n" + content_block) user_content += f"\n\nQuestion: {question}" messages = [ {"role": "system", "content": system}, {"role": "user", "content": user_content}, ] from ...core import agent_roles, audit_log ok = True try: provider.chat(messages, on_text=lambda t: worker.emit_event({"type": "text", "delta": t}), cancel=worker.is_cancelled) except Exception: ok = False raise finally: audit_log.record("tool_call", "graphrag_ask", ok, question[:500], agent_role=agent_roles.KNOWLEDGE) return {"extracted": new_cache} w = AgentWorker(job) w.event.connect(self._on_ask_event) w.finished_ok.connect(self._on_ask_done) w.failed.connect(lambda e: self.detail.setPlainText(f"Error: {e}")) self._ask_worker = w w.start() def _on_ask_event(self, ev: dict) -> None: if ev.get("type") == "text": if self._answer == "": self.detail.clear() self._answer += ev.get("delta", "") self.detail.setPlainText(self._answer) def _on_ask_done(self, result: dict) -> None: # Keep the (temporary) extracted content so repeated questions reuse it # without re-extracting — dropped when leaving the tab (_clear_extracts). if isinstance(result, dict): self._extract_cache.update(result.get("extracted", {}) or {}) self._render_answer() def _candidate_file_paths(self) -> list: """File paths to read for a question: the SELECTED file nodes if any, else every file node in the graph (capped downstream).""" from pathlib import Path as _P if self._graph is None: return [] sel = [item.data for item in self.scene.selectedItems() if isinstance(item, _Node)] nodes = sel or list(self._graph.nodes) out, seen = [], set() for nd in nodes: p = (getattr(nd, "path", "") or "").strip() if p and p not in seen and _P(p).is_file(): seen.add(p) out.append(p) return out def _extract_tmp_dir(self): from pathlib import Path as _P if self._extract_dir is None: import tempfile from ...config import CONFIG_DIR base = CONFIG_DIR / "tmp" / "graphrag_extract" base.mkdir(parents=True, exist_ok=True) self._extract_dir = _P(tempfile.mkdtemp(dir=str(base))) return self._extract_dir def _clear_extracts(self) -> None: """Discard the temporary extracted content (on leaving the tab / switching project). The extraction is a scratch aid, never persisted.""" self._extract_cache = {} d, self._extract_dir = self._extract_dir, None if d is not None: import shutil shutil.rmtree(d, ignore_errors=True)