"""Server-owned run policy for knowledge-base research conversations.""" from __future__ import annotations from copy import deepcopy from typing import Any from fastapi import HTTPException, Request from langchain_core.messages import HumanMessage _SAFE_GENERATION_PREFERENCE_KEYS = frozenset( { "model_name", "mode", "reasoning_effort", "thinking_enabled", "thinking_force_disabled", } ) def _text_only_content(value: Any) -> Any: """Remove file/image/audio blocks while preserving the user's text.""" if isinstance(value, str): return value if isinstance(value, dict): block_type = str(value.get("type") or "text").strip().lower() text = value.get("text") if block_type in {"text", "input_text"} else None if not isinstance(text, str) and block_type in {"text", "input_text"}: text = value.get("content") return ( [{"type": "text", "text": text}] if isinstance(text, str) else "" ) if not isinstance(value, list): return "" kept: list[Any] = [] for block in value: if isinstance(block, str): kept.append(block) continue if not isinstance(block, dict): continue block_type = str(block.get("type") or "text").strip().lower() if block_type not in {"text", "input_text"}: continue text = block.get("text") if not isinstance(text, str): text = block.get("content") if isinstance(text, str): kept.append({"type": "text", "text": text}) return kept def _content_text(value: Any) -> str: if isinstance(value, str): return value if not isinstance(value, list): return "" parts: list[str] = [] for block in value: if isinstance(block, str): parts.append(block) elif isinstance(block, dict): text = block.get("text") or block.get("content") if isinstance(text, str): parts.append(text) return "\n".join(parts) def _sanitize_messages(body: Any) -> None: payload = getattr(body, "input", None) messages = payload.get("messages") if isinstance(payload, dict) else None if not isinstance(messages, list): if isinstance(payload, dict): body.input = {"messages": []} return latest_user_message: Any | None = None for message in messages: if isinstance(message, HumanMessage): latest_user_message = HumanMessage( content=_text_only_content(message.content) ) continue if not isinstance(message, dict): continue role = str(message.get("role") or message.get("type") or "").lower() if role not in {"human", "user"}: # Checkpoint state is the only source of prior assistant/system/tool # turns. The current request may contribute user text only. continue latest_user_message = { "role": "user", "content": _text_only_content(message.get("content")), } # The checkpoint is the only source of conversation history. A run may # contribute exactly one current user turn, so callers cannot fabricate # earlier user messages either. body.input = { "messages": [latest_user_message] if latest_user_message is not None else [] } def _safe_generation_preferences(body: Any) -> dict[str, Any]: """Copy the ordinary-chat model controls, and no other caller context.""" sources: list[dict[str, Any]] = [] config = getattr(body, "config", None) if isinstance(config, dict): for scope in ("configurable", "context"): nested = config.get(scope) if isinstance(nested, dict): sources.append(nested) context = getattr(body, "context", None) if isinstance(context, dict): # Public body.context has the same precedence it has in ordinary Q&A. sources.append(context) preferences: dict[str, Any] = {} for source in sources: for key in _SAFE_GENERATION_PREFERENCE_KEYS: value = source.get(key) if key in {"thinking_enabled", "thinking_force_disabled"}: if isinstance(value, bool): preferences[key] = value elif isinstance(value, str) and value.strip(): preferences[key] = value.strip()[:255] return preferences async def prepare_knowledge_research_run( body: Any, thread_id: str, request: Request, ) -> tuple[Any, dict[str, Any] | None]: """Bind every run to its persisted owner and one authorized knowledge base. Thread metadata is observability only. The database binding is resolved on every turn, and read authorization is rechecked before any retrieval runs. """ store = getattr(request.app.state, "knowledge_research_store", None) if store is None: return body, None session = await store.get_by_thread(thread_id) if session is None: return body, None # Local import avoids a module cycle: analysis_projects invokes this hook. from app.gateway.analysis_projects import actor user_id, is_admin = await actor(request) if session["user_id"] != user_id: # Administrators do not inherit access to another user's private chat. raise HTTPException(status_code=404, detail="研究会话不存在或无权访问") mapping_store = getattr(request.app.state, "llmwiki_store", None) if mapping_store is None: raise HTTPException(status_code=503, detail="知识库服务暂不可用") mapping = await mapping_store.get_authorized( session["knowledge_base_id"], user_id, write=False, is_admin=is_admin, ) if mapping is None: raise HTTPException(status_code=404, detail="知识库不存在或已无访问权限") prepared = body.model_copy(deep=True) if hasattr(body, "model_copy") else deepcopy(body) generation_preferences = _safe_generation_preferences(prepared) # Research accepts user text, not arbitrary run context/metadata. Rebuild # both from server state so task/workbench/notebook/peer context cannot be # smuggled into this otherwise strict single-KB surface. prepared.context = {} prepared.config = {} prepared.metadata = {} _sanitize_messages(prepared) messages = (prepared.input or {}).get("messages") if not isinstance(messages, list) or not messages: raise HTTPException(status_code=422, detail="知识库研究仅接受非空文本问题") latest = messages[-1] latest_content = ( latest.content if isinstance(latest, HumanMessage) else latest.get("content") ) if not _content_text(latest_content).strip(): raise HTTPException(status_code=422, detail="知识库研究不支持纯附件提问") policy = { **generation_preferences, "llmwiki_knowledge_base_ids": [session["knowledge_base_id"]], "assistant_knowledge_base_ids": [], "knowledge_retrieval_mode": "classic", "knowledge_space_strict": True, "rag_mode_enabled": True, # Retrieval is injected by the Gateway; do not let the graph open a # second, client-configurable retrieval path. "knowledge_rag_enabled": False, "agent_id": None, "agent_name": None, "writing_mode": False, "writing_artifact_path": None, "is_plan_mode": False, "subagent_enabled": False, "is_bootstrap": False, "is_writing_setup": False, "is_report_structure_setup": False, "memory_recall_disabled": True, "memory_injection_enabled": False, "prompt_prefix_enabled": False, "disable_tools": True, "excluded_tools": ["memory", "hindsight_recall"], "skill_stop_names": None, "skill_stop_message": None, "knowledge_research_session_id": session["id"], "knowledge_research_knowledge_base_id": session["knowledge_base_id"], } prepared.assistant_id = None prepared.excluded_tools = None prepared.skill_stop_names = None prepared.skill_stop_message = None prepared.context.update(policy) # Pass exactly one LangGraph runtime options container. build_run_config # gives ``context`` precedence when both are present; emitting both here # would discard configurable.thread_id and detach the checkpoint. prepared.config = { "configurable": { "thread_id": thread_id, **policy, } } prepared.command = None prepared.checkpoint_id = None prepared.checkpoint = None prepared.webhook = None prepared.metadata.update( { "system": True, "thread_type": "knowledge_research", "knowledge_research_session_id": session["id"], "knowledge_research_knowledge_base_id": session["knowledge_base_id"], "llmwiki_knowledge_base_ids": [session["knowledge_base_id"]], "assistant_knowledge_base_ids": [], "knowledge_retrieval_mode": "classic", "knowledge_space_strict": True, "rag_mode_enabled": True, "rag_mode": True, } ) marker = { "id": session["id"], "mode": "knowledge_research", "knowledge_base_id": session["knowledge_base_id"], "user_id": session["user_id"], } return prepared, marker def inject_validated_knowledge_research_context( config: dict[str, Any], context: dict[str, Any] | None, ) -> None: """Write non-public research policy into both final runtime scopes.""" if not context or not context.get("knowledge_research_session_id"): return authoritative = { "disable_tools": True, "memory_recall_disabled": True, "memory_injection_enabled": False, "prompt_prefix_enabled": False, "writing_mode": False, "is_plan_mode": False, "subagent_enabled": False, } for scope in ("configurable", "context"): target = config.setdefault(scope, {}) if isinstance(target, dict): target.update(authoritative) __all__ = [ "inject_validated_knowledge_research_context", "prepare_knowledge_research_run", ]