긴 검색 요청의 조건과 사이트 범위를 보존함

apps/api/app/routers/sessions.py

@@ -116,6 +116,7 @@
     declines_web_search,
     requests_web_search,
     search_hints,
+    search_needs_planning,
     search_plan,
     search_query,
     weather_location,
@@ -3063,12 +3064,13 @@ async def send_message(
             outbound_history[fresh_followup_index]
             if fresh_followup_index is not None else content
         )
-        preset_call = (
-            "web_search", {
-                "query": search_query(lookup_content, prefer_primary=fresh_fact),
-                **search_hints(lookup_content),
-            },
-        )
+        if not search_needs_planning(lookup_content):
+            preset_call = (
+                "web_search", {
+                    "query": search_query(lookup_content, prefer_primary=fresh_fact),
+                    **search_hints(lookup_content),
+                },
+            )
     elif forced_tool == "weather" and "weather" in tool_names:
         place = weather_location(content)
         if place:
@@ -3113,9 +3115,8 @@ async def send_message(
                 protect_enrichment=policy.pii_masking or policy.external_data_guard,
                 privacy_audit_id=privacy_audit_id,
                 routing_audit_id=routing_audit_id,
-                # The toggle's search is the server's own first call (a named
-                # `tool_choice` is not reliably obeyed); a weather question whose
-                # place the words do not name is left to the model, forced.
+                # Short lookups are server presets. Long requests and weather
+                # without a named place use the allowed tool's forced planning hop.
                 preset_call=preset_call,
                 freshness_request=content if fresh_fact else None,
                 force_tool=(

apps/api/app/services/context.py

@@ -17,7 +17,11 @@
 
 from app.core.config import settings
 from app.models.chat import SessionKind
-from app.services.freshness import FRESHNESS_INSTRUCTION, without_quoted_transform_sources
+from app.services.freshness import (
+    FRESHNESS_INSTRUCTION,
+    _quoted_spans,
+    without_quoted_transform_sources,
+)
 
 # Models leak Chinese Hanja into Korean prose; parenthesised glosses are allowed.
 _KOREAN_ONLY = (
@@ -444,9 +448,22 @@ def asks_weather(request: str) -> bool:
 )
 
 
+SEARCH_QUERY_LIMIT = 120
+
+
+def search_needs_planning(request: str) -> bool:
+    """Do not cut a long request's subject, period or constraints into a preset."""
+    return len(re.sub(r"\s+", " ", (request or "").strip())) > SEARCH_QUERY_LIMIT
+
+
 def search_query(request: str, *, prefer_primary: bool = False) -> str:
     """The user's sentence as a search query: request phrasing trimmed, capped."""
-    text = re.sub(r"\s+", " ", _HINT_PHRASES.sub(" ", request or "").strip())
+    try:
+        text, _site = search_site_scope(request or "")
+    except ValueError:
+        # Leave conflicting operators for the search tool to reject before lookup.
+        text = request or ""
+    text = re.sub(r"\s+", " ", _HINT_PHRASES.sub(" ", text).strip())
     for _ in range(4):
         text = text.rstrip(" ?？!.。~,")
         peeled = _FILLER.sub(
@@ -456,13 +473,17 @@ def search_query(request: str, *, prefer_primary: bool = False) -> str:
             break
         text = peeled
     text = _DANGLING_UNIT.sub("", _DANGLING_PARTICLE.sub("", text.strip(" ?？!.。~,")))
-    query = (text or (request or "").strip())[:120]
-    if prefer_primary and not _HINT_SITE.search(request or "") and not _HINT_OFFICIAL.search(query):
+    query = (text or (request or "").strip())[:SEARCH_QUERY_LIMIT]
+    if (
+        prefer_primary
+        and not _search_site_operators(request or "")
+        and not _HINT_OFFICIAL.search(query)
+    ):
         # Improve retrieval without assuming a country, authority domain or answer.
         # Mixing an English cue into a Korean query can select the wrong search lane.
         hint = "공식" if _HANGUL.search(request or "") else "official"
         if hint not in query:
-            query = query[:120 - len(hint) - 1].rstrip() + " " + hint
+            query = query[:SEARCH_QUERY_LIMIT - len(hint) - 1].rstrip() + " " + hint
     return query
 
 
@@ -498,7 +519,7 @@ def is_small_talk(request: str) -> bool:
 #: back, in what language. Honoured on the server's own first search.
 _HINT_SITE = re.compile(r"\bsite:([A-Za-z0-9.-]+\.[A-Za-z]{2,})")
 _HINT_OFFICIAL = re.compile(
-    r"공식\s*(?:사이트|홈페이지|자료|발표|문서|출처|기준)|정부\s*(?:자료|발표|사이트)|"
+    r"공식\s*(?:사이트|홈페이지|자료|발표|문서|출처|기준|공지|공고)|정부\s*(?:자료|발표|사이트)|"
     r"기관\s*(?:자료|홈페이지)|공공기관|\bofficial\b",
     re.I,
 )
@@ -523,21 +544,64 @@ def is_small_talk(request: str) -> bool:
 _HINT_NEWS = re.compile(r"뉴스\s*(?:로|에서|기사|위주로)|기사\s*(?:로|에서|위주로)")
 #: The hint phrases, so `search_query` can leave them out of the query itself.
 _HINT_PHRASES = re.compile(
-    r"\bsite:[A-Za-z0-9.-]+|(?:공식|정부|기관)\s*(?:사이트|홈페이지|자료|발표|문서|출처)\s*"
+    r"(?:공식|정부|기관)\s*(?:사이트|홈페이지|자료|발표|문서|출처|공지|공고)\s*"
     r"(?:기준으로|기준|에서|으로|로|만)?|공공기관\s*(?:자료)?\s*(?:기준으로|에서|로)?|"
     r"(?:영어|영문|해외)\s*(?:자료|문서|기사)\s*(?:로|에서|위주로)?|"
     r"(?:뉴스|기사)\s*(?:위주로|로만)",
     re.I,
 )
 
 
+def _search_site_operators(query: str) -> list[re.Match[str]]:
+    """Only standalone positive operators, not quoted text or excluded sites."""
+    quoted = iter(_quoted_spans(query))
+    span = next(quoted, None)
+    operators = []
+    for match in _HINT_SITE.finditer(query):
+        while span and span[1] <= match.start():
+            span = next(quoted, None)
+        if span and span[0] <= match.start() < span[1]:
+            continue
+        if match.start() and not query[match.start() - 1].isspace():
+            continue
+        operators.append(match)
+    return operators
+
+
+def search_site_scope(query: str, site: object = None) -> tuple[str, str]:
+    """Use the existing single-domain hint grammar for model-written operators too."""
+    operators = _search_site_operators(query)
+    if any(query[match.end() : match.end() + 1] in {"/", ":"} for match in operators):
+        raise ValueError("search site must be a domain, not a path or port")
+    inline = {match[1].lower() for match in operators}
+    explicit = str(site or "").strip().lower().removeprefix("site:")
+    if explicit and not _HINT_SITE.fullmatch("site:" + explicit):
+        raise ValueError("invalid search site")
+    sites = inline | ({explicit} if explicit else set())
+    if len(sites) > 1:
+        raise ValueError("multiple search sites require separate queries")
+    scoped = next(iter(sites), "")
+    if inline:
+        parts = []
+        start = 0
+        for match in operators:
+            parts.append(query[start : match.start()])
+            start = match.end()
+        parts.append(query[start:])
+        query = re.sub(r"\s+", " ", " ".join(parts)).strip()
+    if not query:
+        raise ValueError("search query needs a subject beside site")
+    return query, scoped
+
+
 def search_hints(request: str) -> dict[str, object]:
     """What the user's own words say about where and how to search."""
     text = request or ""
     hints: dict[str, object] = {}
-    if match := _HINT_SITE.search(text):
-        hints["site"] = match.group(1).lower()
-    elif _HINT_OFFICIAL.search(text):
+    sites = {match[1].lower() for match in _search_site_operators(text)}
+    if len(sites) == 1:
+        hints["site"] = next(iter(sites))
+    elif not sites and _HINT_OFFICIAL.search(text):
         hints["official"] = True
     for pattern, span in _HINT_RANGE:
         if pattern.search(text):

apps/api/app/services/tools/builtin.py

@@ -374,9 +374,11 @@ async def _searxng(
     hides it); a `site:go.kr` lane for `official`; and `site`, `time_range`
     and `language` hints on every lane. Lane hits that fit the question come
     first."""
+    from app.services.context import search_site_scope
+
     hints = hints or {}
     search_url = f"{base_url.rstrip('/')}/search"
-    site = str(hints.get("site") or "").strip().lstrip("site:")
+    query, site = search_site_scope(query, hints.get("site"))
     q = f"{query} site:{site}" if site else query
     base: dict[str, Any] = {"q": q, "format": "json", "safesearch": 2, "language": "ko-KR"}
     if hints.get("language") in _LANGUAGES:
@@ -387,9 +389,14 @@ async def _searxng(
     lane = _LANES.get("news" if fresh and kind == "web" else kind)
     if lane:
         lane_params = {**base, **lane}
+        if hints.get("time_range") in _TIME_RANGES:
+            lane_params["time_range"] = hints["time_range"]
         if kind == "papers":
             # Titles are English; a Korean locale drags in unrelated Korean journals.
-            lane_params.update(q=_latin_only(query), language="en")
+            paper_query = _latin_only(query)
+            lane_params.update(
+                q=f"{paper_query} site:{site}" if site else paper_query, language="en"
+            )
         lane_requests.append(("kind", lane_params))
     if kind == "web" and base["language"] != "en" and not site:
         names, _ = _anchors(query)
@@ -427,6 +434,16 @@ def collect(payload: dict[str, Any], *, lane: str = "", lane_query: str = "") ->
         if not isinstance(laned, BaseException) and laned.status_code < 400:
             collect(laned.json(), lane=tag, lane_query=str(params["q"]))
     collect(general.json())
+    if site:
+        scoped_hits = []
+        for hit in hits:
+            try:
+                host = urlsplit(hit["url"]).hostname or ""
+            except ValueError:
+                continue
+            if host == site or host.endswith("." + site):
+                scoped_hits.append(hit)
+        hits = scoped_hits
     terms = _terms(query)
     # A `site:` lane answers with whatever the domain has; a hit sharing no
     # word with the question is that, not an answer.