mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-06 04:48:44 +00:00
* fix(gateway): unify thread ID validation at the API boundary
Thread ID entry points accepted arbitrary strings while downstream
consumers (filesystem paths, Kubernetes Provisioner, JSONL event store)
each enforced different character restrictions, so invalid IDs were
persisted first and only failed later during sandbox/workspace init.
Centralize validation in deerflow.utils.thread_id (pattern
^[A-Za-z0-9_-]{1,64}$): validate at routers, RunCreateRequest,
scheduler dispatch, paths.py, JSONL store, embedded client, and align
the Provisioner pattern (pinned by a parity test). UUIDs are still
generated only when no ID is supplied; caller-supplied opaque IDs stay
supported.
Deliberate exceptions: DELETE /threads/{id} keeps str as the legacy
cleanup escape hatch (filesystem cleanup guarded), read-only
client.get_thread stays unvalidated, and scheduler rows with legacy
invalid IDs record a failed dispatch instead of raising out of the
poll loop.
* docs: document canonical thread ID contract
README: caller-supplied thread IDs need not be UUIDs; the canonical
pattern and per-endpoint behavior. AGENTS.md: the shared
deerflow.utils.thread_id contract, its enforcement boundaries, and the
legacy-ID escape hatches.
* fix(gateway): close thread ID validation gaps at remaining entry points
Follow-up to the canonical thread ID contract: a full audit found the
uniform-422 coverage only reached about half of the thread_id surfaces.
- routers: 18 routes still took a bare thread_id: str — 13 in
thread_runs.py (including the five messages/events/workspace-changes
reads that returned 500 on the JSONL event store vs 404/empty on the
DB store), 4 read routes in threads.py, and the suggestions route
flagged in review. DELETE /api/threads/{id} keeps str as the declared
legacy-cleanup escape hatch.
- client: upload_files/delete_upload/list_uploads/get_artifact now
validate up front, fulfilling the RFC's 'all mutating entry points'
clause (get_thread stays unvalidated as the declared legacy read path).
- tui: the /resume literal-ref fallback validates against the canonical
contract and reports a descriptive error instead of failing deep in
the client.
- scripts/support_bundle.py: replace the drifted dot-allowing pattern
with a byte-identical copy of THREAD_ID_PATTERN (kept local so the
script still runs with a broken venv).
* test(gateway): guard the canonical thread ID contract against regressions
- test_thread_id_route_contract.py: static AST sweep asserting every
route handler with a thread_id parameter annotates ThreadId
(whitelist: the DELETE escape hatch), plus a runtime sweep hitting
all 44 thread_id routes with a non-canonical ID and asserting a 422
that names thread_id, plus a websocket upgrade-rejection case.
- test_thread_id_validation.py: client entry-point validation,
support_bundle pattern parity, and TUI literal-ref fallback tests.
- Align two tests that encoded the old contract (dotted IDs).
149 lines
5.5 KiB
Python
149 lines
5.5 KiB
Python
import json
|
|
import logging
|
|
|
|
from fastapi import APIRouter, Depends, Request
|
|
from pydantic import BaseModel, Field
|
|
|
|
import deerflow.utils.llm_text as llm_text
|
|
from app.gateway.authz import require_permission
|
|
from app.gateway.deps import get_config
|
|
from deerflow.config.app_config import AppConfig
|
|
from deerflow.config.suggestions_config import DEFAULT_MAX_SUGGESTIONS, MAX_SUGGESTIONS_LIMIT
|
|
from deerflow.utils.oneshot_llm import run_oneshot_llm
|
|
from deerflow.utils.thread_id import ThreadId
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/api", tags=["suggestions"])
|
|
|
|
|
|
class SuggestionMessage(BaseModel):
|
|
role: str = Field(..., description="Message role: user|assistant")
|
|
content: str = Field(..., description="Message content as plain text")
|
|
|
|
|
|
class SuggestionsRequest(BaseModel):
|
|
messages: list[SuggestionMessage] = Field(..., description="Recent conversation messages")
|
|
n: int = Field(default=DEFAULT_MAX_SUGGESTIONS, ge=1, le=MAX_SUGGESTIONS_LIMIT, description="Number of suggestions to generate")
|
|
model_name: str | None = Field(default=None, description="Optional model override")
|
|
|
|
|
|
class SuggestionsResponse(BaseModel):
|
|
suggestions: list[str] = Field(default_factory=list, description="Suggested follow-up questions")
|
|
|
|
|
|
class SuggestionsConfigResponse(BaseModel):
|
|
enabled: bool = Field(..., description="Whether follow-up suggestions are enabled globally")
|
|
max_suggestions: int = Field(..., ge=1, le=MAX_SUGGESTIONS_LIMIT, description="Maximum number of follow-up suggestions to generate")
|
|
|
|
|
|
_strip_markdown_code_fence = llm_text.strip_markdown_code_fence
|
|
_strip_think_blocks = llm_text.strip_think_blocks
|
|
|
|
|
|
def _parse_json_string_list(text: str) -> list[str] | None:
|
|
candidate = _strip_think_blocks(text)
|
|
candidate = _strip_markdown_code_fence(candidate)
|
|
start = candidate.find("[")
|
|
end = candidate.rfind("]")
|
|
if start == -1 or end == -1 or end <= start:
|
|
return None
|
|
candidate = candidate[start : end + 1]
|
|
try:
|
|
data = json.loads(candidate)
|
|
except Exception:
|
|
return None
|
|
if not isinstance(data, list):
|
|
return None
|
|
out: list[str] = []
|
|
for item in data:
|
|
if not isinstance(item, str):
|
|
continue
|
|
s = item.strip()
|
|
if not s:
|
|
continue
|
|
out.append(s)
|
|
return out
|
|
|
|
|
|
def _format_conversation(messages: list[SuggestionMessage]) -> str:
|
|
parts: list[str] = []
|
|
for m in messages:
|
|
role = m.role.strip().lower()
|
|
if role in ("user", "human"):
|
|
parts.append(f"User: {m.content.strip()}")
|
|
elif role in ("assistant", "ai"):
|
|
parts.append(f"Assistant: {m.content.strip()}")
|
|
else:
|
|
parts.append(f"{m.role}: {m.content.strip()}")
|
|
return "\n".join(parts).strip()
|
|
|
|
|
|
def _configured_max_suggestions(config: AppConfig) -> int:
|
|
return getattr(config.suggestions, "max_suggestions", DEFAULT_MAX_SUGGESTIONS)
|
|
|
|
|
|
@router.get(
|
|
"/suggestions/config",
|
|
response_model=SuggestionsConfigResponse,
|
|
summary="Get Suggestions Configuration",
|
|
description="Returns the global configuration for follow-up suggestions.",
|
|
)
|
|
async def get_suggestions_config(
|
|
config: AppConfig = Depends(get_config),
|
|
) -> SuggestionsConfigResponse:
|
|
return SuggestionsConfigResponse(enabled=config.suggestions.enabled, max_suggestions=_configured_max_suggestions(config))
|
|
|
|
|
|
@router.post(
|
|
"/threads/{thread_id}/suggestions",
|
|
response_model=SuggestionsResponse,
|
|
summary="Generate Follow-up Questions",
|
|
description="Generate short follow-up questions a user might ask next, based on recent conversation context.",
|
|
)
|
|
@require_permission("threads", "read", owner_check=True)
|
|
async def generate_suggestions(
|
|
thread_id: ThreadId,
|
|
body: SuggestionsRequest,
|
|
request: Request,
|
|
config: AppConfig = Depends(get_config),
|
|
) -> SuggestionsResponse:
|
|
if not config.suggestions.enabled:
|
|
return SuggestionsResponse(suggestions=[])
|
|
if not body.messages:
|
|
return SuggestionsResponse(suggestions=[])
|
|
|
|
n = min(body.n, _configured_max_suggestions(config))
|
|
conversation = _format_conversation(body.messages)
|
|
if not conversation:
|
|
return SuggestionsResponse(suggestions=[])
|
|
|
|
system_instruction = (
|
|
"You are generating follow-up questions to help the user continue the conversation.\n"
|
|
f"Based on the conversation below, produce EXACTLY {n} short questions the user might ask next.\n"
|
|
"Requirements:\n"
|
|
"- Questions must be relevant to the preceding conversation.\n"
|
|
"- Questions must be written in the same language as the user.\n"
|
|
"- Keep each question concise (ideally <= 20 words / <= 40 Chinese characters).\n"
|
|
"- Do NOT include numbering, markdown, or any extra text.\n"
|
|
"- Output MUST be a JSON array of strings only.\n"
|
|
)
|
|
user_content = f"Conversation Context:\n{conversation}\n\nGenerate {n} follow-up questions"
|
|
|
|
try:
|
|
raw = await run_oneshot_llm(
|
|
system_instruction=system_instruction,
|
|
user_content=user_content,
|
|
run_name="suggest_agent",
|
|
app_config=config,
|
|
model_name=body.model_name,
|
|
thread_id=thread_id,
|
|
)
|
|
suggestions = _parse_json_string_list(raw) or []
|
|
cleaned = [s.replace("\n", " ").strip() for s in suggestions if s.strip()]
|
|
cleaned = cleaned[:n]
|
|
return SuggestionsResponse(suggestions=cleaned)
|
|
except Exception as exc:
|
|
logger.exception("Failed to generate suggestions: thread_id=%s err=%s", thread_id, exc)
|
|
return SuggestionsResponse(suggestions=[])
|