Agent 工具调用与 Function Calling：从零设计可控的外部能力接口

from openai import OpenAI
client = OpenAI()

# 定义工具
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气信息。当用户询问天气时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"},
                    "date": {"type": "string", "description": "日期(YYYY-MM-DD)"},
                },
                "required": ["city", "date"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "send_email",
            "description": "发送一封邮件。",
            "parameters": {
                "type": "object",
                "properties": {
                    "to": {"type": "string"},
                    "subject": {"type": "string"},
                    "body": {"type": "string"},
                },
                "required": ["to", "subject", "body"],
            },
        },
    },
]

# 第一轮对话：请求工具调用
messages = [{"role": "user", "content": "北京明天天气怎么样？"}]
response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools,
    tool_choice="auto",  # auto | none | {"type": "function", "function": {"name": "xxx"}}
)

# model 决定调用 get_weather
print(response.choices[0].message)  # role=assistant, content=None, tool_calls=[...]

# 解析工具调用
tool_call = response.choices[0].message.tool_calls[0]
print(f"调用的工具: {tool_call.function.name}")
print(f"参数: {tool_call.function.arguments}")

# 执行工具（模拟）
result = {"temp": 28, "weather": "晴", "humidity": 45}

# 第二轮：将结果归还模型
messages.append(response.choices[0].message)  # 把模型请求加回去
messages.append({
    "role": "tool",
    "tool_call_id": tool_call.id,
    "name": tool_call.function.name,
    "content": str(result),
})

final = client.chat.completions.create(model="gpt-4o", messages=messages, tools=tools)
print(final.choices[0].message.content)  # "北京明天晴，28度..."

from anthropic import Anthropic
client = Anthropic()

# Claude 的工具声明与 OpenAI 几乎相同
tools = [
    {
        "name": "get_weather",
        "description": "获取天气信息",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {"type": "string"},
                "date": {"type": "string"},
            },
            "required": ["city"],
        },
    },
]

response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{"role": "user", "content": "北京明天天气"}],
    tools=tools,
)

# 检测是否使用了工具
if response.stop_reason == "tool_use":
    tool_use = response.content[-1]  # 最后一条 content block 是 tool_use
    print(f"Claude 想调用: {tool_use.name}，参数: {tool_use.input}")

# ❌ 差的描述
"description": "搜索功能"

# ✅ 好的描述
"description": "在商品数据库中根据关键词搜索商品信息并返回符合条件的 SKU 列表。适用于用户想要查找特定商品的讲候。不支持搜索用户个人信息。"

# ❌ 模糊命名
"from": str    # 从哪里？时间？地点？

# ✅ 自解释命名
"start_date": str
"origin_city": str

{
    "name": "get_user_profile",
    "description": "获取用户信息（幂等操作）。",
    "parameters": {...},
    "strict": True,  # OpenAI 参数校验强化
}

from typing import Callable, Dict
import asyncio

class RateLimitedTool:
    def __init__(self, fn: Callable, max_concurrent: int = 5):
        self.fn = fn
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def __call__(self, **kwargs):
        async with self.semaphore:
            return await self.fn(**kwargs)

class ToolRegistry:
    """跨 LLM 平台的工具注册器"""
    def __init__(self):
        self._tools: Dict[str, Callable] = {}

    def register(self, name: str, description: str, fn: Callable, schema: dict):
        self._tools[name] = {
            "fn": fn,
            "description": description,
            "schema": schema,
        }
        return self

    def to_openai_format(self):
        return [
            {
                "type": "function",
                "function": {
                    "name": name,
                    "description": meta["description"],
                    "parameters": meta["schema"],
                },
            }
            for name, meta in self._tools.items()
        ]

    def to_claude_format(self):
        return [
            {
                "name": name,
                "description": meta["description"],
                "input_schema": meta["schema"],
            }
            for name, meta in self._tools.items()
        ]

    def execute(self, tool_name: str, arguments: dict):
        if tool_name not in self._tools:
            raise ValueError(f"未知工具: {tool_name}")
        # Schema 校验
        import jsonschema
        jsonschema.validate(arguments, self._tools[tool_name]["schema"])
        return self._tools[tool_name]["fn"](**arguments)


# 使用
registry = ToolRegistry()
registry.register(
    name="get_weather",
    description="获取城市天气（当用户询问天气时使用）",
    fn=lambda city, date: {"temp": 28, "weather": "晴"},
    schema={"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]},
)

# OpenAI 格式
openai_tools = registry.to_openai_format()
# Claude 格式
claude_tools = registry.to_claude_format()

import asyncio
from typing import Any

class ParallelToolExecutor:
    """并行执行多个无依赖工具，有依赖时按 DAG 拓扑排序执行。"""

    async def execute_parallel(self, tool_calls: list[dict]) -> list[dict]:
        """并行执行所有工具调用。"""
        tasks = [
            self._execute_single(call["name"], call["arguments"])
            for call in tool_calls
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        return [
            {"tool": call["name"], "result": res if not isinstance(res, Exception) else {"error": str(res)}}
            for call, res in zip(tool_calls, results)
        ]

    async def execute_dag(self, dag: dict[str, Any]) -> dict[str, Any]:
        """按依赖关系图执行工具（简化版）。"""
        # dag = {
        #   "weather": {"tool": "get_weather", "args": {...}, "deps": []},
        #   "route": {"tool": "plan_route", "args": {...}, "deps": ["weather"]},
        # }
        completed = {}
        pending = set(dag.keys())

        while pending:
            # 找到所有依赖已满足的任务
            ready = {
                k for k in pending
                if all(d in completed for d in dag[k].get("deps", []))
            }
            if not ready:
                raise ValueError("DAG 存在循环依赖或缺失依赖")

            # 并行执行就绪任务
            tasks = {
                k: asyncio.create_task(
                    self._execute_single(dag[k]["tool"], dag[k]["args"])
                )
                for k in ready
            }
            for k, task in tasks.items():
                completed[k] = await task
                pending.remove(k)

        return completed

    async def _execute_single(self, name: str, args: dict) -> Any:
        # 实际执行...
        pass

import inspect

class AutoToolDiscovery:
    def __init__(self):
        self.registry = {}

    def discover(self, module):
        """扫描模块中所有 @tool 装饰的函数"""
        for name, obj in inspect.getmembers(module, inspect.isfunction):
            if hasattr(obj, "_tool_schema"):
                self.registry[name] = {
                    "fn": obj,
                    "schema": obj._tool_schema,
                    "desc": obj._tool_desc,
                }

# 装饰器用法
def tool(description, schema):
    def decorator(fn):
        fn._tool_schema = schema
        fn._tool_desc = description
        return fn
    return decorator

# 应用
@tool(
    description="获取天气信息",
    schema={"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}
)
def get_weather(city: str):
    return f"{city} 今天晴，25度"

class ResilientToolExecutor:
    def execute(self, tool_name, arguments, fallback=None):
        try:
            result = self.registry.execute(tool_name, arguments)
            return {"status": "success", "data": result}
        except jsonschema.ValidationError as e:
            return {
                "status": "validation_error",
                "message": f"参数错误: {e.message}",
                "suggestion": f"请确保按此格式调用: {json.dumps(self.registry.get_schema(tool_name))}",
            }
        except Exception as e:
            if fallback:
                return fallback(arguments)
            return {"status": "error", "message": str(e)}

class ReActAgent:
    """
    ReAct (Reasoning + Acting) 循环:
    Thought → Action → Observation → Thought → ... → Answer
    """
    def __init__(self, llm_client, tools: ToolRegistry):
        self.llm = llm_client
        self.tools = tools
        self.max_iterations = 10

    async def run(self, query: str) -> str:
        history = []
        for i in range(self.max_iterations):
            # 构造 ReAct Prompt
            prompt = self._build_react_prompt(query, history)
            response = await self.llm.complete(prompt)

            # 解析 Thought 和 Action
            thought, action = self._parse_response(response)

            if action is None:
                # 没有 Action，直接给出最终答案
                return thought

            # 执行工具
            tool_name = action["name"]
            tool_args = action["arguments"]
            try:
                observation = self.tools.execute(tool_name, tool_args)
            except Exception as e:
                observation = f"Error: {e}"

            history.append({
                "thought": thought,
                "action": action,
                "observation": observation,
            })

        return "达到最大迭代次数，未找到答案。"

    def _build_react_prompt(self, query: str, history: list) -> str:
        tool_desc = "\n".join(
            f"- {name}: {meta['description']}"
            for name, meta in self.tools._tools.items()
        )

        history_str = "\n".join(
            f"Thought: {h['thought']}\n"
            f"Action: {json.dumps(h['action'])}\n"
            f"Observation: {h['observation']}"
            for h in history
        )

        return f"""你是一个智能助手，可以使用以下工具：
{tool_desc}

请按照以下格式回答问题：
Thought: 你的思考过程
Action: {{"name": "工具名", "arguments": {{...}}}}
Observation: 工具执行结果
...（重复直到找到答案）

问题：{query}
{history_str}
Thought:"""

    def _parse_response(self, response: str) -> tuple[str, dict | None]:
        # 简化解析：提取 Action JSON
        import re
        action_match = re.search(r'Action:\s*(\{.*?\})', response, re.DOTALL)
        if action_match:
            thought = response.split("Action:")[0].replace("Thought:", "").strip()
            return thought, json.loads(action_match.group(1))
        return response, None

class HumanConfirmedToolExecutor:
    """
    高风险工具（如发送邮件、转账、删除数据）需要人类确认。
    """
    HIGH_RISK_TOOLS = {"send_email", "transfer_money", "delete_user", "deploy_production"}

    async def execute(self, tool_name: str, arguments: dict, user_session: str) -> dict:
        if tool_name in self.HIGH_RISK_TOOLS:
            # 暂停执行，等待人类确认
            confirmation_id = await self.request_human_confirmation(
                user_session, tool_name, arguments
            )
            return {
                "status": "awaiting_confirmation",
                "confirmation_id": confirmation_id,
                "message": f"请确认是否执行 {tool_name}：{json.dumps(arguments, ensure_ascii=False)}",
            }

        # 低风险工具直接执行
        return await self._execute_tool(tool_name, arguments)

    async def confirm(self, confirmation_id: str, approved: bool) -> dict:
        pending = self.pending_confirmations.pop(confirmation_id, None)
        if not pending:
            return {"status": "error", "message": "确认请求已过期"}

        if approved:
            return await self._execute_tool(pending["tool"], pending["args"])
        return {"status": "denied", "message": "用户拒绝了操作"}