| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152 |
- """工具调用日志的通用封装。"""
- from __future__ import annotations
- import json
- from typing import Any, Dict
- from .log_capture import log, log_fold
- def _pretty_json_if_possible(text: str) -> str:
- """如果文本是合法 JSON，则返回带缩进的可读格式；否则原样返回。"""
- raw = (text or "").strip()
- if not raw:
- return text
- if not (raw.startswith("{") or raw.startswith("[")):
- return text
- try:
- parsed = json.loads(raw)
- except Exception:
- return text
- return json.dumps(parsed, ensure_ascii=False, indent=2)
- def format_tool_result_for_log(result: Any) -> str:
- """将 ToolResult 或普通字符串格式化为可写入日志的文本（避免过长 metadata 刷屏）。"""
- if result is None:
- return ""
- if isinstance(result, str):
- s = result
- return s if len(s) <= 8000 else s[:8000] + "\n...(truncated)"
- title = getattr(result, "title", "") or ""
- output = getattr(result, "output", None) or ""
- err = getattr(result, "error", None)
- truncated = output if len(output) <= 6000 else output[:6000] + "\n...(truncated)"
- payload: Dict[str, Any] = {"title": title, "output": truncated}
- if err:
- payload["error"] = err
- md = getattr(result, "metadata", None)
- if isinstance(md, dict) and md:
- payload["metadata_keys"] = list(md.keys())
- return json.dumps(payload, ensure_ascii=False)
- def log_tool_call(tool_name: str, params: Dict[str, Any], result: str) -> None:
- """以折叠块结构化输出工具调用参数与返回内容。"""
- with log_fold(f"🔧 {tool_name}"):
- with log_fold("📥 调用参数"):
- log(json.dumps(params, ensure_ascii=False, indent=2))
- with log_fold("📤 返回内容"):
- log(_pretty_json_if_possible(result))
|
