286 lines
10 KiB
Python
286 lines
10 KiB
Python
"""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",
|
|
]
|