439 lines
18 KiB
Python
439 lines
18 KiB
Python
"""Project ownership, knowledge validation and server-controlled run policy."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import re
|
|
from copy import deepcopy
|
|
|
|
from fastapi import HTTPException, Request
|
|
|
|
from app.gateway.auth.disabled_mode import is_auth_disabled
|
|
from app.gateway.deps import get_agent_store, get_checkpointer, get_optional_user_from_request
|
|
from deerflow.agents.analysis_project import (
|
|
SOLUTION_INTENT_BLOCK_LABEL,
|
|
SOLUTION_REQUIREMENTS_CONFIRMATION_TOKEN,
|
|
SOLUTION_REQUIREMENTS_FORCE_TOKEN,
|
|
SOLUTION_REQUIREMENTS_READY_MARKER,
|
|
SOLUTION_TEMPLATE_ID,
|
|
)
|
|
from deerflow.config.agents_config import load_agent_config, validate_agent_id
|
|
from deerflow.runtime import serialize_channel_values
|
|
from deerflow.runtime.user_context import get_effective_user_id
|
|
|
|
_SOLUTION_INTENT_BLOCK_PATTERN = re.compile(
|
|
rf"<!--\s*{re.escape(SOLUTION_INTENT_BLOCK_LABEL)}\s*(?P<payload>[\s\S]*?)-->",
|
|
)
|
|
_SOLUTION_INTENT_BLOCK_OPEN_PATTERN = re.compile(rf"<!--\s*{re.escape(SOLUTION_INTENT_BLOCK_LABEL)}")
|
|
_SOLUTION_INTENT_SCALAR_FIELDS = (
|
|
"goal",
|
|
"audience",
|
|
"success_outcome",
|
|
"scope",
|
|
"constraints",
|
|
"deliverable",
|
|
"acceptance_criteria",
|
|
)
|
|
_SOLUTION_INTENT_MAX_PAYLOAD_CHARS = 20_000
|
|
_SOLUTION_INTENT_MAX_STRING_CHARS = 2_000
|
|
_SOLUTION_INTENT_MAX_LIST_ITEMS = 20
|
|
_SOLUTION_INTENT_MAX_LIST_ITEM_CHARS = 1_000
|
|
|
|
|
|
async def validate_project_agent(request: Request, agent_id: str | None):
|
|
if not agent_id:
|
|
return None
|
|
try:
|
|
validate_agent_id(agent_id)
|
|
except ValueError as exc:
|
|
raise HTTPException(422, "智能体标识无效") from exc
|
|
user_id, is_admin = await actor(request)
|
|
store = get_agent_store(request)
|
|
row = await store.get_visible(agent_id, user_id)
|
|
if row is None and is_admin:
|
|
row = await store.get_any(agent_id)
|
|
if row is None:
|
|
raise HTTPException(422, "所选智能体不存在或无访问权限，请重新选择")
|
|
return row
|
|
|
|
|
|
def project_store(request: Request):
|
|
store = getattr(request.app.state, "analysis_project_store", None)
|
|
if store is None:
|
|
raise HTTPException(503, "项目存储尚未就绪")
|
|
return store
|
|
|
|
|
|
async def actor(request: Request) -> tuple[str, bool]:
|
|
user = await get_optional_user_from_request(request)
|
|
if user:
|
|
return str(user.id), getattr(user, "system_role", None) == "admin"
|
|
if is_auth_disabled():
|
|
return get_effective_user_id(), True
|
|
raise HTTPException(401, "请先登录")
|
|
|
|
|
|
async def owned_project(request: Request, project_id: str) -> dict:
|
|
user_id, _ = await actor(request)
|
|
project = await project_store(request).get(project_id, user_id)
|
|
if project is None:
|
|
raise HTTPException(404, "项目不存在或无权访问")
|
|
return project
|
|
|
|
|
|
async def validate_knowledge(request: Request, project: dict) -> None:
|
|
user_id, is_admin = await actor(request)
|
|
for store_name, field in (("llmwiki_store", "knowledge_base_ids"), ("assistant_knowledge_store", "assistant_knowledge_base_ids")):
|
|
ids = project.get(field) or []
|
|
if not ids:
|
|
continue
|
|
store = getattr(request.app.state, store_name, None)
|
|
if store is None:
|
|
raise HTTPException(503, "知识库服务暂不可用")
|
|
for base_id in ids:
|
|
row = await store.get_authorized(base_id, user_id, write=False, is_admin=is_admin) if field == "knowledge_base_ids" else await store.get_base(base_id)
|
|
if row is None:
|
|
raise HTTPException(422, "所选知识库不存在或已无访问权限，请重新配置项目背景")
|
|
|
|
|
|
def _message_content_text(value) -> str:
|
|
"""Flatten LangGraph's string or content-block message payloads."""
|
|
if isinstance(value, str):
|
|
return value
|
|
if isinstance(value, list):
|
|
return "\n".join(_message_content_text(item) for item in value)
|
|
if isinstance(value, dict):
|
|
block_type = value.get("type")
|
|
if block_type in {None, "text", "input_text", "output_text"} and isinstance(value.get("text"), str):
|
|
return value["text"]
|
|
if block_type in {None, "text", "input_text", "output_text"} and "content" in value:
|
|
return _message_content_text(value["content"])
|
|
return ""
|
|
|
|
|
|
def solution_requirements_confirmed_by_input(payload: dict | None) -> bool:
|
|
"""Accept confirmation only from a user message, never caller context."""
|
|
messages = (payload or {}).get("messages")
|
|
if not isinstance(messages, list) or not messages:
|
|
return False
|
|
message = messages[-1]
|
|
if not isinstance(message, dict):
|
|
return False
|
|
role = message.get("role") or message.get("type")
|
|
return role in {"human", "user"} and SOLUTION_REQUIREMENTS_CONFIRMATION_TOKEN in _message_content_text(message.get("content"))
|
|
|
|
|
|
def solution_force_generate_by_input(payload: dict | None) -> bool:
|
|
"""Accept force-generation only from the current final user message."""
|
|
messages = (payload or {}).get("messages")
|
|
if not isinstance(messages, list) or not messages:
|
|
return False
|
|
message = messages[-1]
|
|
if not isinstance(message, dict):
|
|
return False
|
|
role = message.get("role") or message.get("type")
|
|
return role in {"human", "user"} and SOLUTION_REQUIREMENTS_FORCE_TOKEN in _message_content_text(message.get("content"))
|
|
|
|
|
|
def _json_object_without_duplicate_keys(pairs: list[tuple[str, object]]) -> dict[str, object]:
|
|
value: dict[str, object] = {}
|
|
for key, item in pairs:
|
|
if key in value:
|
|
raise ValueError(f"duplicate JSON key: {key}")
|
|
value[key] = item
|
|
return value
|
|
|
|
|
|
def _normalized_intent_string(value: object, *, max_chars: int) -> str | None:
|
|
if not isinstance(value, str):
|
|
return None
|
|
normalized = value.strip()
|
|
if not normalized or len(normalized) > max_chars:
|
|
return None
|
|
return normalized
|
|
|
|
|
|
def _normalized_intent_string_list(value: object, *, allow_empty: bool) -> list[str] | None:
|
|
if not isinstance(value, list) or len(value) > _SOLUTION_INTENT_MAX_LIST_ITEMS:
|
|
return None
|
|
normalized: list[str] = []
|
|
for item in value:
|
|
text = _normalized_intent_string(item, max_chars=_SOLUTION_INTENT_MAX_LIST_ITEM_CHARS)
|
|
if text is None:
|
|
return None
|
|
normalized.append(text)
|
|
if not allow_empty and not normalized:
|
|
return None
|
|
return normalized
|
|
|
|
|
|
def _validate_solution_intent_snapshot(value: object) -> dict[str, object] | None:
|
|
if not isinstance(value, dict):
|
|
return None
|
|
snapshot: dict[str, object] = {}
|
|
for field in _SOLUTION_INTENT_SCALAR_FIELDS:
|
|
normalized = _normalized_intent_string(value.get(field), max_chars=_SOLUTION_INTENT_MAX_STRING_CHARS)
|
|
if normalized is None:
|
|
return None
|
|
snapshot[field] = normalized
|
|
open_questions = _normalized_intent_string_list(value.get("open_questions"), allow_empty=True)
|
|
if open_questions is None:
|
|
return None
|
|
snapshot["open_questions"] = open_questions
|
|
return snapshot
|
|
|
|
|
|
def _solution_intent_block(text: str) -> tuple[dict[str, object], int, int] | None:
|
|
if len(_SOLUTION_INTENT_BLOCK_OPEN_PATTERN.findall(text)) != 1:
|
|
return None
|
|
matches = list(_SOLUTION_INTENT_BLOCK_PATTERN.finditer(text))
|
|
if len(matches) != 1:
|
|
return None
|
|
match = matches[0]
|
|
payload = match.group("payload").strip()
|
|
if not payload or len(payload) > _SOLUTION_INTENT_MAX_PAYLOAD_CHARS:
|
|
return None
|
|
try:
|
|
value = json.loads(payload, object_pairs_hook=_json_object_without_duplicate_keys)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
snapshot = _validate_solution_intent_snapshot(value)
|
|
if snapshot is None:
|
|
return None
|
|
return snapshot, match.start(), match.end()
|
|
|
|
|
|
def parse_solution_intent_snapshot(content: object) -> dict[str, object] | None:
|
|
"""Parse one bounded, allowlisted intent snapshot from assistant content."""
|
|
parsed = _solution_intent_block(_message_content_text(content))
|
|
return parsed[0] if parsed else None
|
|
|
|
|
|
def latest_solution_intent_snapshot_in_messages(messages: object) -> dict[str, object] | None:
|
|
"""Return the newest valid assistant snapshot, including incomplete ones.
|
|
|
|
Force-generation may reuse an earlier best-effort snapshot whose
|
|
``open_questions`` are still non-empty. It remains only a hint: the force
|
|
prompt requires the model to reconcile it with the complete conversation.
|
|
At least one preceding user turn is required so an orphan assistant message
|
|
cannot manufacture project intent.
|
|
"""
|
|
if not isinstance(messages, list):
|
|
return None
|
|
saw_user = False
|
|
latest: dict[str, object] | None = None
|
|
for message in messages:
|
|
if not isinstance(message, dict):
|
|
continue
|
|
role = message.get("role") or message.get("type")
|
|
if role in {"human", "user"}:
|
|
saw_user = True
|
|
continue
|
|
if role not in {"ai", "assistant"} or not saw_user:
|
|
continue
|
|
parsed = parse_solution_intent_snapshot(message.get("content"))
|
|
if parsed is not None:
|
|
latest = parsed
|
|
return latest
|
|
|
|
|
|
def solution_intent_state_in_messages(messages: object) -> tuple[dict[str, object] | None, bool]:
|
|
"""Return the latest assistant snapshot and whether it is safe to generate."""
|
|
if not isinstance(messages, list):
|
|
return None, False
|
|
latest_user_index = -1
|
|
latest_assistant_index = -1
|
|
latest_assistant_text = ""
|
|
for index, message in enumerate(messages):
|
|
if not isinstance(message, dict):
|
|
continue
|
|
role = message.get("role") or message.get("type")
|
|
if role in {"human", "user"}:
|
|
latest_user_index = index
|
|
elif role in {"ai", "assistant"}:
|
|
latest_assistant_index = index
|
|
latest_assistant_text = _message_content_text(message.get("content"))
|
|
if latest_user_index < 0 or latest_assistant_index <= latest_user_index:
|
|
return None, False
|
|
parsed = _solution_intent_block(latest_assistant_text)
|
|
if parsed is None:
|
|
return None, False
|
|
snapshot, block_start, block_end = parsed
|
|
heading = "## 意图理解"
|
|
heading_index = latest_assistant_text.rfind(heading, 0, block_start)
|
|
ready_index = latest_assistant_text.find(SOLUTION_REQUIREMENTS_READY_MARKER, block_end)
|
|
visible_summary = latest_assistant_text[heading_index + len(heading) : block_start].strip() if heading_index >= 0 else ""
|
|
ready_tail = latest_assistant_text[ready_index + len(SOLUTION_REQUIREMENTS_READY_MARKER) :] if ready_index >= 0 else ""
|
|
ready = bool(heading_index >= 0 and visible_summary and ready_index >= 0 and latest_assistant_text.count(SOLUTION_REQUIREMENTS_READY_MARKER) == 1 and not ready_tail.strip() and not snapshot["open_questions"])
|
|
return snapshot, ready
|
|
|
|
|
|
def solution_requirements_ready_in_messages(messages: object) -> bool:
|
|
"""Require a fresh validated intent snapshot before accepting consent."""
|
|
return solution_intent_state_in_messages(messages)[1]
|
|
|
|
|
|
async def prepare_project_run(body, thread_id: str, request: Request):
|
|
"""Resolve policy from the immutable binding, never from caller metadata.
|
|
|
|
Applies to all Gateway run endpoints, including direct SDK calls. Strip
|
|
forged policy even on ordinary threads; all other chat behavior is preserved.
|
|
"""
|
|
body = body.model_copy(deep=True) if hasattr(body, "model_copy") else deepcopy(body)
|
|
body.context = dict(body.context or {})
|
|
body.context.pop("analysis_project", None)
|
|
body.context.pop("project_memory_scope", None)
|
|
body.context.pop("project_memory", None)
|
|
body.context.pop("solution_requirements_confirmed", None)
|
|
body.context.pop("solution_force_generate", None)
|
|
body.context.pop("solution_intent", None)
|
|
body.config = dict(body.config or {})
|
|
body.config.pop("solution_requirements_confirmed", None)
|
|
body.config.pop("solution_force_generate", None)
|
|
body.config.pop("solution_intent", None)
|
|
for key in ("context", "configurable"):
|
|
if isinstance(body.config.get(key), dict):
|
|
body.config[key].pop("analysis_project", None)
|
|
body.config[key].pop("project_memory_scope", None)
|
|
body.config[key].pop("project_memory", None)
|
|
body.config[key].pop("solution_requirements_confirmed", None)
|
|
body.config[key].pop("solution_force_generate", None)
|
|
body.config[key].pop("solution_intent", None)
|
|
body.metadata = dict(body.metadata or {})
|
|
body.metadata.pop("project_memory_scope", None)
|
|
body.metadata.pop("project_memory", None)
|
|
body.metadata.pop("solution_requirements_confirmed", None)
|
|
body.metadata.pop("solution_force_generate", None)
|
|
body.metadata.pop("solution_intent", None)
|
|
store = getattr(request.app.state, "analysis_project_store", None)
|
|
if store is None:
|
|
return body, None
|
|
project = await store.project_for_thread(thread_id)
|
|
if project is None:
|
|
return body, None
|
|
user_id, _ = await actor(request)
|
|
if project["user_id"] != user_id:
|
|
raise HTTPException(404, "项目不存在或无权访问")
|
|
thread_config = await store.config_for(f"thread:{thread_id}")
|
|
agent_id = thread_config.get("agent_id")
|
|
await validate_project_agent(request, agent_id)
|
|
agent_config = await asyncio.to_thread(load_agent_config, agent_id) if agent_id else None
|
|
knowledge_ids = list(
|
|
dict.fromkeys(
|
|
[
|
|
*(project.get("knowledge_base_ids") or []),
|
|
*(getattr(agent_config, "llmwiki_knowledge_base_ids", None) or []),
|
|
]
|
|
)
|
|
)
|
|
assistant_knowledge_ids = list(
|
|
dict.fromkeys(project.get("assistant_knowledge_base_ids") or [])
|
|
)
|
|
await validate_knowledge(
|
|
request,
|
|
{
|
|
"knowledge_base_ids": knowledge_ids,
|
|
"assistant_knowledge_base_ids": assistant_knowledge_ids,
|
|
},
|
|
)
|
|
for message in (body.input or {}).get("messages", []):
|
|
if isinstance(message, dict) and isinstance(message.get("additional_kwargs"), dict):
|
|
# Source snapshots must come from this run's server retrieval.
|
|
for key in (
|
|
"llmwiki_reference_sources",
|
|
"llmwiki_reference_query",
|
|
"prompt_prefix",
|
|
"solution_intent",
|
|
"solution_requirements_confirmed",
|
|
"solution_force_generate",
|
|
):
|
|
message["additional_kwargs"].pop(key, None)
|
|
artifact_path = None
|
|
solution_intent = None
|
|
solution_requirements_ready = False
|
|
solution_force_generate = False
|
|
if project["mode"] == "solution":
|
|
checkpoint = await get_checkpointer(request).aget_tuple({"configurable": {"thread_id": thread_id}})
|
|
values = checkpoint.checkpoint.get("channel_values", {}) if checkpoint else {}
|
|
if "/mnt/user-data/outputs/report.md" in (values.get("artifacts") or []):
|
|
artifact_path = "/mnt/user-data/outputs/report.md"
|
|
serialized_values = serialize_channel_values(values)
|
|
checkpoint_messages = serialized_values.get("messages")
|
|
solution_intent, solution_requirements_ready = solution_intent_state_in_messages(checkpoint_messages)
|
|
solution_force_generate = solution_force_generate_by_input(body.input)
|
|
if solution_force_generate:
|
|
solution_intent = latest_solution_intent_snapshot_in_messages(checkpoint_messages)
|
|
solution_requirements_confirmed = bool(artifact_path) or (
|
|
project["mode"] == "solution"
|
|
and (
|
|
solution_force_generate
|
|
or (solution_requirements_ready and solution_requirements_confirmed_by_input(body.input))
|
|
)
|
|
)
|
|
project_memory_scope = None
|
|
project_memory = None
|
|
if project["mode"] == "qa":
|
|
from app.gateway.analysis_project_memory import (
|
|
latest_user_query,
|
|
prepare_project_memory,
|
|
)
|
|
|
|
project_memory_scope, project_memory = await prepare_project_memory(
|
|
request=request,
|
|
project=project,
|
|
thread_id=thread_id,
|
|
query=latest_user_query(body),
|
|
)
|
|
policy = {
|
|
"analysis_project": {
|
|
"id": project["id"],
|
|
"name": project["name"],
|
|
"mode": project["mode"],
|
|
"background": project["background"],
|
|
"template_id": SOLUTION_TEMPLATE_ID,
|
|
"report_config": project.get("report_config") or {},
|
|
"strategy_config": project.get("strategy_config") or {},
|
|
},
|
|
"llmwiki_knowledge_base_ids": knowledge_ids,
|
|
"assistant_knowledge_base_ids": assistant_knowledge_ids,
|
|
"writing_mode": project["mode"] == "solution",
|
|
# The artifact path is resolved by the existing agent/middleware, not a
|
|
# client-supplied path that could skip the first-report pipeline.
|
|
"writing_artifact_path": artifact_path,
|
|
"solution_requirements_confirmed": solution_requirements_confirmed,
|
|
"solution_force_generate": solution_force_generate,
|
|
"solution_intent": solution_intent,
|
|
"agent_id": agent_id,
|
|
"agent_name": None,
|
|
"is_plan_mode": False,
|
|
"subagent_enabled": False,
|
|
"is_bootstrap": False,
|
|
"is_writing_setup": False,
|
|
"is_report_structure_setup": False,
|
|
"knowledge_rag_enabled": False,
|
|
}
|
|
if project_memory_scope is not None:
|
|
policy.update(
|
|
{
|
|
"project_memory_scope": project_memory_scope,
|
|
"project_memory": project_memory,
|
|
"memory_recall_disabled": True,
|
|
"memory_injection_enabled": False,
|
|
}
|
|
)
|
|
body.assistant_id = agent_id or "lead_agent"
|
|
body.context.update(policy)
|
|
body.metadata = {
|
|
**(body.metadata or {}),
|
|
"agent_id": agent_id,
|
|
"agent_name": None,
|
|
"analysis_project_id": project["id"],
|
|
"analysis_project_mode": project["mode"],
|
|
"analysis_project_name": project["name"],
|
|
}
|
|
for key in ("context", "configurable"):
|
|
if isinstance(body.config.get(key), dict):
|
|
body.config[key].update(policy)
|
|
return body, project
|