Function Calling — 工具调用

原理

Function Calling 让 LLM 决定何时调用外部工具，并生成结构化的调用参数，是构建 AI Agent 的核心机制。

用户: "招商银行现在股价多少？"
    ↓
LLM 判断: 需要调用 get_stock_price 工具
    ↓
返回: {"name": "get_stock_price", "arguments": {"symbol": "600036"}}
    ↓
代码执行工具，获取真实数据
    ↓
将结果返回给 LLM
    ↓
LLM 生成最终回答

完整实现

import json
from openai import OpenAI

client = OpenAI(api_key="sk-xxx", base_url="...")

# 工具定义
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_stock_price",
            "description": "获取A股实时股价和涨跌幅",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {"type": "string", "description": "股票代码，如600036"},
                    "fields": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "需要的字段：price, change, pe_ratio, market_cap"
                    }
                },
                "required": ["symbol"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_financial_news",
            "description": "搜索最新金融新闻",
            "parameters": {
                "type": "object",
                "properties": {
                    "keyword": {"type": "string"},
                    "days": {"type": "integer", "description": "最近几天，默认7"}
                },
                "required": ["keyword"]
            }
        }
    }
]

# 工具实现
def get_stock_price(symbol: str, fields: list = None) -> dict:
    # 实际调用行情 API
    return {"symbol": symbol, "price": 38.52, "change": "+1.2%", "pe_ratio": 8.5}

def search_financial_news(keyword: str, days: int = 7) -> list:
    return [{"title": f"{keyword}相关新闻", "date": "2024-01-15"}]

TOOL_MAP = {
    "get_stock_price": get_stock_price,
    "search_financial_news": search_financial_news
}

# Agent 循环
def run_agent(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]
    
    while True:
        response = client.chat.completions.create(
            model="qwen-plus",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )
        
        message = response.choices[0].message
        messages.append(message)
        
        # 没有工具调用，返回最终答案
        if not message.tool_calls:
            return message.content
        
        # 执行所有工具调用
        for tool_call in message.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)
            
            print(f"调用工具: {func_name}({func_args})")
            result = TOOL_MAP[func_name](**func_args)
            
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False)
            })

# 测试
result = run_agent("分析招商银行(600036)的最新股价和相关新闻，给出投资建议")
print(result)

并行工具调用

# 现代 LLM 支持一次调用多个工具
response = client.chat.completions.create(
    model="qwen-plus",
    messages=[{"role": "user", "content": "同时查询招商银行和平安银行的股价"}],
    tools=tools
)

# 可能返回多个 tool_calls
for tool_call in response.choices[0].message.tool_calls:
    print(f"并行调用: {tool_call.function.name}")

Pydantic 结构化输出

from pydantic import BaseModel
from typing import List

class StockAnalysis(BaseModel):
    symbol: str
    current_price: float
    recommendation: str  # 买入/持有/卖出
    target_price: float
    risk_factors: List[str]

# 强制 JSON 输出并验证
response = client.chat.completions.create(
    model="qwen-plus",
    messages=[{"role": "user", "content": "分析招商银行股票"}],
    response_format={"type": "json_object"}
)

data = json.loads(response.choices[0].message.content)
analysis = StockAnalysis(**data)
print(f"建议: {analysis.recommendation}, 目标价: {analysis.target_price}")