feat: improve tool-call bridging and env documentation
This commit is contained in:
@@ -2,6 +2,7 @@ from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import json
|
||||
import re
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
@@ -50,10 +51,16 @@ def _tool_event_allowed(
|
||||
*,
|
||||
forced_tool_name: str | None = None,
|
||||
) -> bool:
|
||||
if not (tool_config and isinstance(tool_config.get("tools"), list) and tool_config.get("tools")):
|
||||
if not (
|
||||
tool_config
|
||||
and isinstance(tool_config.get("tools"), list)
|
||||
and tool_config.get("tools")
|
||||
):
|
||||
return True
|
||||
for tool in tool_config.get("tools") or []:
|
||||
if tool_name == _anthropic_tool_name(tool) or tool_name == _openai_tool_name(tool):
|
||||
if tool_name == _anthropic_tool_name(tool) or tool_name == _openai_tool_name(
|
||||
tool
|
||||
):
|
||||
return True
|
||||
return bool(forced_tool_name and tool_name == forced_tool_name)
|
||||
|
||||
@@ -67,7 +74,9 @@ def _allowed_tool_event(
|
||||
if not isinstance(tool, dict):
|
||||
return None
|
||||
tool_name = str(tool.get("name") or "")
|
||||
if not _tool_event_allowed(tool_name, tool_config, forced_tool_name=forced_tool_name):
|
||||
if not _tool_event_allowed(
|
||||
tool_name, tool_config, forced_tool_name=forced_tool_name
|
||||
):
|
||||
return None
|
||||
return tool
|
||||
|
||||
@@ -104,7 +113,9 @@ def _allowed_stream_tool_event(
|
||||
if not isinstance(tool, dict):
|
||||
return None
|
||||
tool_name = str(tool.get("name") or "")
|
||||
if not _tool_event_allowed(tool_name, tool_config, forced_tool_name=forced_tool_name):
|
||||
if not _tool_event_allowed(
|
||||
tool_name, tool_config, forced_tool_name=forced_tool_name
|
||||
):
|
||||
return None
|
||||
return tool
|
||||
|
||||
@@ -150,7 +161,9 @@ def _json_object_from_text(text: str) -> dict[str, Any] | None:
|
||||
return parsed if isinstance(parsed, dict) else None
|
||||
|
||||
|
||||
def _tool_code_single_arg_name(tools: list[dict[str, Any]] | None, forced_tool_name: str) -> str | None:
|
||||
def _tool_code_single_arg_name(
|
||||
tools: list[dict[str, Any]] | None, forced_tool_name: str
|
||||
) -> str | None:
|
||||
if not isinstance(tools, list):
|
||||
return None
|
||||
for tool in tools:
|
||||
@@ -228,7 +241,9 @@ def _forced_tool_event_from_text(
|
||||
) -> dict[str, Any] | None:
|
||||
parsed = _json_object_from_text(text)
|
||||
if parsed is None:
|
||||
parsed = _tool_code_object_from_text(text, forced_tool_name, single_arg_name=single_arg_name)
|
||||
parsed = _tool_code_object_from_text(
|
||||
text, forced_tool_name, single_arg_name=single_arg_name
|
||||
)
|
||||
if parsed is None:
|
||||
return None
|
||||
|
||||
@@ -288,7 +303,9 @@ def _forced_tool_fallback_event(
|
||||
)
|
||||
|
||||
|
||||
def _openai_tool_call(tool: dict[str, Any], *, forced_id: str | None = None) -> dict[str, Any]:
|
||||
def _openai_tool_call(
|
||||
tool: dict[str, Any], *, forced_id: str | None = None
|
||||
) -> dict[str, Any]:
|
||||
return {
|
||||
"id": str(tool.get("id") or forced_id or f"call_{uuid.uuid4().hex}"),
|
||||
"type": "function",
|
||||
@@ -299,6 +316,42 @@ def _openai_tool_call(tool: dict[str, Any], *, forced_id: str | None = None) ->
|
||||
}
|
||||
|
||||
|
||||
def _extract_function_call_event_from_text(
|
||||
text: str,
|
||||
*,
|
||||
forced_tool_name: str | None,
|
||||
) -> dict[str, Any] | None:
|
||||
raw = (text or "").strip()
|
||||
if not raw:
|
||||
return None
|
||||
m = re.search(r"<function_calls>\s*(\{.*?\})\s*</function_calls>", raw, flags=re.S)
|
||||
if not m:
|
||||
return None
|
||||
try:
|
||||
payload = json.loads(m.group(1))
|
||||
except Exception:
|
||||
return None
|
||||
if not isinstance(payload, dict):
|
||||
return None
|
||||
name = payload.get("name")
|
||||
if not isinstance(name, str) or not name.strip():
|
||||
return None
|
||||
name = name.strip()
|
||||
if forced_tool_name and name != forced_tool_name:
|
||||
return None
|
||||
arguments = payload.get("arguments")
|
||||
if isinstance(arguments, str):
|
||||
try:
|
||||
arguments = json.loads(arguments)
|
||||
except Exception:
|
||||
return None
|
||||
if arguments is None:
|
||||
arguments = {}
|
||||
if not isinstance(arguments, dict):
|
||||
return None
|
||||
return {"name": name, "input": arguments}
|
||||
|
||||
|
||||
def _anthropic_tool_use_block(
|
||||
tool: dict[str, Any], *, forced_id: str | None = None
|
||||
) -> dict[str, Any]:
|
||||
|
||||
Reference in New Issue
Block a user