工具调用 schema 排障：docstring、参数校验与工具异常实测

"""What the model actually receives: the JSON schema LangChain builds from a Python tool definition."""
import json
from typing import Literal, Optional
 
from langchain_core.tools import tool
from langchain_core.utils.function_calling import convert_to_openai_tool
from pydantic import BaseModel, Field
 
 
def show(label: str, fn) -> None:
    spec = convert_to_openai_tool(fn)["function"]
    print(f"== {label}")
    print("description:", json.dumps(spec["description"], ensure_ascii=False))
    print("parameters :", json.dumps(spec["parameters"]["properties"], ensure_ascii=False, sort_keys=True))
    print("required   :", spec["parameters"].get("required", []))
 
 
@tool
def search_plain(query: str, limit: int = 5, lang: Literal["zh", "en"] = "zh", tags: Optional[list[str]] = None) -> str:
    """Search the knowledge base.
 
    Args:
        query: What to search for.
        limit: Maximum number of results.
        lang: Result language.
        tags: Optional tag filters.
    """
    return "ok"
 
 
@tool(parse_docstring=True)
def search_parsed(query: str, limit: int = 5, lang: Literal["zh", "en"] = "zh", tags: Optional[list[str]] = None) -> str:
    """Search the knowledge base.
 
    Args:
        query: What to search for.
        limit: Maximum number of results.
        lang: Result language.
        tags: Optional tag filters.
    """
    return "ok"
 
 
class SearchArgs(BaseModel):
    query: str = Field(description="What to search for.")
    limit: int = Field(default=5, ge=1, le=20, description="Maximum number of results (1-20).")
 
 
@tool(args_schema=SearchArgs)
def search_schema(query: str, limit: int = 5) -> str:
    """Search the knowledge base."""
    return "ok"
 
 
show("plain docstring (default)", search_plain)
show("parse_docstring=True", search_parsed)
show("args_schema with Field constraints", search_schema)
 
print("== no docstring")
try:
 
    @tool
    def undocumented(x: int) -> int:
        return x
 
except Exception as exc:
    print(f"{type(exc).__name__}: {exc}")

== plain docstring (default)
description: "Search the knowledge base.\n\n    Args:\n        query: What to search for.\n        limit: Maximum number of results.\n        lang: Result language.\n        tags: Optional tag filters."
parameters : {"lang": {"default": "zh", "enum": ["zh", "en"], "type": "string"}, "limit": {"default": 5, "type": "integer"}, "query": {"type": "string"}, "tags": {"anyOf": [{"items": {"type": "string"}, "type": "array"}, {"type": "null"}], "default": null}}
required   : ['query']
== parse_docstring=True
description: "Search the knowledge base."
parameters : {"lang": {"default": "zh", "description": "Result language.", "enum": ["zh", "en"], "type": "string"}, "limit": {"default": 5, "description": "Maximum number of results.", "type": "integer"}, "query": {"description": "What to search for.", "type": "string"}, "tags": {"anyOf": [{"items": {"type": "string"}, "type": "array"}, {"type": "null"}], "default": null, "description": "Optional tag filters."}}
required   : ['query']
== args_schema with Field constraints
description: "Search the knowledge base."
parameters : {"limit": {"default": 5, "description": "Maximum number of results (1-20).", "maximum": 20, "minimum": 1, "type": "integer"}, "query": {"description": "What to search for.", "type": "string"}}
required   : ['query']
== no docstring
ValueError: Function must have a docstring if description not provided.

"""How strictly are tool arguments validated before the function body runs?"""
from typing import Literal
 
from langchain_core.tools import tool
from pydantic import BaseModel, Field
 
 
class SearchArgs(BaseModel):
    query: str
    limit: int = Field(default=5, ge=1, le=20)
    lang: Literal["zh", "en"] = "zh"
 
 
CALLS: list = []
 
 
@tool(args_schema=SearchArgs)
def search(query: str, limit: int = 5, lang: str = "zh") -> str:
    """Search the knowledge base."""
    CALLS.append((query, limit, lang))
    return "ok"
 
 
CASES = [
    ("string that looks like an int", {"query": "a", "limit": "3"}),
    ("string that is not an int", {"query": "a", "limit": "many"}),
    ("float with a fraction", {"query": "a", "limit": 2.5}),
    ("out of range", {"query": "a", "limit": 99}),
    ("value outside the Literal", {"query": "a", "lang": "fr"}),
    ("unknown extra field", {"query": "a", "page": 2}),
    ("missing required field", {}),
]
 
for label, args in CASES:
    CALLS.clear()
    try:
        search.invoke(args)
        outcome = f"accepted, function received {CALLS[0]}"
    except Exception as exc:
        errors = exc.errors()
        outcome = f"{type(exc).__name__} ({errors[0]['type']} at {errors[0]['loc']})"
    print(f"{label:<32} {outcome}")

string that looks like an int    accepted, function received ('a', 3, 'zh')
string that is not an int        ValidationError (int_parsing at ('limit',))
float with a fraction            ValidationError (int_from_float at ('limit',))
out of range                     ValidationError (less_than_equal at ('limit',))
value outside the Literal        ValidationError (literal_error at ('lang',))
unknown extra field              accepted, function received ('a', 5, 'zh')
missing required field           ValidationError (missing at ('query',))

"""What happens to the agent run when a tool fails? Four ways to configure it."""
import sys
from pathlib import Path
 
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "agent-vs-langgraph"))
 
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain_core.messages import AIMessage, ToolMessage
from langchain_core.tools import ToolException, tool
 
from shared import ScriptedChatModel
 
 
@tool
def raises_runtime(key: str) -> str:
    """Always fails with RuntimeError."""
    raise RuntimeError("database unavailable")
 
 
@tool
def raises_tool_exception(key: str) -> str:
    """Always fails with ToolException."""
    raise ToolException("record not found")
 
 
@tool
def raises_tool_exception_handled(key: str) -> str:
    """Always fails with ToolException, with handle_tool_error enabled."""
    raise ToolException("record not found")
 
 
raises_tool_exception_handled.handle_tool_error = True
 
 
@wrap_tool_call
def convert_failures(request, handler):
    """Turn any tool exception into an error ToolMessage the model can read."""
    try:
        return handler(request)
    except Exception as exc:
        return ToolMessage(
            content=f"Tool failed: {type(exc).__name__}: {exc}",
            tool_call_id=request.tool_call["id"],
            status="error",
        )
 
 
def run(label: str, the_tool, middleware=()) -> None:
    script = [
        AIMessage(content="", tool_calls=[{"name": the_tool.name, "args": {"key": "a"}, "id": "c1"}]),
        AIMessage(content="done"),
    ]
    agent = create_agent(ScriptedChatModel(script=script), tools=[the_tool], middleware=list(middleware))
    try:
        result = agent.invoke({"messages": [("user", "go")]})
        tool_msg = next(m for m in result["messages"] if m.type == "tool")
        outcome = f"run continued; ToolMessage(status={tool_msg.status}): {tool_msg.content}; final answer: {result['messages'][-1].content!r}"
    except Exception as exc:
        outcome = f"run failed with {type(exc).__name__}: {exc}"
    print(f"{label:<44} {outcome}")
 
 
run("RuntimeError, no configuration", raises_runtime)
run("ToolException, no configuration", raises_tool_exception)
run("ToolException, handle_tool_error=True", raises_tool_exception_handled)
run("RuntimeError + @wrap_tool_call middleware", raises_runtime, [convert_failures])

RuntimeError, no configuration               run failed with RuntimeError: database unavailable
ToolException, no configuration              run failed with ToolException: record not found
ToolException, handle_tool_error=True        run continued; ToolMessage(status=error): record not found; final answer: 'done'
RuntimeError + @wrap_tool_call middleware    run continued; ToolMessage(status=error): Tool failed: RuntimeError: database unavailable; final answer: 'done'