918 lines
33 KiB
Python
918 lines
33 KiB
Python
#!/usr/bin/env python3
|
|
"""Probe whether an OpenAI-compatible model endpoint streams tool arguments.
|
|
|
|
This script talks directly to the configured model endpoint, bypassing
|
|
DeerFlow/LangGraph/the sandbox/frontend. It never prints the API key, prompt,
|
|
generated text, or tool-argument contents; only timing and byte/character
|
|
counts are reported.
|
|
|
|
Environment variables:
|
|
DF_LLM_URL Base URL ending in /v1, or the full /chat/completions URL.
|
|
DF_LLM_MODEL Model name sent to the endpoint.
|
|
DF_LLM_KEY Optional Bearer token. Keep it in the environment so it does
|
|
not appear in shell history or the process command line.
|
|
|
|
Examples:
|
|
python scripts/test-model-tool-stream.py --mode tool
|
|
python scripts/test-model-tool-stream.py --mode both --lines 120 \
|
|
--log-file ./model-stream-test.log
|
|
|
|
The default run executes both a normal text-stream control and a forced
|
|
write_file tool-call probe. Output is shown in the terminal and copied to a
|
|
timestamped log file. Exit code 0 means every requested probe showed genuine
|
|
time-separated streaming; 1 means a single block/burst/malformed result; 2+
|
|
means a request or script error. The complete RESULT_JSON lines are intended
|
|
for follow-up diagnosis.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import platform
|
|
import ssl
|
|
import sys
|
|
import time
|
|
from collections.abc import Iterator
|
|
from contextlib import redirect_stderr, redirect_stdout
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, TextIO
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
from urllib.request import Request, urlopen
|
|
|
|
SCRIPT_VERSION = "1.2"
|
|
MAX_NON_SSE_BYTES = 64 * 1024 * 1024
|
|
|
|
_SHORT_OUTCOME_CODES = {
|
|
"http_error": "HTTP_ERR",
|
|
"connection_error": "NET_ERR",
|
|
"upstream_error_event": "UP_ERR",
|
|
"text_progressive": "T_STREAM",
|
|
"text_missing": "T_NONE",
|
|
"text_single_or_burst": "T_BUFFER",
|
|
"forced_tool_ignored": "W_IGNORED",
|
|
"tool_arguments_missing": "W_NONE",
|
|
"tool_multiple_calls": "W_MULTI",
|
|
"tool_single_block": "W_ONE",
|
|
"tool_buffered_burst": "W_BURST",
|
|
"tool_progressive_invalid_json": "W_BADJSON",
|
|
"tool_progressive": "W_STREAM",
|
|
}
|
|
|
|
|
|
class _Tee:
|
|
"""Write console output to both the terminal and the diagnostic log."""
|
|
|
|
def __init__(self, *streams: TextIO) -> None:
|
|
self._streams = streams
|
|
|
|
def write(self, value: str) -> int:
|
|
for stream in self._streams:
|
|
stream.write(value)
|
|
stream.flush()
|
|
return len(value)
|
|
|
|
def flush(self) -> None:
|
|
for stream in self._streams:
|
|
stream.flush()
|
|
|
|
def isatty(self) -> bool:
|
|
return any(
|
|
getattr(stream, "isatty", lambda: False)() for stream in self._streams
|
|
)
|
|
|
|
|
|
def _chat_completions_url(value: str) -> str:
|
|
raw_url = value.strip()
|
|
if not raw_url:
|
|
raise ValueError("DF_LLM_URL/--url is required")
|
|
parsed = urlsplit(raw_url)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
|
raise ValueError("DF_LLM_URL/--url must be an absolute http(s) URL")
|
|
path = parsed.path.rstrip("/")
|
|
if not path.endswith("/chat/completions"):
|
|
path = f"{path}/chat/completions"
|
|
return urlunsplit((parsed.scheme, parsed.netloc, path, parsed.query, ""))
|
|
|
|
|
|
def _safe_url_for_log(value: str) -> str:
|
|
"""Remove credentials, query parameters, and fragments from log output."""
|
|
parsed = urlsplit(value)
|
|
hostname = parsed.hostname or ""
|
|
if ":" in hostname and not hostname.startswith("["):
|
|
hostname = f"[{hostname}]"
|
|
netloc = hostname
|
|
if parsed.port is not None:
|
|
netloc = f"{netloc}:{parsed.port}"
|
|
return urlunsplit((parsed.scheme, netloc, parsed.path, "", ""))
|
|
|
|
|
|
def _sse_events(response: Any) -> Iterator[tuple[str, str]]:
|
|
event_name = "message"
|
|
data_lines: list[str] = []
|
|
for raw_line in response:
|
|
line = raw_line.decode("utf-8", errors="replace").rstrip("\r\n")
|
|
if not line:
|
|
if data_lines:
|
|
yield event_name, "\n".join(data_lines)
|
|
event_name = "message"
|
|
data_lines = []
|
|
continue
|
|
if line.startswith(":"):
|
|
continue
|
|
if line.startswith("event:"):
|
|
event_name = line[6:].strip() or "message"
|
|
elif line.startswith("data:"):
|
|
data_lines.append(line[5:].lstrip())
|
|
if data_lines:
|
|
yield event_name, "\n".join(data_lines)
|
|
|
|
|
|
def _choices(payload: dict[str, Any]) -> list[dict[str, Any]]:
|
|
value = payload.get("choices")
|
|
if not isinstance(value, list):
|
|
nested = payload.get("chunk")
|
|
value = nested.get("choices") if isinstance(nested, dict) else None
|
|
return [item for item in value or [] if isinstance(item, dict)]
|
|
|
|
|
|
def _argument_text(value: Any) -> tuple[str, bool]:
|
|
if isinstance(value, str):
|
|
return value, False
|
|
if isinstance(value, (dict, list)):
|
|
return json.dumps(value, ensure_ascii=False, separators=(",", ":")), True
|
|
return "", False
|
|
|
|
|
|
@dataclass
|
|
class ProbeStats:
|
|
started_at: float
|
|
response_opened_at: float | None = None
|
|
finished_at: float | None = None
|
|
first_event_at: float | None = None
|
|
first_tool_at: float | None = None
|
|
last_tool_at: float | None = None
|
|
first_text_at: float | None = None
|
|
last_text_at: float | None = None
|
|
events: int = 0
|
|
tool_chunks: int = 0
|
|
text_chunks: int = 0
|
|
structured_tool_chunks: int = 0
|
|
argument_updates: int = 0
|
|
first_argument_part_chars: int = 0
|
|
last_argument_part_chars: int = 0
|
|
max_argument_part_chars: int = 0
|
|
malformed_events: int = 0
|
|
application_error_events: int = 0
|
|
finish_reasons: list[str] = field(default_factory=list)
|
|
arguments: str = ""
|
|
arguments_by_call: dict[str, str] = field(default_factory=dict)
|
|
first_argument_update_at: float | None = None
|
|
last_argument_update_at: float | None = None
|
|
|
|
def elapsed(self, instant: float | None) -> float:
|
|
return -1.0 if instant is None else instant - self.started_at
|
|
|
|
def add_arguments(
|
|
self,
|
|
part: str,
|
|
*,
|
|
call_key: str,
|
|
structured: bool,
|
|
now: float,
|
|
) -> tuple[str, float]:
|
|
previous_tool_at = self.last_tool_at
|
|
self.tool_chunks += 1
|
|
if self.first_argument_part_chars == 0:
|
|
self.first_argument_part_chars = len(part)
|
|
self.last_argument_part_chars = len(part)
|
|
self.max_argument_part_chars = max(self.max_argument_part_chars, len(part))
|
|
if structured:
|
|
self.structured_tool_chunks += 1
|
|
if self.first_tool_at is None:
|
|
self.first_tool_at = now
|
|
self.last_tool_at = now
|
|
|
|
previous = self.arguments_by_call.get(call_key, "")
|
|
if not previous:
|
|
style = "initial"
|
|
assembled = part
|
|
elif part.startswith(previous):
|
|
style = "snapshot"
|
|
assembled = part
|
|
elif previous.startswith(part):
|
|
style = "repeat/rewind"
|
|
assembled = previous
|
|
else:
|
|
style = "delta"
|
|
assembled = previous + part
|
|
|
|
self.arguments_by_call[call_key] = assembled
|
|
self.arguments = max(self.arguments_by_call.values(), key=len, default="")
|
|
if len(assembled) > len(previous):
|
|
self.argument_updates += 1
|
|
if self.first_argument_update_at is None:
|
|
self.first_argument_update_at = now
|
|
self.last_argument_update_at = now
|
|
gap_ms = -1.0 if previous_tool_at is None else (now - previous_tool_at) * 1000
|
|
return style, gap_ms
|
|
|
|
|
|
def _tool_call_key(tool_call: dict[str, Any], stats: ProbeStats) -> str:
|
|
index = tool_call.get("index")
|
|
if isinstance(index, int):
|
|
return f"index:{index}"
|
|
call_id = tool_call.get("id")
|
|
if isinstance(call_id, str) and call_id:
|
|
return f"id:{call_id}"
|
|
if len(stats.arguments_by_call) == 1:
|
|
return next(iter(stats.arguments_by_call))
|
|
return "default"
|
|
|
|
|
|
def _observe_payload(stats: ProbeStats, event: dict[str, Any], now: float) -> None:
|
|
if event.get("error"):
|
|
stats.application_error_events += 1
|
|
print(
|
|
f"UPSTREAM_ERROR_EVENT t={stats.elapsed(now):.3f}s "
|
|
f"error_type={type(event.get('error')).__name__} content=redacted"
|
|
)
|
|
for choice in _choices(event):
|
|
reason = choice.get("finish_reason")
|
|
if isinstance(reason, str) and reason:
|
|
stats.finish_reasons.append(reason)
|
|
print(f"FINISH t={stats.elapsed(now):.3f}s reason={reason}")
|
|
delta = choice.get("delta")
|
|
if not isinstance(delta, dict):
|
|
message = choice.get("message")
|
|
delta = message if isinstance(message, dict) else {}
|
|
|
|
content = delta.get("content")
|
|
if isinstance(content, str) and content:
|
|
previous_text_at = stats.last_text_at
|
|
stats.text_chunks += 1
|
|
if stats.first_text_at is None:
|
|
stats.first_text_at = now
|
|
stats.last_text_at = now
|
|
gap_ms = (
|
|
-1.0 if previous_text_at is None else (now - previous_text_at) * 1000
|
|
)
|
|
print(
|
|
f"TEXT #{stats.text_chunks:04d} "
|
|
f"t={stats.elapsed(now):8.3f}s gap_ms={gap_ms:8.1f} "
|
|
f"chars={len(content):5d}"
|
|
)
|
|
|
|
tool_calls = delta.get("tool_calls")
|
|
if isinstance(tool_calls, dict):
|
|
tool_calls = [tool_calls]
|
|
if not isinstance(tool_calls, list):
|
|
continue
|
|
for tool_call in tool_calls:
|
|
if not isinstance(tool_call, dict):
|
|
continue
|
|
function = tool_call.get("function")
|
|
if not isinstance(function, dict) or "arguments" not in function:
|
|
continue
|
|
part, structured = _argument_text(function.get("arguments"))
|
|
if not part:
|
|
continue
|
|
style, gap_ms = stats.add_arguments(
|
|
part,
|
|
call_key=_tool_call_key(tool_call, stats),
|
|
structured=structured,
|
|
now=now,
|
|
)
|
|
print(
|
|
f"TOOL #{stats.tool_chunks:04d} "
|
|
f"t={stats.elapsed(now):8.3f}s gap_ms={gap_ms:8.1f} "
|
|
f"chars={len(part):5d} "
|
|
f"assembled={len(stats.arguments):6d} style={style}"
|
|
)
|
|
|
|
|
|
def _safe_response_headers(response: Any) -> dict[str, str]:
|
|
allowed = (
|
|
"Content-Type",
|
|
"Transfer-Encoding",
|
|
"Content-Encoding",
|
|
"Cache-Control",
|
|
"X-Accel-Buffering",
|
|
"Connection",
|
|
"Server",
|
|
"Via",
|
|
"X-Request-Id",
|
|
"Date",
|
|
)
|
|
result: dict[str, str] = {}
|
|
for name in allowed:
|
|
value = response.headers.get(name)
|
|
if isinstance(value, str) and value:
|
|
result[name.lower()] = value[:300]
|
|
version = getattr(response, "version", None)
|
|
if version in {10, 11}:
|
|
result["http_version"] = "1.0" if version == 10 else "1.1"
|
|
elif version is not None:
|
|
result["http_version"] = str(version)[:30]
|
|
return result
|
|
|
|
|
|
def _emit_result(
|
|
*,
|
|
args: argparse.Namespace,
|
|
mode: str,
|
|
outcome: str,
|
|
verdict: str,
|
|
exit_code: int,
|
|
metrics: dict[str, Any],
|
|
) -> int:
|
|
print(f"VERDICT: {verdict}")
|
|
record = {
|
|
"record_type": "model_stream_probe_result",
|
|
"script_version": SCRIPT_VERSION,
|
|
"timestamp_utc": datetime.now(timezone.utc).isoformat(),
|
|
"mode": mode,
|
|
"outcome": outcome,
|
|
"exit_code": exit_code,
|
|
"endpoint": _safe_url_for_log(_chat_completions_url(args.url)),
|
|
"model": args.model,
|
|
**metrics,
|
|
}
|
|
print(
|
|
"RESULT_JSON="
|
|
+ json.dumps(record, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
)
|
|
response_headers = metrics.get("response_headers")
|
|
headers = response_headers if isinstance(response_headers, dict) else {}
|
|
content_type = str(headers.get("content-type", "")).lower()
|
|
finish_reasons = metrics.get("finish_reasons")
|
|
finish = (
|
|
finish_reasons[-1]
|
|
if isinstance(finish_reasons, list) and finish_reasons
|
|
else "none"
|
|
)
|
|
|
|
def compact(value: Any, default: str = "-") -> str:
|
|
if value is None or value == "":
|
|
return default
|
|
if isinstance(value, bool):
|
|
return "1" if value else "0"
|
|
text = str(value)
|
|
return "".join(
|
|
char if char.isalnum() or char in "._:/-" else "_" for char in text
|
|
)[:40]
|
|
|
|
short_fields = {
|
|
"cv": "1",
|
|
"sv": SCRIPT_VERSION,
|
|
"md": args.model,
|
|
"m": mode,
|
|
"c": _SHORT_OUTCOME_CODES.get(outcome, outcome),
|
|
"h": metrics.get("http_status"),
|
|
"sse": "1" if "text/event-stream" in content_type else "0",
|
|
"txt": metrics.get("text_chunks"),
|
|
"ts": metrics.get("text_span_ms"),
|
|
"tc": metrics.get("tool_arg_chunks"),
|
|
"au": metrics.get("argument_updates"),
|
|
"fa": metrics.get("first_argument_part_chars"),
|
|
"aa": metrics.get("assembled_argument_chars"),
|
|
"ft": metrics.get("first_tool_arg_ms"),
|
|
"us": metrics.get("argument_update_span_ms"),
|
|
"pw": metrics.get("preview_window_ms"),
|
|
"tt": metrics.get("total_ms"),
|
|
"js": metrics.get("valid_final_json"),
|
|
"ce": headers.get("content-encoding", "none"),
|
|
"te": headers.get("transfer-encoding", "none"),
|
|
"xb": headers.get("x-accel-buffering", "none"),
|
|
"fr": finish,
|
|
"ev": metrics.get("events"),
|
|
"bad": metrics.get("malformed_events"),
|
|
"ae": metrics.get("application_error_events"),
|
|
"et": metrics.get("error_type"),
|
|
"rt": metrics.get("reason_type"),
|
|
}
|
|
print(
|
|
"COPY_THIS "
|
|
+ " ".join(f"{key}={compact(value)}" for key, value in short_fields.items())
|
|
)
|
|
return exit_code
|
|
|
|
|
|
def _tool_schema() -> dict[str, Any]:
|
|
return {
|
|
"type": "function",
|
|
"function": {
|
|
"name": "write_file",
|
|
"description": "Write complete text content to a file.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"path": {"type": "string"},
|
|
"content": {"type": "string"},
|
|
},
|
|
"required": ["path", "content"],
|
|
"additionalProperties": False,
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def _request_payload(args: argparse.Namespace, mode: str) -> dict[str, Any]:
|
|
if mode == "tool":
|
|
user_prompt = (
|
|
"只调用一次 write_file，不要输出普通正文。将 Markdown 写到 "
|
|
"/mnt/user-data/outputs/model-stream-probe.md。Markdown 第一行为标题，"
|
|
f"随后生成 {args.lines} 行有实际内容的编号测试句；每行至少 30 个中文字符，"
|
|
"行与行内容不要完全相同。必须把全部 Markdown 放在 content 参数中。"
|
|
)
|
|
payload: dict[str, Any] = {
|
|
"model": args.model,
|
|
"messages": [
|
|
{"role": "system", "content": "严格遵循工具调用要求。"},
|
|
{"role": "user", "content": user_prompt},
|
|
],
|
|
"stream": True,
|
|
"temperature": 0,
|
|
"max_tokens": args.max_tokens,
|
|
"tools": [_tool_schema()],
|
|
"tool_choice": {
|
|
"type": "function",
|
|
"function": {"name": "write_file"},
|
|
},
|
|
}
|
|
else:
|
|
payload = {
|
|
"model": args.model,
|
|
"messages": [
|
|
{
|
|
"role": "user",
|
|
"content": (
|
|
f"逐行输出 {args.lines} 行编号测试句，每行至少 30 个中文字符。"
|
|
"不要调用工具，不要使用代码块。"
|
|
),
|
|
}
|
|
],
|
|
"stream": True,
|
|
"temperature": 0,
|
|
"max_tokens": args.max_tokens,
|
|
}
|
|
return payload
|
|
|
|
|
|
def _run_probe(args: argparse.Namespace, mode: str) -> int:
|
|
url = _chat_completions_url(args.url)
|
|
body = json.dumps(
|
|
_request_payload(args, mode), ensure_ascii=False, separators=(",", ":")
|
|
).encode("utf-8")
|
|
headers = {
|
|
"Accept": "text/event-stream",
|
|
"Accept-Encoding": "identity",
|
|
"Content-Type": "application/json; charset=utf-8",
|
|
"Cache-Control": "no-cache",
|
|
"User-Agent": "deerflow-tool-stream-probe/1.0",
|
|
}
|
|
api_key = os.environ.get("DF_LLM_KEY", "").strip()
|
|
if api_key:
|
|
headers["Authorization"] = f"Bearer {api_key}"
|
|
|
|
request = Request(url=url, data=body, headers=headers, method="POST")
|
|
context = (
|
|
ssl._create_unverified_context()
|
|
if args.insecure
|
|
else ssl.create_default_context()
|
|
)
|
|
stats = ProbeStats(started_at=time.monotonic())
|
|
response_headers: dict[str, str] = {}
|
|
http_status: int | str = "unknown"
|
|
|
|
print(f"\n=== {mode.upper()} probe ===")
|
|
print(f"endpoint={_safe_url_for_log(url)}")
|
|
print(f"model={args.model} lines={args.lines} max_tokens={args.max_tokens}")
|
|
try:
|
|
with urlopen(request, timeout=args.timeout, context=context) as response:
|
|
stats.response_opened_at = time.monotonic()
|
|
content_type = response.headers.get("Content-Type", "unknown")
|
|
http_status = getattr(response, "status", "unknown")
|
|
response_headers = _safe_response_headers(response)
|
|
response_open_ms = stats.elapsed(stats.response_opened_at) * 1000
|
|
print(
|
|
f"http_status={http_status} content_type={content_type} "
|
|
f"response_open_ms={response_open_ms:.0f}"
|
|
)
|
|
print(
|
|
"response_headers="
|
|
+ json.dumps(
|
|
response_headers,
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
)
|
|
)
|
|
if "text/event-stream" not in content_type.lower():
|
|
raw_payload = response.read(MAX_NON_SSE_BYTES + 1)
|
|
now = time.monotonic()
|
|
stats.events = 1
|
|
stats.first_event_at = now
|
|
if len(raw_payload) > MAX_NON_SSE_BYTES:
|
|
print("NON_SSE_RESPONSE_TOO_LARGE bytes_gt=67108864")
|
|
stats.malformed_events += 1
|
|
else:
|
|
try:
|
|
event = json.loads(
|
|
raw_payload.decode("utf-8", errors="replace")
|
|
)
|
|
except (UnicodeDecodeError, json.JSONDecodeError):
|
|
stats.malformed_events += 1
|
|
print(f"NON_JSON_RESPONSE chars={len(raw_payload)}")
|
|
else:
|
|
if isinstance(event, dict):
|
|
print("NON_SSE_JSON_RESPONSE buffered=true")
|
|
_observe_payload(stats, event, now)
|
|
else:
|
|
stats.malformed_events += 1
|
|
else:
|
|
for event_name, data in _sse_events(response):
|
|
now = time.monotonic()
|
|
stats.events += 1
|
|
if stats.first_event_at is None:
|
|
stats.first_event_at = now
|
|
if data.strip() == "[DONE]":
|
|
print(f"DONE t={stats.elapsed(now):.3f}s event={event_name}")
|
|
break
|
|
try:
|
|
event = json.loads(data)
|
|
except json.JSONDecodeError:
|
|
stats.malformed_events += 1
|
|
print(
|
|
f"NON_JSON t={stats.elapsed(now):.3f}s "
|
|
f"event={event_name} chars={len(data)}"
|
|
)
|
|
continue
|
|
if isinstance(event, dict):
|
|
_observe_payload(stats, event, now)
|
|
except HTTPError as exc:
|
|
error_body = exc.read(2000).decode("utf-8", errors="replace")
|
|
print(f"HTTP_ERROR status={exc.code} reason={exc.reason}", file=sys.stderr)
|
|
if error_body:
|
|
print(
|
|
f"HTTP_ERROR_BODY_REDACTED chars={len(error_body)}",
|
|
file=sys.stderr,
|
|
)
|
|
return _emit_result(
|
|
args=args,
|
|
mode=mode,
|
|
outcome="http_error",
|
|
verdict="The upstream endpoint rejected the request; inspect the HTTP status and model/tool compatibility.",
|
|
exit_code=2,
|
|
metrics={
|
|
"http_status": exc.code,
|
|
"http_error_body_chars": len(error_body),
|
|
},
|
|
)
|
|
except (URLError, TimeoutError, OSError) as exc:
|
|
reason = getattr(exc, "reason", None)
|
|
reason_type = type(reason).__name__ if reason is not None else "none"
|
|
print(
|
|
f"CONNECT_ERROR type={type(exc).__name__} reason_type={reason_type}",
|
|
file=sys.stderr,
|
|
)
|
|
return _emit_result(
|
|
args=args,
|
|
mode=mode,
|
|
outcome="connection_error",
|
|
verdict="The direct upstream request failed before a complete response was observed.",
|
|
exit_code=2,
|
|
metrics={
|
|
"error_type": type(exc).__name__,
|
|
"reason_type": reason_type,
|
|
},
|
|
)
|
|
finally:
|
|
stats.finished_at = time.monotonic()
|
|
|
|
tool_span = (
|
|
stats.last_tool_at - stats.first_tool_at
|
|
if stats.first_tool_at is not None and stats.last_tool_at is not None
|
|
else 0.0
|
|
)
|
|
update_span = (
|
|
stats.last_argument_update_at - stats.first_argument_update_at
|
|
if stats.first_argument_update_at is not None
|
|
and stats.last_argument_update_at is not None
|
|
else 0.0
|
|
)
|
|
text_span = (
|
|
stats.last_text_at - stats.first_text_at
|
|
if stats.first_text_at is not None and stats.last_text_at is not None
|
|
else 0.0
|
|
)
|
|
valid_json = False
|
|
if stats.arguments:
|
|
try:
|
|
parsed = json.loads(stats.arguments)
|
|
valid_json = isinstance(parsed, dict)
|
|
except (TypeError, ValueError):
|
|
pass
|
|
|
|
print("--- summary ---")
|
|
print(
|
|
"events={} malformed={} tool_chunks={} tool_span={:.3f}s "
|
|
"argument_updates={} update_span={:.3f}s tool_calls={} "
|
|
"assembled_chars={} valid_final_json={} structured_tool_chunks={} "
|
|
"text_chunks={} text_span={:.3f}s total={:.3f}s finish={}".format(
|
|
stats.events,
|
|
stats.malformed_events,
|
|
stats.tool_chunks,
|
|
tool_span,
|
|
stats.argument_updates,
|
|
update_span,
|
|
len(stats.arguments_by_call),
|
|
len(stats.arguments),
|
|
str(valid_json).lower(),
|
|
stats.structured_tool_chunks,
|
|
stats.text_chunks,
|
|
text_span,
|
|
stats.elapsed(stats.finished_at),
|
|
",".join(stats.finish_reasons) or "none",
|
|
)
|
|
)
|
|
|
|
def elapsed_ms(instant: float | None) -> int | None:
|
|
return round(stats.elapsed(instant) * 1000) if instant is not None else None
|
|
|
|
metrics: dict[str, Any] = {
|
|
"http_status": http_status,
|
|
"response_headers": response_headers,
|
|
"response_open_ms": elapsed_ms(stats.response_opened_at),
|
|
"first_event_ms": elapsed_ms(stats.first_event_at),
|
|
"first_tool_arg_ms": elapsed_ms(stats.first_tool_at),
|
|
"last_tool_arg_ms": elapsed_ms(stats.last_tool_at),
|
|
"tool_arg_span_ms": round(tool_span * 1000),
|
|
"argument_update_span_ms": round(update_span * 1000),
|
|
"preview_window_ms": (
|
|
round((stats.finished_at - stats.first_tool_at) * 1000)
|
|
if stats.finished_at is not None and stats.first_tool_at is not None
|
|
else -1
|
|
),
|
|
"first_text_ms": elapsed_ms(stats.first_text_at),
|
|
"text_span_ms": round(text_span * 1000),
|
|
"total_ms": round(stats.elapsed(stats.finished_at) * 1000),
|
|
"events": stats.events,
|
|
"malformed_events": stats.malformed_events,
|
|
"application_error_events": stats.application_error_events,
|
|
"tool_arg_chunks": stats.tool_chunks,
|
|
"argument_updates": stats.argument_updates,
|
|
"first_argument_part_chars": stats.first_argument_part_chars,
|
|
"last_argument_part_chars": stats.last_argument_part_chars,
|
|
"max_argument_part_chars": stats.max_argument_part_chars,
|
|
"tool_calls": len(stats.arguments_by_call),
|
|
"assembled_argument_chars": len(stats.arguments),
|
|
"valid_final_json": valid_json,
|
|
"structured_tool_chunks": stats.structured_tool_chunks,
|
|
"text_chunks": stats.text_chunks,
|
|
"finish_reasons": stats.finish_reasons,
|
|
}
|
|
|
|
if stats.application_error_events:
|
|
return _emit_result(
|
|
args=args,
|
|
mode=mode,
|
|
outcome="upstream_error_event",
|
|
verdict="The HTTP stream contained an upstream error event; its content was redacted from this log.",
|
|
exit_code=2,
|
|
metrics=metrics,
|
|
)
|
|
|
|
if mode == "text":
|
|
if stats.text_chunks > 1 and text_span >= args.live_span:
|
|
return _emit_result(
|
|
args=args,
|
|
mode=mode,
|
|
outcome="text_progressive",
|
|
verdict="Ordinary text is genuinely streaming from the upstream endpoint.",
|
|
exit_code=0,
|
|
metrics=metrics,
|
|
)
|
|
elif stats.text_chunks == 0:
|
|
return _emit_result(
|
|
args=args,
|
|
mode=mode,
|
|
outcome="text_missing",
|
|
verdict="No ordinary text deltas were observed.",
|
|
exit_code=1,
|
|
metrics=metrics,
|
|
)
|
|
else:
|
|
return _emit_result(
|
|
args=args,
|
|
mode=mode,
|
|
outcome="text_single_or_burst",
|
|
verdict="Text arrived in one chunk or a short burst; upstream/proxy buffering is possible.",
|
|
exit_code=1,
|
|
metrics=metrics,
|
|
)
|
|
|
|
if stats.tool_chunks == 0:
|
|
if stats.text_chunks:
|
|
verdict = (
|
|
"The upstream returned ordinary text instead of the forced tool call."
|
|
)
|
|
outcome = "forced_tool_ignored"
|
|
else:
|
|
verdict = "No tool-argument chunks were observed; inspect the raw model gateway response."
|
|
outcome = "tool_arguments_missing"
|
|
return _emit_result(
|
|
args=args,
|
|
mode=mode,
|
|
outcome=outcome,
|
|
verdict=verdict,
|
|
exit_code=1,
|
|
metrics=metrics,
|
|
)
|
|
if len(stats.arguments_by_call) != 1:
|
|
return _emit_result(
|
|
args=args,
|
|
mode=mode,
|
|
outcome="tool_multiple_calls",
|
|
verdict=(
|
|
"The endpoint produced multiple write_file calls even though exactly one was "
|
|
"requested. This run is not valid for measuring one call's progressive arguments."
|
|
),
|
|
exit_code=1,
|
|
metrics=metrics,
|
|
)
|
|
if stats.argument_updates <= 1:
|
|
return _emit_result(
|
|
args=args,
|
|
mode=mode,
|
|
outcome="tool_single_block",
|
|
verdict=(
|
|
"Tool arguments arrived as one complete block (or that block was merely repeated). "
|
|
"The model-serving/OpenAI-compatible gateway is not progressively streaming this "
|
|
"tool call; DeerFlow was bypassed."
|
|
),
|
|
exit_code=1,
|
|
metrics=metrics,
|
|
)
|
|
if update_span < args.live_span:
|
|
return _emit_result(
|
|
args=args,
|
|
mode=mode,
|
|
outcome="tool_buffered_burst",
|
|
verdict=(
|
|
"Multiple growing tool chunks arrived only as a short burst. A model gateway or "
|
|
"HTTP proxy may be buffering; this is not proof of live tool streaming."
|
|
),
|
|
exit_code=1,
|
|
metrics=metrics,
|
|
)
|
|
if not valid_json:
|
|
return _emit_result(
|
|
args=args,
|
|
mode=mode,
|
|
outcome="tool_progressive_invalid_json",
|
|
verdict=(
|
|
"Tool arguments streamed progressively, but the final assembled arguments were not "
|
|
"valid JSON. Check model output truncation and max_tokens before testing the UI."
|
|
),
|
|
exit_code=1,
|
|
metrics=metrics,
|
|
)
|
|
return _emit_result(
|
|
args=args,
|
|
mode=mode,
|
|
outcome="tool_progressive",
|
|
verdict=(
|
|
"Tool arguments arrived in multiple time-separated growing chunks. The upstream endpoint "
|
|
"can stream write_file arguments; if DeerFlow still updates only at the end, investigate "
|
|
"the backend SSE/proxy/frontend path."
|
|
),
|
|
exit_code=0,
|
|
metrics=metrics,
|
|
)
|
|
|
|
|
|
def _parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Directly test OpenAI-compatible text/tool-call streaming."
|
|
)
|
|
parser.add_argument("--url", default=os.environ.get("DF_LLM_URL", ""))
|
|
parser.add_argument("--model", default=os.environ.get("DF_LLM_MODEL", ""))
|
|
parser.add_argument("--mode", choices=("tool", "text", "both"), default="both")
|
|
parser.add_argument("--lines", type=int, default=80)
|
|
parser.add_argument("--max-tokens", type=int, default=8192)
|
|
parser.add_argument("--timeout", type=float, default=600.0)
|
|
parser.add_argument(
|
|
"--log-file",
|
|
default=os.environ.get("DF_LLM_LOG_FILE", ""),
|
|
help=(
|
|
"Diagnostic log path. Defaults to model-tool-stream-probe-<timestamp>.log "
|
|
"in the current directory."
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--live-span",
|
|
type=float,
|
|
default=0.20,
|
|
help="Minimum first-to-last chunk span considered genuinely live (seconds).",
|
|
)
|
|
parser.add_argument(
|
|
"--insecure",
|
|
action="store_true",
|
|
help="Disable TLS certificate verification for an internal self-signed endpoint.",
|
|
)
|
|
args = parser.parse_args()
|
|
if not args.model.strip():
|
|
parser.error("DF_LLM_MODEL/--model is required")
|
|
if not args.url.strip():
|
|
parser.error("DF_LLM_URL/--url is required")
|
|
if args.lines < 1 or args.max_tokens < 1 or args.timeout <= 0 or args.live_span < 0:
|
|
parser.error("numeric arguments must be positive")
|
|
try:
|
|
_chat_completions_url(args.url)
|
|
except ValueError as exc:
|
|
parser.error(str(exc))
|
|
if not args.log_file.strip():
|
|
timestamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
|
|
args.log_file = f"model-tool-stream-probe-{timestamp}.log"
|
|
return args
|
|
|
|
|
|
def main() -> int:
|
|
args = _parse_args()
|
|
log_path = Path(args.log_file).expanduser().resolve()
|
|
try:
|
|
log_path.parent.mkdir(parents=True, exist_ok=True)
|
|
log_stream = log_path.open("a", encoding="utf-8", buffering=1)
|
|
except OSError as exc:
|
|
print(
|
|
f"LOG_OPEN_ERROR path={log_path} type={type(exc).__name__}",
|
|
file=sys.stderr,
|
|
)
|
|
return 2
|
|
|
|
console_out = sys.stdout
|
|
console_err = sys.stderr
|
|
with (
|
|
log_stream,
|
|
redirect_stdout(_Tee(console_out, log_stream)),
|
|
redirect_stderr(_Tee(console_err, log_stream)),
|
|
):
|
|
started_local = datetime.now().astimezone()
|
|
print("=" * 78)
|
|
print("DEERFLOW DIRECT MODEL STREAM PROBE")
|
|
print(f"script_version={SCRIPT_VERSION}")
|
|
print(f"started_at_local={started_local.isoformat()}")
|
|
print(f"started_at_utc={started_local.astimezone(timezone.utc).isoformat()}")
|
|
print(
|
|
f"python={platform.python_version()} implementation={platform.python_implementation()}"
|
|
)
|
|
print(f"platform={platform.platform()}")
|
|
print(f"log_file={log_path}")
|
|
print(f"endpoint={_safe_url_for_log(_chat_completions_url(args.url))}")
|
|
print(f"model={args.model}")
|
|
print(
|
|
f"mode={args.mode} lines={args.lines} max_tokens={args.max_tokens} "
|
|
f"timeout={args.timeout:.1f}s live_span={args.live_span:.3f}s"
|
|
)
|
|
print(
|
|
f"api_key_present={str(bool(os.environ.get('DF_LLM_KEY', '').strip())).lower()} "
|
|
f"tls_verification={str(not args.insecure).lower()}"
|
|
)
|
|
print(
|
|
"privacy=API key, prompts, generated text, and tool-argument content are not logged"
|
|
)
|
|
print("=" * 78)
|
|
|
|
modes = ("text", "tool") if args.mode == "both" else (args.mode,)
|
|
try:
|
|
results = [_run_probe(args, mode) for mode in modes]
|
|
exit_code = max(results, default=0)
|
|
except KeyboardInterrupt:
|
|
print("PROBE_INTERRUPTED by_user=true", file=sys.stderr)
|
|
exit_code = 130
|
|
except Exception as exc: # noqa: BLE001 - top-level probe must leave a redacted log
|
|
print(
|
|
f"UNEXPECTED_PROBE_ERROR type={type(exc).__name__} details=redacted",
|
|
file=sys.stderr,
|
|
)
|
|
exit_code = 3
|
|
|
|
print("=" * 78)
|
|
print(f"finished_at_local={datetime.now().astimezone().isoformat()}")
|
|
print(f"overall_exit_code={exit_code}")
|
|
print(f"log_file={log_path}")
|
|
print(
|
|
"If you cannot send the log file, type only the two COPY_THIS lines above."
|
|
)
|
|
print("=" * 78)
|
|
return exit_code
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|