LLM 函数调用实战：让大模型真正替你干活

{
  "name": "get_weather",
  "description": "查询指定城市指定日期的天气信息，包括温度、湿度、风速和天气状况。如果用户未指定日期，默认查询今天的天气。",
  "parameters": {
    "type": "object",
    "properties": {
      "location": {
        "type": "string",
        "description": "城市名称，支持中文或英文，例如'北京'、'Shanghai'。不要使用'这个城市'等代词。",
        "examples": ["北京", "上海", "东京"]
      },
      "date": {
        "type": "string",
        "description": "查询日期，格式为 YYYY-MM-DD。不写则默认今天。",
        "examples": ["2026-03-31", "2026-04-01"]
      },
      "unit": {
        "type": "string",
        "enum": ["celsius", "fahrenheit"],
        "description": "温度单位，不填默认 celsius（摄氏度）。"
      }
    },
    "required": ["location"]
  }
}

// ❌ 差：信息不足
{ "name": "search", "description": "搜索文档", "parameters": { ... } }

// ✅ 好：明确边界和触发时机
{ "name": "search_documents", "description": "在企业知识库中搜索内部文档。仅当用户明确要求查找'文档'、'资料'、'相关规定'时使用。不要用于查询外部信息或实时数据。", "parameters": { ... } }

"properties": {
  "action": {
    "type": "string",
    "description": "要执行的操作类型",
    "examples": ["start", "stop", "restart", "status"]
  }
}

// ❌ 过度限制：LLM 猜不中这些值
{ "status": { "enum": ["ACTIVE", "INACTIVE", "PENDING"] } }

// ✅ 适度限制：给出合理范围，但保留灵活性
{ "status": { "type": "string", "description": "订单状态，支持：pending（待处理）、paid（已支付）、shipped（已发货）、completed（已完成）、cancelled（已取消）" } }

# 用户: "帮我取消订单 #ABC123"
# LLM 输出: { "name": "cancel_order", "arguments": { "order_id": "ABC123" } }

# 但系统要求订单 ID 必须是大写数字
# API 返回: 400 Bad Request: "order_id 格式不正确"

# 优雅处理：参数预处理
def cancel_order_handler(args):
    order_id = args["order_id"]
    # 标准化：去掉空格、转大写、验证格式
    order_id = order_id.strip().upper()
    if not order_id.match(r'^[A-Z0-9]{6,20}$'):
        return {"error": f"订单号格式无效: {order_id}"}
    return call_cancel_api(order_id)

# 优雅重试 + 降级策略
def call_weather_api(location, retries=3):
    for attempt in range(retries):
        try:
            return requests.get(f"https://api.weather.com/v3", timeout=5).json()
        except (Timeout, ConnectionError) as e:
            if attempt == retries - 1:
                # 降级：返回「无法查询」而非抛出异常
                return {"error": "weather_api_unavailable", "message": f"天气服务暂时不可用，请稍后重试"}
            time.sleep(2 ** attempt)  # 指数退避

def handle_tool_error(tool_name, error, context):
    return {
        "tool_result": {
            "success": False,
            "error": str(error),
            "recoverable": False,  # 关键标志
            "suggestion": "天气服务暂时不可用，建议告知用户当前无法提供天气信息，可建议用户稍后重试或手动查询。"
        }
    }

AVAILABLE_TOOLS = {
    "get_weather": get_weather_func,
    "search_documents": search_documents_func,
    "send_email": send_email_func,
    # ...
}

def execute_tool_call(tool_name, arguments):
    if tool_name not in AVAILABLE_TOOLS:
        # 尝试模糊匹配（处理 LLM 大小写/拼写错误）
        matches = difflib.get_close_matches(tool_name, AVAILABLE_TOOLS.keys(), n=1, cutoff=0.6)
        if matches:
            tool_name = matches[0]  # 自动纠正
        else:
            return {"error": f"未找到工具: {tool_name}，可用工具: {list(AVAILABLE_TOOLS.keys())}"}
    
    return AVAILABLE_TOOLS[tool_name](arguments)

def compact_conversation(messages, max_tokens=3000):
    """将旧的工具调用和结果压缩为摘要"""
    compacted = []
    tool_call_count = 0
    
    for msg in messages:
        if msg["role"] == "tool":
            tool_call_count += 1
            if tool_call_count <= 3:
                compacted.append(msg)  # 保留最近3次工具调用详情
            elif tool_call_count == 4:
                compacted.append({
                    "role": "system",
                    "content": f"[{tool_call_count}次工具调用已省略最近结果，均成功]"
                })
        else:
            compacted.append(msg)
    
    return compacted

# ❌ 危险：Agent 有完整数据库权限
tools = [
    {"name": "execute_sql", "description": "执行任意 SQL", "permissions": "FULL_DATABASE"},
]

# ✅ 安全：只读权限，限制操作类型
tools = [
    {"name": "query_orders", "description": "查询订单", "permissions": "READ_ONLY", "allowed_operations": ["SELECT"]},
]

def safe_delete_file(args):
    # 1. 验证路径：不允许路径穿越
    filepath = os.path.abspath(args["filepath"])
    ALLOWED_DIRS = ["/data/uploads", "/data/exports"]
    if not any(filepath.startswith(d) for d in ALLOWED_DIRS):
        raise PermissionError(f"禁止访问目录之外的文件: {filepath}")
    
    # 2. 危险操作二次确认
    if args.get("confirm") != True:
        return {"status": "awaiting_confirmation", "message": "确定要删除此文件吗？请确认。"}
    
    return os.remove(filepath)

import structlog
logger = structlog.get_logger()

def audited_tool_call(tool_name, arguments, user_id, session_id):
    logger.info(
        "tool_call",
        tool=tool_name,
        args=sanitize_args(arguments),  # 移除敏感字段如密码
        user=user_id,
        session=session_id,
        timestamp=datetime.utcnow().isoformat()
    )
    # 实际执行...

// 工具发布到 MCP 注册表
{
  "name": "github-repos",
  "version": "1.0.0",
  "description": "查询 GitHub 仓库信息、PR 和 Issue 状态",
  "provider": "github",
  "schema_url": "https://registry.mcp.tools/schemas/github-repos/v1.json",
  "auth": { "type": "oauth2", "scopes": ["repo"] }
}

from mcp_registry import RegistryClient

registry = RegistryClient("https://registry.mcp.tools")

# Agent 启动时：动态加载用户所需的工具
available_tools = registry.discover_tools(
    tags=["productivity", "data-analysis"],
    auth_required=False,
    quality_score_gte=0.8
)

# 加载到 Agent
agent.load_tools(available_tools)

from dataclasses import dataclass
from enum import Enum
from typing import Any, Optional
import structlog

logger = structlog.get_logger()

class ToolResultStatus(Enum):
    SUCCESS = "success"
    RETRYABLE_ERROR = "retryable_error"
    FATAL_ERROR = "fatal_error"

@dataclass
class ToolResult:
    status: ToolResultStatus
    data: Optional[dict] = None
    error: Optional[str] = None
    suggestion: Optional[str] = None

class ToolRegistry:
    """生产级工具注册中心"""
    
    def __init__(self):
        self._tools: dict[str, callable] = {}
        self._schemas: dict[str, dict] = {}
    
    def register(self, schema: dict, handler: callable):
        self._tools[schema["name"]] = handler
        self._schemas[schema["name"]] = schema
        logger.info("tool_registered", name=schema["name"])
    
    def get_schema(self, name: str) -> Optional[dict]:
        return self._schemas.get(name)
    
    def list_tools(self) -> list[dict]:
        return list(self._schemas.values())
    
    def execute(self, name: str, args: dict, context: dict) -> ToolResult:
        if name not in self._tools:
            return ToolResult(
                status=ToolResultStatus.FATAL_ERROR,
                error=f"未知工具: {name}"
            )
        
        try:
            result = self._tools[name](args, context)
            return ToolResult(status=ToolResultStatus.SUCCESS, data=result)
        except RetryableError as e:
            return ToolResult(
                status=ToolResultStatus.RETRYABLE_ERROR,
                error=str(e),
                suggestion=e.suggestion
            )
        except Exception as e:
            logger.error("tool_execution_failed", tool=name, error=str(e))
            return ToolResult(
                status=ToolResultStatus.FATAL_ERROR,
                error=f"工具执行失败: {str(e)}"
            )

# 使用示例
registry = ToolRegistry()
registry.register(WEATHER_SCHEMA, get_weather_handler)
registry.register(SEARCH_SCHEMA, search_handler)