04 工具调用（Tool / Function Calling） ​

import json
from typing import Any
from jsonschema import validate, ValidationError
from openai import OpenAI

# 工具 JSON Schema（parameters）
WEATHER_PARAMS = {
    "type": "object",
    "properties": {
        "city": {"type": "string", "description": "城市名，中文或英文"},
        "date": {"type": "string", "description": "日期 YYYY-MM-DD"},
    },
    "required": ["city"],
    "additionalProperties": False,
}

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市在某日期的天气。用户只说「今天」时，请换算为具体日期再调用。",
            "parameters": WEATHER_PARAMS,
        },
    }
]

# 假实现：真实项目里对接 HTTP API
def get_weather(city: str, date: str | None = None) -> dict[str, Any]:
    return {"city": city, "date": date or "today", "condition": "晴", "temp_c": 22}

def run_tool(name: str, arguments: str) -> str:
    args = json.loads(arguments or "{}")
    validate(instance=args, schema=WEATHER_PARAMS)  # 与 TOOLS 中一致
    if name == "get_weather":
        return json.dumps(get_weather(**args), ensure_ascii=False)
    raise ValueError(f"unknown tool: {name}")

def chat_with_tools(user_message: str) -> str:
    client = OpenAI()
    messages: list[dict[str, Any]] = [
        {"role": "system", "content": "你是助手，需要数据时调用工具，不要编造天气。"},
        {"role": "user", "content": user_message},
    ]
    for _ in range(5):  # 防止死循环
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
        )
        msg = resp.choices[0].message
        messages.append(msg.model_dump())
        if not msg.tool_calls:
            return msg.content or ""
        for tc in msg.tool_calls:
            try:
                result = run_tool(tc.function.name, tc.function.arguments)
            except (json.JSONDecodeError, ValidationError, ValueError) as e:
                result = json.dumps({"error": str(e)}, ensure_ascii=False)
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": result,
                }
            )
    return "超过最大工具调用轮次"

if __name__ == "__main__":
    print(chat_with_tools("北京明天天气怎么样？"))

import json
from typing import Any
from jsonschema import validate, ValidationError
from openai import OpenAI

# 工具 JSON Schema（parameters）
WEATHER_PARAMS = {
    "type": "object",
    "properties": {
        "city": {"type": "string", "description": "城市名，中文或英文"},
        "date": {"type": "string", "description": "日期 YYYY-MM-DD"},
    },
    "required": ["city"],
    "additionalProperties": False,
}

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市在某日期的天气。用户只说「今天」时，请换算为具体日期再调用。",
            "parameters": WEATHER_PARAMS,
        },
    }
]

# 假实现：真实项目里对接 HTTP API
def get_weather(city: str, date: str | None = None) -> dict[str, Any]:
    return {"city": city, "date": date or "today", "condition": "晴", "temp_c": 22}

def run_tool(name: str, arguments: str) -> str:
    args = json.loads(arguments or "{}")
    validate(instance=args, schema=WEATHER_PARAMS)  # 与 TOOLS 中一致
    if name == "get_weather":
        return json.dumps(get_weather(**args), ensure_ascii=False)
    raise ValueError(f"unknown tool: {name}")

def chat_with_tools(user_message: str) -> str:
    client = OpenAI()
    messages: list[dict[str, Any]] = [
        {"role": "system", "content": "你是助手，需要数据时调用工具，不要编造天气。"},
        {"role": "user", "content": user_message},
    ]
    for _ in range(5):  # 防止死循环
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
        )
        msg = resp.choices[0].message
        messages.append(msg.model_dump())
        if not msg.tool_calls:
            return msg.content or ""
        for tc in msg.tool_calls:
            try:
                result = run_tool(tc.function.name, tc.function.arguments)
            except (json.JSONDecodeError, ValidationError, ValueError) as e:
                result = json.dumps({"error": str(e)}, ensure_ascii=False)
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": result,
                }
            )
    return "超过最大工具调用轮次"

if __name__ == "__main__":
    print(chat_with_tools("北京明天天气怎么样？"))

from typing import Literal
from langchain_core.tools import tool

@tool
def search_product(
    query: str,
    category: Literal["book", "electronics", ""] = "",
) -> str:
    """在电商站内搜索商品。用户要找商品、比价、看库存时用；不要用于闲聊。

    Args:
        query: 搜索关键词
        category: 可选类目过滤，不知道则留空
    """
    # 伪实现
    return f"[dummy] results for {query!r} in {category or 'all'}"

# 注册到 Agent 时通常传入 tools=[search_product]
# 例如 create_react_agent(llm, tools=[search_product], ...)

from typing import Literal
from langchain_core.tools import tool

@tool
def search_product(
    query: str,
    category: Literal["book", "electronics", ""] = "",
) -> str:
    """在电商站内搜索商品。用户要找商品、比价、看库存时用；不要用于闲聊。

    Args:
        query: 搜索关键词
        category: 可选类目过滤，不知道则留空
    """
    # 伪实现
    return f"[dummy] results for {query!r} in {category or 'all'}"

# 注册到 Agent 时通常传入 tools=[search_product]
# 例如 create_react_agent(llm, tools=[search_product], ...)

# 需要: pip install mcp
# 以下为概念示例，包名与 API 请以官方文档为准
try:
    from mcp.server.fastmcp import FastMCP
except ImportError:
    FastMCP = None  # 环境未安装时仅作结构说明

if FastMCP:
    mcp = FastMCP("demo")

    @mcp.tool()
    def add(a: int, b: int) -> int:
        """返回两个整数之和。"""
        return a + b

# 通常以 `mcp.run(transport="stdio")` 由 Host 拉起子进程

# 需要: pip install mcp
# 以下为概念示例，包名与 API 请以官方文档为准
try:
    from mcp.server.fastmcp import FastMCP
except ImportError:
    FastMCP = None  # 环境未安装时仅作结构说明

if FastMCP:
    mcp = FastMCP("demo")

    @mcp.tool()
    def add(a: int, b: int) -> int:
        """返回两个整数之和。"""
        return a + b

# 通常以 `mcp.run(transport="stdio")` 由 Host 拉起子进程

from dataclasses import dataclass
# 需要: pip install numpy openai
import numpy as np
from openai import OpenAI

@dataclass
class ToolSpec:
    name: str
    description: str

def embed_texts(client: OpenAI, model: str, texts: list[str]) -> np.ndarray:
    resp = client.embeddings.create(model=model, input=texts)
    vecs = [np.array(d.embedding, dtype=np.float32) for d in resp.data]
    mat = np.stack(vecs, axis=0)
    norms = np.linalg.norm(mat, axis=1, keepdims=True) + 1e-8
    return mat / norms

def route_tools(
    query: str,
    tools: list[ToolSpec],
    client: OpenAI,
    embed_model: str = "text-embedding-3-small",
    top_k: int = 3,
) -> list[ToolSpec]:
    corpus = [f"{t.name}\n{t.description}" for t in tools]
    doc_emb = embed_texts(client, embed_model, corpus)
    q_emb = embed_texts(client, embed_model, [query])[0]
    scores = doc_emb @ q_emb
    idx = np.argsort(-scores)[:top_k]
    return [tools[i] for i in idx]

# 使用：仅把 route_tools 返回的子集塞进 chat.completions 的 tools 参数

from dataclasses import dataclass
# 需要: pip install numpy openai
import numpy as np
from openai import OpenAI

@dataclass
class ToolSpec:
    name: str
    description: str

def embed_texts(client: OpenAI, model: str, texts: list[str]) -> np.ndarray:
    resp = client.embeddings.create(model=model, input=texts)
    vecs = [np.array(d.embedding, dtype=np.float32) for d in resp.data]
    mat = np.stack(vecs, axis=0)
    norms = np.linalg.norm(mat, axis=1, keepdims=True) + 1e-8
    return mat / norms

def route_tools(
    query: str,
    tools: list[ToolSpec],
    client: OpenAI,
    embed_model: str = "text-embedding-3-small",
    top_k: int = 3,
) -> list[ToolSpec]:
    corpus = [f"{t.name}\n{t.description}" for t in tools]
    doc_emb = embed_texts(client, embed_model, corpus)
    q_emb = embed_texts(client, embed_model, [query])[0]
    scores = doc_emb @ q_emb
    idx = np.argsort(-scores)[:top_k]
    return [tools[i] for i in idx]

# 使用：仅把 route_tools 返回的子集塞进 chat.completions 的 tools 参数

import concurrent.futures
import json
from typing import Any, Callable

ToolFn = Callable[..., Any]

def safe_call(name: str, fn: ToolFn, kwargs: dict[str, Any]) -> dict[str, Any]:
    try:
        return {"tool": name, "ok": True, "result": fn(**kwargs)}
    except Exception as e:
        return {"tool": name, "ok": False, "error": str(e)}

def run_parallel_tools(
    calls: list[tuple[str, ToolFn, dict[str, Any]]],
) -> list[dict[str, Any]]:
    with concurrent.futures.ThreadPoolExecutor(max_workers=8) as ex:
        futs = [ex.submit(safe_call, n, f, k) for n, f, k in calls]
        return [f.result() for f in futs]

# 例：先串行拿 user_id，再并行查订单与积分（伪函数）
def lookup_user_id(email: str) -> str:
    return "u_123"

def fetch_orders(uid: str) -> list:
    return [{"id": 1}]

def fetch_points(uid: str) -> int:
    return 42

def orchestrate(user_email: str) -> str:
    uid = lookup_user_id(user_email)
    second = run_parallel_tools(
        [
            ("orders", fetch_orders, {"uid": uid}),
            ("points", fetch_points, {"uid": uid}),
        ]
    )
    return json.dumps(second, ensure_ascii=False)

import concurrent.futures
import json
from typing import Any, Callable

ToolFn = Callable[..., Any]

def safe_call(name: str, fn: ToolFn, kwargs: dict[str, Any]) -> dict[str, Any]:
    try:
        return {"tool": name, "ok": True, "result": fn(**kwargs)}
    except Exception as e:
        return {"tool": name, "ok": False, "error": str(e)}

def run_parallel_tools(
    calls: list[tuple[str, ToolFn, dict[str, Any]]],
) -> list[dict[str, Any]]:
    with concurrent.futures.ThreadPoolExecutor(max_workers=8) as ex:
        futs = [ex.submit(safe_call, n, f, k) for n, f, k in calls]
        return [f.result() for f in futs]

# 例：先串行拿 user_id，再并行查订单与积分（伪函数）
def lookup_user_id(email: str) -> str:
    return "u_123"

def fetch_orders(uid: str) -> list:
    return [{"id": 1}]

def fetch_points(uid: str) -> int:
    return 42

def orchestrate(user_email: str) -> str:
    uid = lookup_user_id(user_email)
    second = run_parallel_tools(
        [
            ("orders", fetch_orders, {"uid": uid}),
            ("points", fetch_points, {"uid": uid}),
        ]
    )
    return json.dumps(second, ensure_ascii=False)

import os
import urllib.parse
import urllib.request

def web_search(query: str, max_results: int = 5) -> list[dict]:
    """占位：真实环境使用官方搜索 API 并处理分页。"""
    q = urllib.parse.quote(query)
    url = f"https://duckduckgo.com/html/?q={q}"  # 示例仅作结构说明，生产请用合规 API
    req = urllib.request.Request(url, headers={"User-Agent": "AgentBot/1.0"})
    with urllib.request.urlopen(req, timeout=10) as resp:
        html = resp.read(200_000)
    return [{"title": "stub", "snippet": html[:200].decode(errors="ignore"), "url": url}]

import os
import urllib.parse
import urllib.request

def web_search(query: str, max_results: int = 5) -> list[dict]:
    """占位：真实环境使用官方搜索 API 并处理分页。"""
    q = urllib.parse.quote(query)
    url = f"https://duckduckgo.com/html/?q={q}"  # 示例仅作结构说明，生产请用合规 API
    req = urllib.request.Request(url, headers={"User-Agent": "AgentBot/1.0"})
    with urllib.request.urlopen(req, timeout=10) as resp:
        html = resp.read(200_000)
    return [{"title": "stub", "snippet": html[:200].decode(errors="ignore"), "url": url}]

def query_user_orders(conn, user_id: str, limit: int = 20) -> list:
    sql = "SELECT id, amount, created_at FROM orders WHERE user_id = %s ORDER BY created_at DESC LIMIT %s"
    with conn.cursor() as cur:
        cur.execute(sql, (user_id, limit))
        return list(cur.fetchall())

def query_user_orders(conn, user_id: str, limit: int = 20) -> list:
    sql = "SELECT id, amount, created_at FROM orders WHERE user_id = %s ORDER BY created_at DESC LIMIT %s"
    with conn.cursor() as cur:
        cur.execute(sql, (user_id, limit))
        return list(cur.fetchall())

import os

SANDBOX_ROOT = "/var/agent_sandbox"

def safe_read_file(path: str, max_bytes: int = 50_000) -> str:
    full = os.path.realpath(os.path.join(SANDBOX_ROOT, path))
    if not full.startswith(os.path.realpath(SANDBOX_ROOT) + os.sep):
        raise PermissionError("path escapes sandbox")
    with open(full, "rb") as f:
        return f.read(max_bytes).decode("utf-8", errors="replace")

import os

SANDBOX_ROOT = "/var/agent_sandbox"

def safe_read_file(path: str, max_bytes: int = 50_000) -> str:
    full = os.path.realpath(os.path.join(SANDBOX_ROOT, path))
    if not full.startswith(os.path.realpath(SANDBOX_ROOT) + os.sep):
        raise PermissionError("path escapes sandbox")
    with open(full, "rb") as f:
        return f.read(max_bytes).decode("utf-8", errors="replace")

import ast
import operator

_ALLOWED = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
    ast.USub: operator.neg,
    ast.Pow: operator.pow,
}

def eval_expr(node: ast.AST) -> float:
    if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
        return float(node.value)
    if isinstance(node, ast.BinOp) and type(node.op) in _ALLOWED:
        return _ALLOWED[type(node.op)](eval_expr(node.left), eval_expr(node.right))
    if isinstance(node, ast.UnaryOp) and type(node.op) in _ALLOWED:
        return _ALLOWED[type(node.op)](eval_expr(node.operand))
    raise ValueError("unsupported expression")

def calculator(expr: str) -> float:
    tree = ast.parse(expr, mode="eval")
    return eval_expr(tree.body)

import ast
import operator

_ALLOWED = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
    ast.USub: operator.neg,
    ast.Pow: operator.pow,
}

def eval_expr(node: ast.AST) -> float:
    if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
        return float(node.value)
    if isinstance(node, ast.BinOp) and type(node.op) in _ALLOWED:
        return _ALLOWED[type(node.op)](eval_expr(node.left), eval_expr(node.right))
    if isinstance(node, ast.UnaryOp) and type(node.op) in _ALLOWED:
        return _ALLOWED[type(node.op)](eval_expr(node.operand))
    raise ValueError("unsupported expression")

def calculator(expr: str) -> float:
    tree = ast.parse(expr, mode="eval")
    return eval_expr(tree.body)