491 lines
16 KiB
Python
491 lines
16 KiB
Python
"""Keep one day of the exact text sent to a model and the text it sent back.
|
|
|
|
The admin page reads this store. Rows older than a day are dropped. Bodies are
|
|
capped so a long tool argument cannot grow the file without limit. Recording
|
|
never raises into the model call.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextvars
|
|
import json
|
|
import logging
|
|
import os
|
|
import tempfile
|
|
import threading
|
|
import uuid
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_RETENTION = timedelta(days=1)
|
|
_STORE_NAME = "model-exchanges.jsonl"
|
|
_MAX_RECORDS = 100
|
|
_MAX_FIELD_CHARS = 200_000
|
|
_PREVIEW_CHARS = 80
|
|
_TRUNCATED = "\n…（超出保存长度，后面已截断）"
|
|
|
|
_LOCK = threading.Lock()
|
|
_ACTIVE: contextvars.ContextVar[ExchangeCapture | None] = contextvars.ContextVar(
|
|
"deerflow_model_exchange",
|
|
default=None,
|
|
)
|
|
_store_override: Path | None = None
|
|
|
|
|
|
@dataclass
|
|
class ExchangeCapture:
|
|
"""One in-flight model call. Text stays here until the call finishes."""
|
|
|
|
model: str = ""
|
|
enable_thinking: str = "unset"
|
|
request_text: str = ""
|
|
request_body: str = ""
|
|
request_preview: str = ""
|
|
content_parts: list[str] = field(default_factory=list)
|
|
reasoning_by_field: dict[str, list[str]] = field(default_factory=dict)
|
|
reasoning_fields: list[str] = field(default_factory=list)
|
|
raw_chunks: list[str] = field(default_factory=list)
|
|
owned: bool = False
|
|
token: contextvars.Token | None = None
|
|
saved: bool = False
|
|
|
|
|
|
def use_exchange_store(path: Path | None) -> None:
|
|
"""Point the store at a test file. ``None`` restores the runtime data dir."""
|
|
global _store_override
|
|
_store_override = path
|
|
|
|
|
|
def open_exchange() -> ExchangeCapture:
|
|
return ExchangeCapture()
|
|
|
|
|
|
def bind_exchange(capture: ExchangeCapture) -> contextvars.Token:
|
|
return _ACTIVE.set(capture)
|
|
|
|
|
|
def unbind_exchange(token: contextvars.Token) -> None:
|
|
_ACTIVE.reset(token)
|
|
|
|
|
|
def activate_exchange() -> tuple[ExchangeCapture, contextvars.Token]:
|
|
capture = open_exchange()
|
|
return capture, bind_exchange(capture)
|
|
|
|
|
|
def deactivate_exchange(capture: ExchangeCapture, token: contextvars.Token) -> None:
|
|
try:
|
|
finish_exchange(capture)
|
|
finally:
|
|
unbind_exchange(token)
|
|
|
|
|
|
def remember_request(payload: Mapping[str, Any]) -> None:
|
|
"""Copy the outbound messages once, after the payload is final."""
|
|
try:
|
|
capture = _ACTIVE.get()
|
|
if capture is None:
|
|
capture = ExchangeCapture(owned=True)
|
|
capture.token = _ACTIVE.set(capture)
|
|
elif capture.owned and capture.request_text and not capture.saved:
|
|
_discard_owned(capture)
|
|
capture = ExchangeCapture(owned=True)
|
|
capture.token = _ACTIVE.set(capture)
|
|
if capture.request_text:
|
|
return
|
|
messages = payload.get("messages")
|
|
capture.model = str(payload.get("model") or "")
|
|
capture.enable_thinking = _thinking_flag(payload)
|
|
capture.request_preview = _preview(_last_user_text(messages))
|
|
capture.request_text = _request_text(payload, messages)
|
|
capture.request_body = _replay_request(payload)
|
|
except Exception:
|
|
logger.debug("model exchange request was not captured", exc_info=True)
|
|
|
|
|
|
def remember_provider_chunk(chunk: Mapping[str, Any]) -> None:
|
|
"""Append one provider chunk exactly as it arrived."""
|
|
try:
|
|
capture = _ACTIVE.get()
|
|
if capture is None:
|
|
return
|
|
_store_raw(capture, chunk)
|
|
for choice in _choices(chunk):
|
|
delta = choice.get("delta") if isinstance(choice.get("delta"), Mapping) else {}
|
|
message = choice.get("message") if isinstance(choice.get("message"), Mapping) else {}
|
|
_append_channel(capture.content_parts, _text_field(delta, "content") or _text_field(message, "content"))
|
|
_take_reasoning(capture, delta)
|
|
_take_reasoning(capture, message)
|
|
except Exception:
|
|
logger.debug("model exchange chunk was not captured", exc_info=True)
|
|
|
|
|
|
def remember_raw_response(payload: Mapping[str, Any]) -> None:
|
|
"""Keep a non-streaming response body exactly as the provider sent it."""
|
|
try:
|
|
capture = _ACTIVE.get()
|
|
if capture is None:
|
|
return
|
|
_store_raw(capture, payload, only_if_empty=True)
|
|
except Exception:
|
|
logger.debug("model exchange response was not captured", exc_info=True)
|
|
|
|
|
|
def remember_final_message(message: Mapping[str, Any]) -> None:
|
|
"""Record a non-streaming message when no chunks were captured."""
|
|
try:
|
|
capture = _ACTIVE.get()
|
|
if capture is None:
|
|
return
|
|
_store_raw(capture, message, only_if_empty=True)
|
|
if capture.content_parts or _has_reasoning(capture):
|
|
return
|
|
_append_channel(capture.content_parts, _text_field(message, "content"))
|
|
_take_reasoning(capture, message)
|
|
except Exception:
|
|
logger.debug("model exchange message was not captured", exc_info=True)
|
|
|
|
|
|
def finish_owned_exchange() -> None:
|
|
capture = _ACTIVE.get()
|
|
if capture is None or not capture.owned:
|
|
return
|
|
token = capture.token
|
|
try:
|
|
finish_exchange(capture)
|
|
finally:
|
|
if token is not None:
|
|
_ACTIVE.reset(token)
|
|
|
|
|
|
def finish_exchange(capture: ExchangeCapture) -> None:
|
|
if capture.saved or not capture.request_text:
|
|
capture.saved = True
|
|
return
|
|
capture.saved = True
|
|
content = "".join(capture.content_parts)
|
|
reasoning_by_field = {
|
|
name: "".join(capture.reasoning_by_field.get(name, []))
|
|
for name in capture.reasoning_fields
|
|
if "".join(capture.reasoning_by_field.get(name, []))
|
|
}
|
|
reasoning = "".join(reasoning_by_field.values())
|
|
reasoning_field = "、".join(reasoning_by_field) or "（没有单独的思考字段）"
|
|
response_raw = _cap("\n\n".join(capture.raw_chunks))
|
|
preview_source = content or reasoning
|
|
record = {
|
|
"id": uuid.uuid4().hex,
|
|
"at": datetime.now(UTC).isoformat(timespec="seconds"),
|
|
"model": capture.model,
|
|
"user_id": _current_user_id(),
|
|
"enable_thinking": capture.enable_thinking,
|
|
"request_preview": capture.request_preview,
|
|
"response_preview": _preview(preview_source),
|
|
"request_text": _cap(capture.request_text),
|
|
"request_body": capture.request_body,
|
|
"response_text": response_raw,
|
|
"response_raw": response_raw,
|
|
"response_content": _cap(content),
|
|
"response_reasoning": _cap(reasoning),
|
|
"reasoning_field": reasoning_field,
|
|
}
|
|
_append_record(record)
|
|
|
|
|
|
def query_model_exchanges() -> list[dict[str, Any]]:
|
|
"""Newest first. Entries older than one day are dropped."""
|
|
with _LOCK:
|
|
records = _read_records()
|
|
_write_records(records)
|
|
return list(reversed(records))
|
|
|
|
|
|
def _discard_owned(capture: ExchangeCapture) -> None:
|
|
capture.saved = True
|
|
token = capture.token
|
|
if token is not None:
|
|
_ACTIVE.reset(token)
|
|
|
|
|
|
def _store_path() -> Path:
|
|
if _store_override is not None:
|
|
return _store_override
|
|
from deerflow.config.paths import get_paths
|
|
|
|
return get_paths().base_dir / _STORE_NAME
|
|
|
|
|
|
def _append_record(record: dict[str, Any]) -> None:
|
|
try:
|
|
with _LOCK:
|
|
records = _read_records()
|
|
records.append(record)
|
|
if len(records) > _MAX_RECORDS:
|
|
records = records[-_MAX_RECORDS:]
|
|
_write_records(records)
|
|
except Exception:
|
|
logger.debug("model exchange was not saved", exc_info=True)
|
|
|
|
|
|
def _read_records() -> list[dict[str, Any]]:
|
|
path = _store_path()
|
|
if not path.is_file():
|
|
return []
|
|
cutoff = datetime.now(UTC) - _RETENTION
|
|
kept: list[dict[str, Any]] = []
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
if not line.strip():
|
|
continue
|
|
try:
|
|
record = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if isinstance(record, dict) and _record_time(record) >= cutoff:
|
|
kept.append(record)
|
|
if len(kept) > _MAX_RECORDS:
|
|
kept = kept[-_MAX_RECORDS:]
|
|
return kept
|
|
|
|
|
|
def _write_records(records: list[dict[str, Any]]) -> None:
|
|
path = _store_path()
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=".model-exchanges.", suffix=".tmp")
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
|
for record in records:
|
|
handle.write(json.dumps(record, ensure_ascii=False) + "\n")
|
|
os.replace(tmp, path)
|
|
except Exception:
|
|
try:
|
|
os.unlink(tmp)
|
|
except OSError:
|
|
pass
|
|
raise
|
|
|
|
|
|
def _record_time(record: Mapping[str, Any]) -> datetime:
|
|
raw = str(record.get("at") or "")
|
|
try:
|
|
parsed = datetime.fromisoformat(raw)
|
|
except ValueError:
|
|
return datetime.min.replace(tzinfo=UTC)
|
|
if parsed.tzinfo is None:
|
|
return parsed.replace(tzinfo=UTC)
|
|
return parsed.astimezone(UTC)
|
|
|
|
|
|
def _current_user_id() -> str:
|
|
try:
|
|
from deerflow.runtime.user_context import get_effective_user_id
|
|
|
|
return get_effective_user_id()
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def _thinking_flag(payload: Mapping[str, Any]) -> str:
|
|
extra_body = payload.get("extra_body")
|
|
value = extra_body.get("enable_thinking", payload.get("enable_thinking")) if isinstance(extra_body, Mapping) else payload.get("enable_thinking")
|
|
if value is True:
|
|
return "true"
|
|
if value is False:
|
|
return "false"
|
|
if value is None:
|
|
return "unset"
|
|
return "other"
|
|
|
|
|
|
def _choices(chunk: Mapping[str, Any]) -> list[Mapping[str, Any]]:
|
|
choices = chunk.get("choices")
|
|
if not isinstance(choices, list) or not choices:
|
|
nested = chunk.get("chunk")
|
|
if isinstance(nested, Mapping):
|
|
choices = nested.get("choices")
|
|
if not isinstance(choices, list):
|
|
return []
|
|
return [choice for choice in choices if isinstance(choice, Mapping)]
|
|
|
|
|
|
def _has_reasoning(capture: ExchangeCapture) -> bool:
|
|
return any(parts for parts in capture.reasoning_by_field.values())
|
|
|
|
|
|
def _take_reasoning(capture: ExchangeCapture, source: Mapping[str, Any]) -> None:
|
|
"""Keep thinking text under the wire field that actually carried it."""
|
|
for name in ("reasoning_content", "reasoning"):
|
|
text = _text_field(source, name)
|
|
if not text:
|
|
continue
|
|
if name not in capture.reasoning_fields:
|
|
capture.reasoning_fields.append(name)
|
|
_append_channel(capture.reasoning_by_field.setdefault(name, []), text)
|
|
|
|
|
|
def _text_field(payload: Mapping[str, Any], key: str) -> str:
|
|
value = payload.get(key)
|
|
if isinstance(value, str):
|
|
return value
|
|
return _content_to_text(value) if key == "content" else ""
|
|
|
|
|
|
def _content_to_text(content: Any) -> str:
|
|
if isinstance(content, str):
|
|
return content
|
|
if isinstance(content, list):
|
|
parts: list[str] = []
|
|
for item in content:
|
|
if isinstance(item, str):
|
|
parts.append(item)
|
|
elif isinstance(item, Mapping):
|
|
text = item.get("text")
|
|
if isinstance(text, str) and text:
|
|
parts.append(text)
|
|
elif item.get("type") == "image_url":
|
|
parts.append("[image]")
|
|
return "\n".join(parts)
|
|
if content is None:
|
|
return ""
|
|
return str(content)
|
|
|
|
|
|
def _last_user_text(messages: Any) -> str:
|
|
if not isinstance(messages, list):
|
|
return ""
|
|
for message in reversed(messages):
|
|
if isinstance(message, Mapping) and message.get("role") == "user":
|
|
return _content_to_text(message.get("content"))
|
|
return ""
|
|
|
|
|
|
_SECRET_KEYS = frozenset({"api_key", "openai_api_key", "authorization", "token"})
|
|
|
|
|
|
def _replay_request(payload: Mapping[str, Any]) -> str:
|
|
"""JSON body the OpenAI client actually posts. ``extra_body`` is flattened."""
|
|
ready = _json_ready(payload)
|
|
if not isinstance(ready, dict):
|
|
return "{}"
|
|
extra = ready.pop("extra_body", None)
|
|
if isinstance(extra, dict):
|
|
ready.update(extra)
|
|
return _fit_replay(ready)
|
|
|
|
|
|
def _store_raw(capture: ExchangeCapture, payload: Mapping[str, Any], *, only_if_empty: bool = False) -> None:
|
|
if only_if_empty and capture.raw_chunks:
|
|
return
|
|
if sum(len(part) for part in capture.raw_chunks) >= _MAX_FIELD_CHARS:
|
|
return
|
|
capture.raw_chunks.append(json.dumps(payload, ensure_ascii=False, indent=2, default=str))
|
|
|
|
|
|
def _json_ready(value: Any) -> Any:
|
|
if value is None or isinstance(value, (str, int, float, bool)):
|
|
return value
|
|
if isinstance(value, Mapping):
|
|
return {
|
|
str(key): _json_ready(item)
|
|
for key, item in value.items()
|
|
if item is not None and str(key) not in _SECRET_KEYS
|
|
}
|
|
if isinstance(value, (list, tuple)):
|
|
return [_json_ready(item) for item in value]
|
|
return str(value)
|
|
|
|
|
|
def _fit_replay(body: dict[str, Any]) -> str:
|
|
encoded = json.dumps(body, ensure_ascii=False, indent=2)
|
|
if len(encoded) <= _MAX_FIELD_CHARS:
|
|
return encoded
|
|
tools = body.get("tools")
|
|
if isinstance(tools, list) and tools:
|
|
names: list[str] = []
|
|
for tool in tools:
|
|
function = tool.get("function") if isinstance(tool, dict) else None
|
|
name = function.get("name") if isinstance(function, dict) else None
|
|
if isinstance(name, str) and name and name not in names:
|
|
names.append(name)
|
|
body["tools"] = [{"_omitted": "工具定义过长，这里只保留工具名", "names": names}]
|
|
encoded = json.dumps(body, ensure_ascii=False, indent=2)
|
|
if len(encoded) <= _MAX_FIELD_CHARS:
|
|
return encoded
|
|
messages = body.get("messages")
|
|
if isinstance(messages, list):
|
|
for _ in range(12):
|
|
if len(encoded) <= _MAX_FIELD_CHARS:
|
|
break
|
|
longest: dict[str, Any] | None = None
|
|
longest_len = 0
|
|
for message in messages:
|
|
if not isinstance(message, dict):
|
|
continue
|
|
content = message.get("content")
|
|
if isinstance(content, str) and len(content) > longest_len:
|
|
longest = message
|
|
longest_len = len(content)
|
|
if longest is None or longest_len < 200:
|
|
break
|
|
longest["content"] = longest["content"][: max(80, longest_len // 2)] + _TRUNCATED.strip()
|
|
encoded = json.dumps(body, ensure_ascii=False, indent=2)
|
|
if len(encoded) > _MAX_FIELD_CHARS:
|
|
body["messages"] = [{"role": "user", "content": _TRUNCATED.strip()}]
|
|
encoded = json.dumps(body, ensure_ascii=False, indent=2)
|
|
return encoded
|
|
|
|
|
|
def _request_text(payload: Mapping[str, Any], messages: Any) -> str:
|
|
tool_names: list[str] = []
|
|
tools = payload.get("tools")
|
|
if isinstance(tools, list):
|
|
for tool in tools:
|
|
function = tool.get("function") if isinstance(tool, Mapping) else None
|
|
name = function.get("name") if isinstance(function, Mapping) else None
|
|
if isinstance(name, str) and name and name not in tool_names:
|
|
tool_names.append(name)
|
|
lines = [
|
|
f"enable_thinking: {_thinking_flag(payload)}",
|
|
f"tools: {', '.join(tool_names) if tool_names else '（无）'}",
|
|
"",
|
|
]
|
|
if isinstance(messages, list):
|
|
for message in messages:
|
|
if not isinstance(message, Mapping):
|
|
continue
|
|
role = str(message.get("role") or "unknown")
|
|
lines.append(f"[{role}]")
|
|
lines.append(_content_to_text(message.get("content")))
|
|
lines.append("")
|
|
return "\n".join(lines).rstrip()
|
|
|
|
|
|
def _append_channel(parts: list[str], incoming: str) -> None:
|
|
if not incoming:
|
|
return
|
|
current = "".join(parts)
|
|
if current and (incoming == current or incoming.startswith(current)):
|
|
parts.clear()
|
|
parts.append(incoming)
|
|
return
|
|
if current and current.endswith(incoming):
|
|
return
|
|
parts.append(incoming)
|
|
|
|
|
|
def _preview(text: str) -> str:
|
|
compact = " ".join(text.split())
|
|
if len(compact) <= _PREVIEW_CHARS:
|
|
return compact
|
|
return compact[: _PREVIEW_CHARS - 1] + "…"
|
|
|
|
|
|
def _cap(text: str) -> str:
|
|
if len(text) <= _MAX_FIELD_CHARS:
|
|
return text
|
|
return text[: _MAX_FIELD_CHARS - len(_TRUNCATED)] + _TRUNCATED
|