"""Deep Research configuration — frozen per-task snapshot. Every deep-research job captures the *full* ``DeepResearchConfig`` as an immutable ``config_snapshot`` at creation time; the running job reads **only** that snapshot and never consults mutable global config mid-run (prevents mid-flight setting changes from corrupting a long research session). Server-side clamps (``clamp()``) bound every numeric field regardless of what the client sent — client-side limits are convenience only. """ from __future__ import annotations from typing import Any, Literal from pydantic import BaseModel, Field, field_validator # ── enums (kept as plain Literal unions so the model is JSON-serialisable) ── ResearchMode = Literal["quick", "basic", "detailed", "deep", "multi_agent"] ResearchTone = Literal["objective", "formal", "analytical", "persuasive"] ReportFormat = Literal["markdown", "apa", "mla", "chicago"] ModelRole = Literal["fast", "smart", "strategic"] MaterialChannel = Literal["web_search", "knowledge", "skill", "mcp", "file", "other"] SourceType = Literal["web", "knowledge", "skill", "mcp", "file", "other"] # How the collecting stage runs. ``chat`` (the default) means the frontend # drives a real collector-agent conversation on the session's runtime thread # and the writing job harvests materials from that thread; ``legacy`` keeps the # runner-driven skill/web tiers searching on their own. CollectionMode = Literal["chat", "legacy"] # How the writer treats ``custom_outline``. # ``fixed`` (固定大纲) reproduces the stored headings and hierarchy. # ``adaptive`` (弹性大纲) keeps the current material-driven layout. StructureMode = Literal["fixed", "adaptive"] # ── server-side bounds (§4.3 / §19.4) ──────────────────────────────────────── # Centralised so the router, the executor, and tests all share one truth source. _MAX_SEARCH_RESULTS_PER_QUERY = 20 _MAX_ITERATIONS = 10 _MAX_SUBTOPICS = 20 _MAX_DEEP_BREADTH = 8 _MAX_DEEP_DEPTH = 4 _MAX_CONCURRENCY = 8 _MAX_CONTEXT_WORDS = 120_000 _MAX_REVISION_LOOPS = 5 MAX_GENERATED_IMAGES = 4 _MAX_CUSTOM_OUTLINE_CHARS = 4_000 _MAX_REPORT_INSTRUCTION_CHARS = 4_000 _MAX_CONTROL_PLANE_TIMEOUT_SECONDS = 120.0 # Per-user active job ceiling (§19.4 / §26). DEFAULT_MAX_ACTIVE_JOBS_PER_USER = 3 class DeepResearchConfig(BaseModel): """Per-task research configuration. Field names mirror the upstream GPT Researcher knobs but use DeerFlow conventions (``snake_case``, explicit clamps, no ``os.environ`` reads). """ mode: ResearchMode = "basic" collection_mode: CollectionMode = "chat" language: str = "zh-CN" tone: ResearchTone = "objective" report_format: ReportFormat = "markdown" # Model role overrides; ``None`` → fall back to the system default model. fast_model: str | None = None smart_model: str | None = None strategic_model: str | None = None thinking_enabled: bool = False # Research budget knobs (client may request, server clamps). max_search_results_per_query: int = Field(default=8, ge=1) max_iterations: int = Field(default=3, ge=1) max_subtopics: int = Field(default=8, ge=1) deep_breadth: int = Field(default=4, ge=1) deep_depth: int = Field(default=2, ge=1) max_concurrency: int = Field(default=3, ge=1) max_context_words: int = Field(default=24_000, ge=500) # A provider may retain a hung HTTP request for several minutes. Planning, # query generation, and review calls must fail in a bounded time so the # durable job can either fall back to the topic or finish with a real error. # Report prose deliberately uses ``stream_complete`` and is not subject to # this total-duration cap. control_plane_timeout_seconds: float = Field(default=45.0, ge=0.1) # Editorial / quality knobs. curate_sources: bool = True generate_images: bool = False max_generated_images: int = Field(default=2, ge=1) include_human_feedback: bool = False max_revision_loops: int = Field(default=2, ge=0) # Material channels — only admin-configured tool groups are accepted # (router validates against capabilities; ``other`` reserved for future). allowed_material_channels: list[MaterialChannel] = Field(default_factory=lambda: ["web_search"]) allowed_domains: list[str] = Field(default_factory=list) # Optional user-supplied outline (sanitised; never overrides system safety # prompts in the vendored writer). custom_outline: str | None = Field(default=None, max_length=_MAX_CUSTOM_OUTLINE_CHARS) # Optional write-time requirements confirmed by the user. This is kept # separate from ``custom_outline`` so prose/style requirements can never be # misread as extra chapter headings by detailed runners. report_instruction: str | None = Field(default=None, max_length=_MAX_REPORT_INSTRUCTION_CHARS) # Writing constraint copied from the selected report structure (结构约束). # Default ``adaptive`` preserves today's material-driven layout. structure_mode: StructureMode = "adaptive" # Optional search facets copied from the selected report structure (检索方向). # Empty → unconstrained query-planning. Non-empty → split retrievable terms # along those facets; never treat the labels themselves as search queries. retrieval_directions: list[str] = Field(default_factory=list) # Optional skill names copied from the selected report structure (检索来源). # Empty → collector / material provider keep today's default sources. retrieval_skills: list[str] = Field(default_factory=list) model_config = {"extra": "forbid"} # ── server-side clamping ──────────────────────────────────────────────── def clamp(self) -> DeepResearchConfig: """Return a copy with every numeric field bounded to its server ceiling. Call this at job-creation time so the persisted snapshot is already safe; the executor can then trust ``config_snapshot`` unconditionally. """ return self.model_copy( update={ "max_search_results_per_query": min( self.max_search_results_per_query, _MAX_SEARCH_RESULTS_PER_QUERY ), "max_iterations": min(self.max_iterations, _MAX_ITERATIONS), "max_subtopics": min(self.max_subtopics, _MAX_SUBTOPICS), "deep_breadth": min(self.deep_breadth, _MAX_DEEP_BREADTH), "deep_depth": min(self.deep_depth, _MAX_DEEP_DEPTH), "max_concurrency": min(self.max_concurrency, _MAX_CONCURRENCY), "max_context_words": min(self.max_context_words, _MAX_CONTEXT_WORDS), "control_plane_timeout_seconds": min( self.control_plane_timeout_seconds, _MAX_CONTROL_PLANE_TIMEOUT_SECONDS ), "max_revision_loops": min(self.max_revision_loops, _MAX_REVISION_LOOPS), "max_generated_images": min(self.max_generated_images, MAX_GENERATED_IMAGES), } ) @field_validator("allowed_material_channels") @classmethod def _dedupe_channels(cls, v: list[MaterialChannel]) -> list[MaterialChannel]: # Preserve order, drop duplicates. seen: set[str] = set() out: list[MaterialChannel] = [] for ch in v: if ch not in seen: seen.add(ch) out.append(ch) return out @field_validator("custom_outline", mode="before") @classmethod def _sanitize_custom_outline(cls, value: Any) -> str | None: """Keep a bounded user outline as content, never as a system prompt. Newlines are retained because Markdown/numbered headings are useful to report planning; NUL/control separators are removed so the snapshot is portable across the database, events and artifact JSON. """ if value is None: return None cleaned = str(value).replace("\x00", "").replace("\r", "").strip() return cleaned or None @field_validator("report_instruction", mode="before") @classmethod def _sanitize_report_instruction(cls, value: Any) -> str | None: if value is None: return None cleaned = str(value).replace("\x00", "").replace("\r", "").strip() return cleaned or None @field_validator("structure_mode", mode="before") @classmethod def _normalize_structure_mode(cls, value: Any) -> StructureMode: raw = str(value or "").strip().lower() if raw in {"fixed", "canonical", "prescriptive", "rigid"}: return "fixed" return "adaptive" @field_validator("retrieval_directions", mode="before") @classmethod def _normalize_retrieval_directions(cls, value: Any) -> list[str]: from deerflow.persistence.report_structures.directions import normalize_retrieval_directions return normalize_retrieval_directions(value) @field_validator("retrieval_skills", mode="before") @classmethod def _normalize_retrieval_skills(cls, value: Any) -> list[str]: from deerflow.persistence.report_structures.skills import normalize_retrieval_skills return normalize_retrieval_skills(value) def resolve_model_name(self, role: ModelRole) -> str | None: """Return the configured model name for a role, or ``None`` for system default.""" return { "fast": self.fast_model, "smart": self.smart_model, "strategic": self.strategic_model, }[role] # Convenience: the full set of supported modes, for capability discovery. ALL_MODES: tuple[ResearchMode, ...] = ( "quick", "basic", "detailed", "deep", "multi_agent", ) def config_from_dict(data: dict[str, Any]) -> DeepResearchConfig: """Build + clamp a config from a raw dict (e.g. JSON column / request body).""" return DeepResearchConfig.model_validate(data).clamp()