实现RAG参考文献显示模式

offline-backend-20260512/backend/app/gateway/routers/system_settings.py

@@ -49,6 +49,14 @@ class PrivateSquareUpdateRequest(BaseModel):
     password: str | None = Field(default=None, description="New password in plain text. Pass an empty string to clear.")
 
 
+class CitationDisplayResponse(BaseModel):
+    cleanup_words: list[str] = Field(default_factory=list, description="Global citation cleanup words")
+
+
+class CitationDisplayUpdateRequest(BaseModel):
+    cleanup_words: list[str] | None = None
+
+
 def _to_response(settings: PromptPrefixSettings) -> PromptPrefixResponse:
     return PromptPrefixResponse(
         enabled=settings.enabled,
@@ -83,6 +91,34 @@ async def update_prompt_prefix(body: PromptPrefixUpdateRequest, request: Request
     return _to_response(saved.prompt_prefix)
 
 
+@router.get("/citation-display", response_model=CitationDisplayResponse)
+async def get_citation_display() -> CitationDisplayResponse:
+    """Return global citation display cleanup settings."""
+    settings = load_system_settings().citation_display
+    return CitationDisplayResponse(cleanup_words=settings.cleanup_words)
+
+
+@router.put("/citation-display", response_model=CitationDisplayResponse)
+async def update_citation_display(
+    body: CitationDisplayUpdateRequest,
+    request: Request,
+) -> CitationDisplayResponse:
+    """Update global citation display cleanup settings. Admin only."""
+    await _require_admin(request)
+
+    current = load_system_settings()
+    words = [
+        str(item).strip()
+        for item in (body.cleanup_words or [])
+        if str(item).strip()
+    ]
+    updated = current.citation_display.model_copy(update={"cleanup_words": words})
+    saved = save_system_settings(
+        current.model_copy(update={"citation_display": updated})
+    )
+    return CitationDisplayResponse(cleanup_words=saved.citation_display.cleanup_words)
+
+
 @router.get("/private-square", response_model=PrivateSquareStatusResponse)
 async def get_private_square_status() -> PrivateSquareStatusResponse:
     """Return whether the private square is enabled and a password is set."""

offline-backend-20260512/backend/app/gateway/routers/threads.py

@@ -12,9 +12,12 @@ matching the LangGraph Platform wire format expected by the
 
 from __future__ import annotations
 
-import logging
-import uuid
-from typing import Any
+import logging
+import json
+import os
+from pathlib import Path
+import uuid
+from typing import Any
 
 from fastapi import APIRouter, HTTPException, Request
 from langgraph.checkpoint.base import empty_checkpoint
@@ -23,13 +26,18 @@ from pydantic import BaseModel, Field, field_validator
 from app.gateway.authz import require_permission
 from app.gateway.deps import get_checkpointer
 from app.gateway.utils import sanitize_log_param
-from deerflow.config.paths import Paths, get_paths
-from deerflow.runtime import serialize_channel_values
-from deerflow.runtime.user_context import get_effective_user_id
-from deerflow.utils.time import coerce_iso, now_iso
+from deerflow.config.paths import Paths, get_paths
+from deerflow.runtime import serialize_channel_values
+from deerflow.runtime.references import (
+    build_reference_batches as build_runtime_reference_batches,
+    build_reference_debug_summary,
+)
+from deerflow.runtime.user_context import get_effective_user_id
+from deerflow.utils.time import coerce_iso, now_iso
 
-logger = logging.getLogger(__name__)
-router = APIRouter(prefix="/api/threads", tags=["threads"])
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/api/threads", tags=["threads"])
+_RAG_DEBUG_LOGGER_NAME = "rag_citation_debug"
 
 
 # Metadata keys that the server controls; clients are not allowed to set
@@ -41,11 +49,38 @@ router = APIRouter(prefix="/api/threads", tags=["threads"])
 _SERVER_RESERVED_METADATA_KEYS: frozenset[str] = frozenset({"owner_id", "user_id"})
 
 
-def _strip_reserved_metadata(metadata: dict[str, Any] | None) -> dict[str, Any]:
-    """Return ``metadata`` with server-controlled keys removed."""
-    if not metadata:
-        return metadata or {}
-    return {k: v for k, v in metadata.items() if k not in _SERVER_RESERVED_METADATA_KEYS}
+def _strip_reserved_metadata(metadata: dict[str, Any] | None) -> dict[str, Any]:
+    """Return ``metadata`` with server-controlled keys removed."""
+    if not metadata:
+        return metadata or {}
+    return {k: v for k, v in metadata.items() if k not in _SERVER_RESERVED_METADATA_KEYS}
+
+
+def _get_rag_debug_logger() -> logging.Logger:
+    debug_logger = logging.getLogger(_RAG_DEBUG_LOGGER_NAME)
+    if getattr(debug_logger, "_rag_file_configured", False):
+        return debug_logger
+    log_path = Path(os.environ.get("RAG_CITATION_DEBUG_LOG", "logs/rag_citation_debug.log"))
+    if not log_path.is_absolute():
+        log_path = Path.cwd() / log_path
+    log_path.parent.mkdir(parents=True, exist_ok=True)
+    handler = logging.FileHandler(log_path, encoding="utf-8")
+    handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
+    debug_logger.addHandler(handler)
+    debug_logger.setLevel(logging.INFO)
+    debug_logger.propagate = False
+    setattr(debug_logger, "_rag_file_configured", True)
+    setattr(debug_logger, "_rag_file_path", str(log_path))
+    return debug_logger
+
+
+def _rag_debug_file_path() -> str:
+    return str(getattr(_get_rag_debug_logger(), "_rag_file_path", "logs/rag_citation_debug.log"))
+
+
+def _log_rag_debug(message: str, *args: Any) -> None:
+    logger.info(message, *args)
+    _get_rag_debug_logger().info(message, *args)
 
 
 # ---------------------------------------------------------------------------
@@ -132,11 +167,43 @@ class HistoryEntry(BaseModel):
     next: list[str] = Field(default_factory=list)
 
 
-class ThreadHistoryRequest(BaseModel):
+class ThreadHistoryRequest(BaseModel):
     """Request body for checkpoint history."""
 
     limit: int = Field(default=10, ge=1, le=100, description="Maximum entries")
-    before: str | None = Field(default=None, description="Cursor for pagination")
+    before: str | None = Field(default=None, description="Cursor for pagination")
+
+
+class ReferenceSourceResponse(BaseModel):
+    index: int
+    id: str | None = None
+    skillName: str = "backend-reference"
+    mode: str = "citation"
+    title: str
+    type: str = "result"
+    snippet: str = ""
+    source: str | None = None
+    time: str | None = None
+    url: str | None = None
+    content: str | None = None
+    score: float | None = None
+    raw: dict[str, Any] = Field(default_factory=dict)
+
+
+class ReferenceBatchResponse(BaseModel):
+    id: str
+    assistant_message_id: str
+    sources: list[ReferenceSourceResponse] = Field(default_factory=list)
+
+
+class ReferenceBatchesResponse(BaseModel):
+    batches: list[ReferenceBatchResponse] = Field(default_factory=list)
+
+
+class ReferenceDebugLogRequest(BaseModel):
+    event: str
+    thread_id: str | None = None
+    payload: dict[str, Any] = Field(default_factory=dict)
 
 
 # ---------------------------------------------------------------------------
@@ -163,7 +230,7 @@ def _delete_thread_data(thread_id: str, paths: Paths | None = None, *, user_id:
     return ThreadDeleteResponse(success=True, message=f"Deleted local thread data for {thread_id}")
 
 
-def _derive_thread_status(checkpoint_tuple) -> str:
+def _derive_thread_status(checkpoint_tuple) -> str:
     """Derive thread status from checkpoint metadata."""
     if checkpoint_tuple is None:
         return "idle"
@@ -179,7 +246,251 @@ def _derive_thread_status(checkpoint_tuple) -> str:
     if tasks:
         return "interrupted"
 
-    return "idle"
+    return "idle"
+
+
+def _message_type(message: Any) -> str:
+    if isinstance(message, dict):
+        return str(message.get("type") or message.get("role") or "")
+    return str(getattr(message, "type", "") or "")
+
+
+def _message_id(message: Any) -> str | None:
+    if isinstance(message, dict):
+        value = message.get("id")
+    else:
+        value = getattr(message, "id", None)
+    return str(value) if value else None
+
+
+def _message_content(message: Any) -> Any:
+    if isinstance(message, dict):
+        return message.get("content")
+    return getattr(message, "content", None)
+
+
+def _message_name(message: Any) -> str:
+    if isinstance(message, dict):
+        return str(message.get("name") or (message.get("additional_kwargs") or {}).get("name") or "tool")
+    return str(getattr(message, "name", None) or "tool")
+
+
+def _has_tool_calls(message: Any) -> bool:
+    if isinstance(message, dict):
+        direct = message.get("tool_calls")
+        chunks = message.get("tool_call_chunks")
+        extra = (message.get("additional_kwargs") or {}).get("tool_calls")
+    else:
+        direct = getattr(message, "tool_calls", None)
+        chunks = getattr(message, "tool_call_chunks", None)
+        extra = (getattr(message, "additional_kwargs", None) or {}).get("tool_calls")
+    return bool(direct or chunks or extra)
+
+
+def _has_answer_content(message: Any) -> bool:
+    content = _message_content(message)
+    if isinstance(content, str):
+        return bool(content.strip())
+    if isinstance(content, list):
+        for part in content:
+            if isinstance(part, dict) and part.get("type") == "text" and str(part.get("text") or "").strip():
+                return True
+    return False
+
+
+def _extract_tool_payload(message: Any) -> Any:
+    content = _message_content(message)
+    if isinstance(content, dict):
+        return _unwrap_tool_payload(content)
+    if isinstance(content, list):
+        texts: list[str] = []
+        for part in content:
+            if not isinstance(part, dict):
+                continue
+            if isinstance(part.get("json"), dict):
+                return _unwrap_tool_payload(part["json"])
+            if part.get("type") == "json" and isinstance(part.get("json"), dict):
+                return _unwrap_tool_payload(part["json"])
+            if part.get("type") == "text" and isinstance(part.get("text"), str):
+                texts.append(part["text"])
+            unwrapped = _unwrap_tool_payload(part)
+            if unwrapped is not None:
+                return unwrapped
+        content = "\n".join(texts)
+    if isinstance(content, str):
+        return _parse_json_like(content)
+    return None
+
+
+def _unwrap_tool_payload(value: dict[str, Any]) -> Any:
+    if any(isinstance(value.get(key), list) for key in ("results", "data", "items", "hits", "documents", "records", "sources")):
+        return value
+    for key in ("result", "output", "stdout", "content", "text", "payload", "response"):
+        nested = value.get(key)
+        if isinstance(nested, str):
+            parsed = _parse_json_like(nested)
+            if parsed is not None:
+                return parsed
+        if isinstance(nested, dict):
+            unwrapped = _unwrap_tool_payload(nested)
+            if unwrapped is not None:
+                return unwrapped
+    return value
+
+
+def _parse_json_like(text: str) -> Any:
+    trimmed = text.strip()
+    if not trimmed:
+        return None
+    try:
+        return json.loads(trimmed)
+    except Exception:
+        pass
+    first_obj = trimmed.find("{")
+    last_obj = trimmed.rfind("}")
+    if 0 <= first_obj < last_obj:
+        try:
+            return json.loads(trimmed[first_obj : last_obj + 1])
+        except Exception:
+            return None
+    return None
+
+
+def _first_list(payload: Any) -> list[Any]:
+    if isinstance(payload, list):
+        return payload
+    if not isinstance(payload, dict):
+        return []
+    for key in ("results", "data", "items", "hits", "documents", "records", "entries", "matches", "chunks", "sources"):
+        value = payload.get(key)
+        if isinstance(value, list):
+            return value
+    for key in ("result", "payload", "response", "output", "data"):
+        nested = payload.get(key)
+        found = _first_list(nested)
+        if found:
+            return found
+    return []
+
+
+def _nested_records(raw: Any) -> list[dict[str, Any]]:
+    if not isinstance(raw, dict):
+        return []
+    layers = [raw]
+    for key in ("metadata", "meta", "attrs", "attributes", "properties", "extra"):
+        nested = raw.get(key)
+        if isinstance(nested, dict):
+            layers.append(nested)
+    return layers
+
+
+def _pick(layers: list[dict[str, Any]], keys: tuple[str, ...]) -> str | None:
+    for layer in layers:
+        for key in keys:
+            value = layer.get(key)
+            if isinstance(value, (str, int, float)) and str(value).strip():
+                return str(value).strip()
+    return None
+
+
+def _normalize_reference_item(raw: Any, fallback_index: int) -> dict[str, Any] | None:
+    if not isinstance(raw, dict):
+        return None
+    layers = _nested_records(raw)
+    title = _pick(layers, ("title", "m_title", "name", "question", "keyword", "query", "text"))
+    content = _pick(layers, ("content", "page_content", "summary", "snippet", "description", "content_preview", "abstract", "body"))
+    url = _pick(layers, ("url", "link", "href", "source_url", "web_url"))
+    source = _pick(layers, ("source", "source1", "site", "provider", "from"))
+    time = _pick(layers, ("time", "date", "published_date", "publishedDate", "m_publish", "publish_time", "created_at"))
+    item_id = _pick(layers, ("recUuid", "rec_uuid", "uuid", "docId", "doc_id", "id", "recordId", "taskId"))
+    score_raw = _pick(layers, ("score", "relevance", "similarity"))
+    score = None
+    if score_raw is not None:
+        try:
+            score = float(score_raw)
+        except Exception:
+            score = None
+    if not title:
+        title = source or f"结果 {fallback_index}"
+    if not content:
+        content = ""
+    if not title and not content and not url:
+        return None
+    return {
+        "id": item_id or url,
+        "title": title,
+        "type": "result",
+        "snippet": content,
+        "content": content,
+        "source": source,
+        "time": time,
+        "url": url,
+        "score": score,
+        "raw": raw,
+    }
+
+
+def _reference_key(source: dict[str, Any]) -> str:
+    raw = source.get("raw") if isinstance(source.get("raw"), dict) else {}
+    item_id = (
+        source.get("id")
+        or source.get("url")
+        or raw.get("recUuid")
+        or raw.get("uuid")
+        or raw.get("docId")
+        or raw.get("id")
+    )
+    if item_id:
+        return str(item_id).casefold().strip()
+    return "|".join(
+        str(source.get(key) or "").casefold().strip()
+        for key in ("title", "source", "time", "content", "snippet")
+    )
+
+
+def _build_reference_batches(messages: list[Any]) -> list[ReferenceBatchResponse]:
+    batches: list[ReferenceBatchResponse] = []
+    for index, message in enumerate(messages):
+        if _message_type(message) != "human":
+            continue
+        end = len(messages)
+        for cursor in range(index + 1, len(messages)):
+            if _message_type(messages[cursor]) == "human":
+                end = cursor
+                break
+        assistant_message: Any | None = None
+        for cursor in range(end - 1, index, -1):
+            candidate = messages[cursor]
+            if (
+                _message_type(candidate) == "ai"
+                and _message_id(candidate)
+                and _has_answer_content(candidate)
+                and not _has_tool_calls(candidate)
+            ):
+                assistant_message = candidate
+                break
+        if assistant_message is None:
+            continue
+        seen: set[str] = set()
+        sources: list[ReferenceSourceResponse] = []
+        for tool_message in messages[index + 1 : end]:
+            if _message_type(tool_message) != "tool":
+                continue
+            payload = _extract_tool_payload(tool_message)
+            for raw_item in _first_list(payload):
+                normalized = _normalize_reference_item(raw_item, len(sources) + 1)
+                if not normalized:
+                    continue
+                key = _reference_key(normalized)
+                if key and key in seen:
+                    continue
+                if key:
+                    seen.add(key)
+                sources.append(ReferenceSourceResponse(index=len(sources) + 1, **normalized))
+        if sources:
+            mid = _message_id(assistant_message) or str(index)
+            batches.append(ReferenceBatchResponse(id=mid, assistant_message_id=mid, sources=sources))
+    return batches
 
 
 # ---------------------------------------------------------------------------
@@ -406,9 +717,9 @@ async def get_thread(thread_id: str, request: Request) -> ThreadResponse:
 
 
 # ---------------------------------------------------------------------------
-@router.get("/{thread_id}/state", response_model=ThreadStateResponse)
-@require_permission("threads", "read", owner_check=True)
-async def get_thread_state(thread_id: str, request: Request) -> ThreadStateResponse:
+@router.get("/{thread_id}/state", response_model=ThreadStateResponse)
+@require_permission("threads", "read", owner_check=True)
+async def get_thread_state(thread_id: str, request: Request) -> ThreadStateResponse:
     """Get the latest state snapshot for a thread.
 
     Channel values are serialized to ensure LangChain message objects
@@ -455,12 +766,84 @@ async def get_thread_state(thread_id: str, request: Request) -> ThreadStateRespo
         parent_checkpoint_id=parent_checkpoint_id,
         created_at=coerce_iso(metadata.get("created_at", "")),
         tasks=tasks,
-    )
-
-
-@router.post("/{thread_id}/state", response_model=ThreadStateResponse)
-@require_permission("threads", "write", owner_check=True, require_existing=True)
-async def update_thread_state(thread_id: str, body: ThreadStateUpdateRequest, request: Request) -> ThreadStateResponse:
+    )
+
+
+@router.get("/{thread_id}/reference-batches", response_model=ReferenceBatchesResponse)
+@require_permission("threads", "read", owner_check=True)
+async def get_thread_reference_batches(thread_id: str, request: Request) -> ReferenceBatchesResponse:
+    """Return backend-built citation batches for completed assistant turns."""
+    checkpointer = get_checkpointer(request)
+    config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
+    try:
+        checkpoint_tuple = await checkpointer.aget_tuple(config)
+    except Exception:
+        logger.exception("Failed to get reference batches for thread %s", sanitize_log_param(thread_id))
+        raise HTTPException(status_code=500, detail="Failed to get reference batches")
+
+    if checkpoint_tuple is None:
+        raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found")
+
+    checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {}
+    values = serialize_channel_values(checkpoint.get("channel_values", {}) or {})
+    messages = values.get("messages")
+    if not isinstance(messages, list):
+        _log_rag_debug(
+            "backend reference-batches thread=%s messages=missing batches=0 log_file=%s",
+            sanitize_log_param(thread_id),
+            _rag_debug_file_path(),
+        )
+        return ReferenceBatchesResponse(batches=[])
+    raw_batches = build_runtime_reference_batches(messages)
+    debug_summary = build_reference_debug_summary(messages, raw_batches)
+    _log_rag_debug(
+        "backend reference-batches thread=%s messages=%s turns=%s tool_messages=%s raw_items=%s "
+        "normalized_items=%s kept_items=%s batches=%s batch_sources=%s batch_ids=%s turns_detail=%s log_file=%s",
+        sanitize_log_param(thread_id),
+        debug_summary["messages"],
+        debug_summary["turns"],
+        debug_summary["tool_messages"],
+        debug_summary["raw_items"],
+        debug_summary["normalized_items"],
+        debug_summary["kept_items"],
+        len(raw_batches),
+        ",".join(str(batch.get("source_count", 0)) for batch in debug_summary["batches"]),
+        ",".join(str(batch.get("id") or "") for batch in debug_summary["batches"]),
+        json.dumps(debug_summary["turns_detail"], ensure_ascii=False),
+        _rag_debug_file_path(),
+    )
+    return ReferenceBatchesResponse(
+        batches=[
+            ReferenceBatchResponse(
+                id=str(batch.get("id") or ""),
+                assistant_message_id=str(batch.get("assistant_message_id") or batch.get("id") or ""),
+                sources=[
+                    ReferenceSourceResponse(**source)
+                    for source in batch.get("sources", [])
+                    if isinstance(source, dict)
+                ],
+            )
+            for batch in raw_batches
+        ]
+    )
+
+
+@router.post("/reference-debug-log")
+async def write_reference_debug_log(body: ReferenceDebugLogRequest) -> dict[str, str | bool]:
+    """Append frontend citation diagnostics to the backend RAG debug log."""
+    _get_rag_debug_logger().info(
+        "frontend %s thread=%s payload=%s log_file=%s",
+        body.event,
+        sanitize_log_param(body.thread_id or ""),
+        json.dumps(body.payload, ensure_ascii=False, default=str),
+        _rag_debug_file_path(),
+    )
+    return {"ok": True, "log_file": _rag_debug_file_path()}
+
+
+@router.post("/{thread_id}/state", response_model=ThreadStateResponse)
+@require_permission("threads", "write", owner_check=True, require_existing=True)
+async def update_thread_state(thread_id: str, body: ThreadStateUpdateRequest, request: Request) -> ThreadStateResponse:
     """Update thread state (e.g. for human-in-the-loop resume or title rename).
 
     Writes a new checkpoint that merges *body.values* into the latest

offline-backend-20260512/backend/app/gateway/services.py

@@ -100,7 +100,14 @@ _DEFAULT_ASSISTANT_ID = "lead_agent"
 
 # Built-in LangGraph graph IDs registered in langgraph.json.
 # These are NOT stored in the agents DB and must bypass the custom-agent access check.
-_BUILTIN_GRAPH_IDS: frozenset[str] = frozenset({"lead_agent", "canvas_agent", "ai_writing"})
+_BUILTIN_GRAPH_IDS: frozenset[str] = frozenset({"lead_agent", "canvas_agent", "ai_writing"})
+
+_RAG_CITATION_PROMPT = """当前对话已开启参考文献模式。
+如果本轮调用了搜索、知识库、检索类技能或工具，请基于本轮所有检索结果整体回答。
+你必须把同一轮内多次检索得到的资料视为同一批参考文献，按出现顺序从 [1] 开始连续编号，不要让每个检索词各自从 [1] 重新编号。
+回答正文中的关键结论必须标注对应编号，例如 [1]、[2]，多个来源可写为 [1][3][5]。
+不要引用不存在的编号，不要使用技能内部自带的编号，不要把参考文献完整列表堆在正文末尾。
+如果没有任何可用检索结果，请直接说明未检索到可靠参考资料。"""
 
 
 def _prepend_prompt_prefix(graph_input: dict[str, Any], prefix: str) -> dict[str, Any]:
@@ -129,7 +136,7 @@ def _prepend_prompt_prefix(graph_input: dict[str, Any], prefix: str) -> dict[str
     return {**graph_input, "messages": new_messages}
 
 
-def _prompt_prefix_opted_in(body: Any) -> bool:
+def _prompt_prefix_opted_in(body: Any) -> bool:
     """Detect whether the caller asked for the prompt prefix to be applied."""
     context = getattr(body, "context", None)
     if isinstance(context, Mapping) and context.get("prompt_prefix_enabled"):
@@ -137,7 +144,44 @@ def _prompt_prefix_opted_in(body: Any) -> bool:
     metadata = getattr(body, "metadata", None)
     if isinstance(metadata, Mapping) and metadata.get("prompt_prefix_enabled"):
         return True
-    return False
+    return False
+
+
+def _rag_mode_requested(body: Any) -> bool:
+    context = getattr(body, "context", None)
+    if isinstance(context, Mapping) and context.get("rag_mode_enabled"):
+        return True
+    metadata = getattr(body, "metadata", None)
+    if isinstance(metadata, Mapping) and (
+        metadata.get("rag_mode_enabled") or metadata.get("rag_mode")
+    ):
+        return True
+    return False
+
+
+def _thread_rag_mode_enabled(record: Any) -> bool:
+    if not isinstance(record, Mapping):
+        return False
+    metadata = record.get("metadata")
+    return isinstance(metadata, Mapping) and bool(
+        metadata.get("rag_mode_enabled") or metadata.get("rag_mode")
+    )
+
+
+def _append_invisible_prompt(graph_input: dict[str, Any], prompt: str) -> dict[str, Any]:
+    messages = graph_input.get("messages")
+    if not isinstance(messages, list) or not messages:
+        return graph_input
+    new_messages = list(messages)
+    for idx, msg in enumerate(new_messages):
+        if not isinstance(msg, HumanMessage):
+            continue
+        extra = dict(getattr(msg, "additional_kwargs", None) or {})
+        existing = str(extra.get("prompt_prefix") or "").strip()
+        extra["prompt_prefix"] = f"{existing}\n\n{prompt}".strip() if existing else prompt
+        new_messages[idx] = msg.model_copy(update={"additional_kwargs": extra})
+        break
+    return {**graph_input, "messages": new_messages}
 
 
 # Whitelist of run-context keys that the langgraph-compat layer forwards from
@@ -399,16 +443,29 @@ async def start_run(
     # Upsert thread metadata so the thread appears in /threads/search,
     # even for threads that were never explicitly created via POST /threads
     # (e.g. stateless runs).
-    try:
-        existing = await run_ctx.thread_store.get(thread_id)
-        if existing is None:
-            await run_ctx.thread_store.create(
-                thread_id,
-                assistant_id=body.assistant_id,
-                metadata=body.metadata,
-            )
-        else:
-            await run_ctx.thread_store.update_status(thread_id, "running")
+    requested_rag_mode = _rag_mode_requested(body)
+    effective_rag_mode = requested_rag_mode
+    try:
+        existing = await run_ctx.thread_store.get(thread_id)
+        effective_rag_mode = requested_rag_mode or _thread_rag_mode_enabled(existing)
+        if existing is None:
+            metadata = dict(body.metadata or {})
+            if effective_rag_mode:
+                metadata["rag_mode_enabled"] = True
+                metadata["rag_mode"] = True
+            await run_ctx.thread_store.create(
+                thread_id,
+                assistant_id=body.assistant_id,
+                metadata=metadata,
+            )
+        else:
+            await run_ctx.thread_store.update_status(thread_id, "running")
+            if requested_rag_mode and not _thread_rag_mode_enabled(existing):
+                await run_ctx.thread_store.update_metadata(
+                    thread_id,
+                    {"rag_mode_enabled": True, "rag_mode": True},
+                )
+                effective_rag_mode = True
             # SDK 的 ``client.threads.create()`` 不传 assistant_id —— 先建出来空的 thread，
             # 真正提交 run 时才知道用哪个 assistant。这里回填一次，避免历史列表 / 按
             # assistant 筛选时拿不到关联。
@@ -428,17 +485,21 @@ async def start_run(
     # applies once. We intentionally do NOT gate on thread-row existence: the
     # thread_meta row is frequently pre-created, which previously skipped the
     # prefix for genuine first turns.
-    if _prompt_prefix_opted_in(body):
-        prefix_settings = load_system_settings().prompt_prefix
-        if prefix_settings.enabled and prefix_settings.prompt_prefix.strip():
-            graph_input = _prepend_prompt_prefix(graph_input, prefix_settings.prompt_prefix.strip())
-    config = build_run_config(thread_id, body.config, body.metadata, assistant_id=body.assistant_id)
+    if _prompt_prefix_opted_in(body):
+        prefix_settings = load_system_settings().prompt_prefix
+        if prefix_settings.enabled and prefix_settings.prompt_prefix.strip():
+            graph_input = _prepend_prompt_prefix(graph_input, prefix_settings.prompt_prefix.strip())
+    if effective_rag_mode:
+        graph_input = _append_invisible_prompt(graph_input, _RAG_CITATION_PROMPT)
+    config = build_run_config(thread_id, body.config, body.metadata, assistant_id=body.assistant_id)
 
     # Merge DeerFlow-specific context overrides into both ``configurable`` and ``context``.
     # The ``context`` field is a custom extension for the langgraph-compat layer
     # that carries agent configuration (model_name, thinking_enabled, etc.).
     # Only agent-relevant keys are forwarded; unknown keys (e.g. thread_id) are ignored.
-    body_context = dict(getattr(body, "context", None) or {})
+    body_context = dict(getattr(body, "context", None) or {})
+    if effective_rag_mode:
+        body_context["rag_mode_enabled"] = True
     # Promote whitelisted top-level body fields into the context map so
     # callers can pass them either nested under ``context`` or as flat
     # top-level fields (``excluded_tools``, ``skill_stop_names``, …).

offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py

@@ -7,9 +7,10 @@ from langchain_core.runnables import RunnableConfig
 from deerflow.agents.lead_agent.prompt import apply_prompt_template
 from deerflow.agents.middlewares.clarification_middleware import ClarificationMiddleware
 from deerflow.agents.middlewares.loop_detection_middleware import LoopDetectionMiddleware
-from deerflow.agents.middlewares.memory_middleware import MemoryMiddleware
-from deerflow.agents.middlewares.prompt_prefix_middleware import PromptPrefixMiddleware
-from deerflow.agents.middlewares.setup_agent_approval_middleware import SetupAgentApprovalMiddleware
+from deerflow.agents.middlewares.memory_middleware import MemoryMiddleware
+from deerflow.agents.middlewares.prompt_prefix_middleware import PromptPrefixMiddleware
+from deerflow.agents.middlewares.rag_citation_middleware import RagCitationMiddleware
+from deerflow.agents.middlewares.setup_agent_approval_middleware import SetupAgentApprovalMiddleware
 from deerflow.agents.middlewares.subagent_limit_middleware import SubagentLimitMiddleware
 from deerflow.agents.middlewares.summarization_middleware import BeforeSummarizationHook, DeerFlowSummarizationMiddleware
 from deerflow.agents.middlewares.title_middleware import TitleMiddleware
@@ -252,13 +253,16 @@ def _build_middlewares(
 
     Returns:
         List of middleware instances.
-    """
-    resolved_app_config = app_config or get_app_config()
-    middlewares = build_lead_runtime_middlewares(app_config=resolved_app_config, lazy_init=True)
+    """
+    resolved_app_config = app_config or get_app_config()
+    cfg = _get_runtime_config(config)
+    middlewares = build_lead_runtime_middlewares(app_config=resolved_app_config, lazy_init=True)
 
-    # Inject the admin prompt prefix into the model request (kept out of the
-    # persisted/displayed message so the user never sees it).
-    middlewares.append(PromptPrefixMiddleware())
+    # Inject the admin prompt prefix into the model request (kept out of the
+    # persisted/displayed message so the user never sees it).
+    middlewares.append(PromptPrefixMiddleware())
+    if cfg.get("rag_mode_enabled", False):
+        middlewares.append(RagCitationMiddleware())
 
     # Add summarization middleware if enabled
     summarization_middleware = _create_summarization_middleware(app_config=resolved_app_config)
@@ -266,8 +270,7 @@ def _build_middlewares(
         middlewares.append(summarization_middleware)
 
     # Add TodoList middleware if plan mode is enabled
-    cfg = _get_runtime_config(config)
-    is_plan_mode = cfg.get("is_plan_mode", False)
+    is_plan_mode = cfg.get("is_plan_mode", False)
     is_bootstrap = cfg.get("is_bootstrap", False)
     todo_list_middleware = _create_todo_list_middleware(is_plan_mode)
     if todo_list_middleware is not None:

offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/rag_citation_middleware.py

@@ -0,0 +1,109 @@
+"""Inject stable citation numbering before RAG final answers."""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+from collections.abc import Awaitable, Callable
+from typing import Any, override
+
+from langchain.agents import AgentState
+from langchain.agents.middleware import AgentMiddleware
+from langchain.agents.middleware.types import ModelCallResult, ModelRequest, ModelResponse
+from langchain_core.messages import SystemMessage
+
+from deerflow.runtime.references import build_latest_reference_prompt, build_reference_debug_summary
+
+logger = logging.getLogger(__name__)
+_RAG_DEBUG_LOGGER_NAME = "rag_citation_debug"
+
+
+def _get_rag_debug_logger() -> logging.Logger:
+    debug_logger = logging.getLogger(_RAG_DEBUG_LOGGER_NAME)
+    if getattr(debug_logger, "_rag_file_configured", False):
+        return debug_logger
+    log_path = Path(os.environ.get("RAG_CITATION_DEBUG_LOG", "logs/rag_citation_debug.log"))
+    if not log_path.is_absolute():
+        log_path = Path.cwd() / log_path
+    log_path.parent.mkdir(parents=True, exist_ok=True)
+    handler = logging.FileHandler(log_path, encoding="utf-8")
+    handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
+    debug_logger.addHandler(handler)
+    debug_logger.setLevel(logging.INFO)
+    debug_logger.propagate = False
+    setattr(debug_logger, "_rag_file_configured", True)
+    setattr(debug_logger, "_rag_file_path", str(log_path))
+    return debug_logger
+
+
+def _log_rag_debug(message: str, *args: Any) -> None:
+    logger.info(message, *args)
+    _get_rag_debug_logger().info(message, *args)
+
+
+class RagCitationMiddleware(AgentMiddleware[AgentState]):
+    """Give the model the same numbered references the frontend will render."""
+
+    @staticmethod
+    def _enabled(request: ModelRequest) -> bool:
+        runtime = getattr(request, "runtime", None)
+        context = getattr(runtime, "context", None) if runtime is not None else None
+        if isinstance(context, dict) and context.get("rag_mode_enabled"):
+            return True
+        config = getattr(runtime, "config", None) if runtime is not None else None
+        if isinstance(config, dict):
+            configurable = config.get("configurable")
+            if isinstance(configurable, dict) and configurable.get("rag_mode_enabled"):
+                return True
+            runtime_context = config.get("context")
+            if isinstance(runtime_context, dict) and runtime_context.get("rag_mode_enabled"):
+                return True
+        return False
+
+    @staticmethod
+    def _patch(request: ModelRequest) -> ModelRequest | None:
+        if not RagCitationMiddleware._enabled(request):
+            return None
+        messages = list(request.messages)
+        prompt, batch = build_latest_reference_prompt(messages)
+        if not prompt:
+            _log_rag_debug(
+                "middleware reference-prompt enabled=true prompt=false messages=%s",
+                len(messages),
+            )
+            return None
+        summary = build_reference_debug_summary(messages, [batch] if batch else None)
+        _log_rag_debug(
+            "middleware reference-prompt enabled=true prompt=true messages=%s tool_messages=%s "
+            "raw_items=%s normalized_items=%s kept_items=%s latest_batch_sources=%s",
+            summary["messages"],
+            summary["tool_messages"],
+            summary["raw_items"],
+            summary["normalized_items"],
+            summary["kept_items"],
+            len((batch or {}).get("sources") or []),
+        )
+        patched_messages = [
+            *messages,
+            SystemMessage(content=prompt),
+        ]
+        return request.override(messages=patched_messages)
+
+    @override
+    def wrap_model_call(
+        self,
+        request: ModelRequest,
+        handler: Callable[[ModelRequest], ModelResponse],
+    ) -> ModelCallResult:
+        patched = self._patch(request)
+        return handler(patched or request)
+
+    @override
+    async def awrap_model_call(
+        self,
+        request: ModelRequest,
+        handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
+    ) -> ModelCallResult:
+        patched = self._patch(request)
+        return await handler(patched or request)

offline-backend-20260512/backend/packages/harness/deerflow/config/system_settings.py

@@ -54,6 +54,16 @@ class AppearanceSettings(BaseModel):
     model_config = ConfigDict(extra="ignore")
 
 
+class CitationDisplaySettings(BaseModel):
+    """System-wide cleanup rules for citation cards and detail dialogs."""
+
+    cleanup_words: list[str] = Field(
+        default_factory=list,
+        description="Words or phrases used to remove noisy advertisement/disclaimer lines from citation display.",
+    )
+    model_config = ConfigDict(extra="ignore")
+
+
 class PrivateSquareSettings(BaseModel):
     """Admin-configurable settings for the agent private-square (私密广场).
 
@@ -74,6 +84,7 @@ class SystemSettings(BaseModel):
     prompt_prefix: PromptPrefixSettings = Field(default_factory=PromptPrefixSettings)
     private_square: PrivateSquareSettings = Field(default_factory=PrivateSquareSettings)
     appearance: AppearanceSettings = Field(default_factory=AppearanceSettings)
+    citation_display: CitationDisplaySettings = Field(default_factory=CitationDisplaySettings)
     model_config = ConfigDict(extra="ignore")
 
 

offline-backend-20260512/backend/packages/harness/deerflow/runtime/references.py

@@ -0,0 +1,366 @@
+"""Build stable per-turn citation batches from LangGraph messages."""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+
+def build_reference_batches(messages: list[Any]) -> list[dict[str, Any]]:
+    batches: list[dict[str, Any]] = []
+    for index, message in enumerate(messages):
+        if _message_type(message) != "human":
+            continue
+        end = len(messages)
+        for cursor in range(index + 1, len(messages)):
+            if _message_type(messages[cursor]) == "human":
+                end = cursor
+                break
+        assistant = _find_last_assistant_answer(messages, index + 1, end)
+        if assistant is None:
+            continue
+        sources = _collect_sources(messages[index + 1 : end])
+        if sources:
+            message_id = _message_id(assistant) or str(index)
+            batches.append(
+                {
+                    "id": message_id,
+                    "assistant_message_id": message_id,
+                    "sources": sources,
+                }
+            )
+    return batches
+
+
+def build_latest_reference_prompt(messages: list[Any]) -> tuple[str, dict[str, Any] | None]:
+    human_index = _find_latest_human_index(messages)
+    if human_index is None:
+        return "", None
+    sources = _collect_sources(messages[human_index + 1 :])
+    if not sources:
+        return "", None
+    assistant = _find_last_assistant_answer(messages, human_index + 1, len(messages))
+    message_id = _message_id(assistant) or _message_id(messages[human_index]) or str(human_index)
+    batch = {
+        "id": message_id,
+        "assistant_message_id": message_id,
+        "sources": sources,
+    }
+    lines = [
+        "本轮检索结果已由系统统一合并并重新编号。最终回答必须使用下面这套编号。",
+        "如果你仍需要补充检索，请先继续调用工具；最终回答只能引用最新合并后的编号。",
+        "",
+    ]
+    for source in sources:
+        title = source.get("title") or f"参考文献 {source.get('index')}"
+        meta = " · ".join(
+            str(item)
+            for item in (source.get("source"), source.get("time"))
+            if item
+        )
+        content = str(source.get("content") or source.get("snippet") or "").strip()
+        if len(content) > 520:
+            content = content[:517] + "..."
+        head = f"[{source.get('index')}] {title}"
+        if meta:
+            head += f"（{meta}）"
+        lines.append(head)
+        if content:
+            lines.append(content)
+        lines.append("")
+    lines.extend(
+        [
+            "引用规则：",
+            "1. 回答正文中的关键结论必须标注对应编号，例如 [1]、[2]。",
+            "2. 多个来源可写为 [1][3][5]。",
+            "3. 只能引用上方存在的编号，不要编造编号。",
+            "4. 不要使用工具原始编号，也不要重新排序编号。",
+            "5. 不要在正文末尾堆完整参考文献列表，前端会展示右侧参考文献。",
+        ]
+    )
+    return "\n".join(lines).strip(), batch
+
+
+def _find_latest_human_index(messages: list[Any]) -> int | None:
+    for index in range(len(messages) - 1, -1, -1):
+        if _message_type(messages[index]) == "human":
+            return index
+    return None
+
+
+def build_reference_debug_summary(messages: list[Any], batches: list[dict[str, Any]] | None = None) -> dict[str, Any]:
+    """Return compact counts for citation extraction logs."""
+    turn_summaries: list[dict[str, Any]] = []
+    total_tool_messages = 0
+    total_raw_items = 0
+    total_normalized_items = 0
+    total_kept_items = 0
+    for index, message in enumerate(messages):
+        if _message_type(message) != "human":
+            continue
+        end = len(messages)
+        for cursor in range(index + 1, len(messages)):
+            if _message_type(messages[cursor]) == "human":
+                end = cursor
+                break
+        tool_messages = 0
+        raw_items = 0
+        normalized_items = 0
+        kept_items = 0
+        for tool_message in messages[index + 1 : end]:
+            if _message_type(tool_message) != "tool":
+                continue
+            tool_messages += 1
+            items = _first_list(_extract_tool_payload(tool_message))
+            raw_items += len(items)
+            for raw_item in items:
+                normalized = _normalize_reference_item(raw_item, normalized_items + 1)
+                if not normalized:
+                    continue
+                normalized_items += 1
+                kept_items += 1
+        total_tool_messages += tool_messages
+        total_raw_items += raw_items
+        total_normalized_items += normalized_items
+        total_kept_items += kept_items
+        turn_summaries.append(
+            {
+                "human_index": index,
+                "tool_messages": tool_messages,
+                "raw_items": raw_items,
+                "normalized_items": normalized_items,
+                "kept_items": kept_items,
+            }
+        )
+    batch_summaries = [
+        {
+            "id": batch.get("id"),
+            "assistant_message_id": batch.get("assistant_message_id"),
+            "source_count": len(batch.get("sources") or []),
+        }
+        for batch in (batches or build_reference_batches(messages))
+    ]
+    return {
+        "messages": len(messages),
+        "turns": len(turn_summaries),
+        "tool_messages": total_tool_messages,
+        "raw_items": total_raw_items,
+        "normalized_items": total_normalized_items,
+        "kept_items": total_kept_items,
+        "turns_detail": turn_summaries,
+        "batches": batch_summaries,
+    }
+
+
+def _collect_sources(messages: list[Any]) -> list[dict[str, Any]]:
+    sources: list[dict[str, Any]] = []
+    for message in messages:
+        if _message_type(message) != "tool":
+            continue
+        payload = _extract_tool_payload(message)
+        for raw_item in _first_list(payload):
+            normalized = _normalize_reference_item(raw_item, len(sources) + 1)
+            if not normalized:
+                continue
+            normalized["index"] = len(sources) + 1
+            sources.append(normalized)
+    return sources
+
+
+def _find_last_assistant_answer(messages: list[Any], start: int, end: int) -> Any | None:
+    for index in range(end - 1, start - 1, -1):
+        candidate = messages[index]
+        if (
+            _message_type(candidate) == "ai"
+            and _message_id(candidate)
+            and _has_answer_content(candidate)
+            and not _has_tool_calls(candidate)
+        ):
+            return candidate
+    return None
+
+
+def _message_type(message: Any) -> str:
+    if isinstance(message, dict):
+        return str(message.get("type") or message.get("role") or "")
+    return str(getattr(message, "type", "") or "")
+
+
+def _message_id(message: Any) -> str | None:
+    value = message.get("id") if isinstance(message, dict) else getattr(message, "id", None)
+    return str(value) if value else None
+
+
+def _message_content(message: Any) -> Any:
+    return message.get("content") if isinstance(message, dict) else getattr(message, "content", None)
+
+
+def _has_tool_calls(message: Any) -> bool:
+    if isinstance(message, dict):
+        direct = message.get("tool_calls")
+        chunks = message.get("tool_call_chunks")
+        extra = (message.get("additional_kwargs") or {}).get("tool_calls")
+    else:
+        direct = getattr(message, "tool_calls", None)
+        chunks = getattr(message, "tool_call_chunks", None)
+        extra = (getattr(message, "additional_kwargs", None) or {}).get("tool_calls")
+    return bool(direct or chunks or extra)
+
+
+def _has_answer_content(message: Any) -> bool:
+    content = _message_content(message)
+    if isinstance(content, str):
+        return bool(content.strip())
+    if isinstance(content, list):
+        return any(
+            isinstance(part, dict)
+            and part.get("type") == "text"
+            and str(part.get("text") or "").strip()
+            for part in content
+        )
+    return False
+
+
+def _extract_tool_payload(message: Any) -> Any:
+    content = _message_content(message)
+    if isinstance(content, dict):
+        return _unwrap_tool_payload(content)
+    if isinstance(content, list):
+        texts: list[str] = []
+        for part in content:
+            if not isinstance(part, dict):
+                continue
+            if isinstance(part.get("json"), dict):
+                return _unwrap_tool_payload(part["json"])
+            if part.get("type") == "json" and isinstance(part.get("json"), dict):
+                return _unwrap_tool_payload(part["json"])
+            if part.get("type") == "text" and isinstance(part.get("text"), str):
+                texts.append(part["text"])
+            unwrapped = _unwrap_tool_payload(part)
+            if unwrapped is not None:
+                return unwrapped
+        content = "\n".join(texts)
+    if isinstance(content, str):
+        return _parse_json_like(content)
+    return None
+
+
+def _unwrap_tool_payload(value: dict[str, Any]) -> Any:
+    list_keys = ("results", "data", "items", "hits", "documents", "records", "sources")
+    if any(isinstance(value.get(key), list) for key in list_keys):
+        return value
+    for key in ("result", "output", "stdout", "content", "text", "payload", "response"):
+        nested = value.get(key)
+        if isinstance(nested, str):
+            parsed = _parse_json_like(nested)
+            if parsed is not None:
+                return parsed
+        if isinstance(nested, dict):
+            unwrapped = _unwrap_tool_payload(nested)
+            if unwrapped is not None:
+                return unwrapped
+    return value
+
+
+def _parse_json_like(text: str) -> Any:
+    trimmed = text.strip()
+    if not trimmed:
+        return None
+    try:
+        return json.loads(trimmed)
+    except Exception:
+        pass
+    first = trimmed.find("{")
+    last = trimmed.rfind("}")
+    if 0 <= first < last:
+        try:
+            return json.loads(trimmed[first : last + 1])
+        except Exception:
+            return None
+    return None
+
+
+def _first_list(payload: Any) -> list[Any]:
+    if isinstance(payload, list):
+        return payload
+    if not isinstance(payload, dict):
+        return []
+    for key in ("results", "data", "items", "hits", "documents", "records", "entries", "matches", "chunks", "sources"):
+        value = payload.get(key)
+        if isinstance(value, list):
+            return value
+    for key in ("result", "payload", "response", "output", "data"):
+        found = _first_list(payload.get(key))
+        if found:
+            return found
+    return []
+
+
+def _nested_records(raw: Any) -> list[dict[str, Any]]:
+    if not isinstance(raw, dict):
+        return []
+    layers = [raw]
+    for key in ("metadata", "meta", "attrs", "attributes", "properties", "extra"):
+        nested = raw.get(key)
+        if isinstance(nested, dict):
+            layers.append(nested)
+    return layers
+
+
+def _pick(layers: list[dict[str, Any]], keys: tuple[str, ...]) -> str | None:
+    for layer in layers:
+        for key in keys:
+            value = layer.get(key)
+            if isinstance(value, (str, int, float)) and str(value).strip():
+                return str(value).strip()
+    return None
+
+
+def _normalize_reference_item(raw: Any, fallback_index: int) -> dict[str, Any] | None:
+    if not isinstance(raw, dict):
+        return None
+    layers = _nested_records(raw)
+    title = _pick(layers, ("title", "m_title", "name", "question", "keyword", "query", "text"))
+    content = _pick(layers, ("content", "page_content", "summary", "snippet", "description", "content_preview", "abstract", "body"))
+    url = _pick(layers, ("url", "link", "href", "source_url", "web_url"))
+    source = _pick(layers, ("source", "source1", "site", "provider", "from"))
+    time = _pick(layers, ("time", "date", "published_date", "publishedDate", "m_publish", "publish_time", "created_at"))
+    item_id = _pick(layers, ("recUuid", "rec_uuid", "uuid", "docId", "doc_id", "id", "recordId", "taskId"))
+    score_raw = _pick(layers, ("score", "relevance", "similarity"))
+    score = None
+    if score_raw is not None:
+        try:
+            score = float(score_raw)
+        except Exception:
+            score = None
+    if not title:
+        title = source or f"结果 {fallback_index}"
+    if not content:
+        content = ""
+    if not title and not content and not url:
+        return None
+    return {
+        "id": item_id or url,
+        "skillName": "backend-reference",
+        "mode": "citation",
+        "title": title,
+        "type": "result",
+        "snippet": content,
+        "content": content,
+        "source": source,
+        "time": time,
+        "url": url,
+        "score": score,
+        "raw": raw,
+    }
+
+
+def _reference_key(source: dict[str, Any]) -> str:
+    raw = source.get("raw") if isinstance(source.get("raw"), dict) else {}
+    item_id = source.get("id") or source.get("url") or raw.get("recUuid") or raw.get("uuid") or raw.get("docId") or raw.get("id")
+    if item_id:
+        return str(item_id).casefold().strip()
+    return "|".join(
+        str(source.get(key) or "").casefold().strip()
+        for key in ("title", "source", "time", "content", "snippet")
+    )