Python 模块化 × LLM 工具调用：构建可扩展的智能问答系统

一、环境搭建与依赖管理

1.1 安装必要库

bash
编辑
pip install openai requests python-dotenv  # 推荐加入 dotenv 管理密钥

• openai：官方 SDK，兼容所有 OpenAI-like API（DeepSeek、Moonshot、Ollama 等）
• requests：发起 HTTP 请求（用于调用天气/股票 API）
• python-dotenv：从 .env 文件加载环境变量，避免密钥泄露

1.2 配置 API 密钥（安全实践）

创建 .env 文件：

env
编辑
DEEPSEEK_API_KEY=sk-3d61560ac9f74daeaed1b8c53e2b7a5a
SENIVERSE_KEY=SeZYXHV7f3RJvQaZH

Python 中加载：

python
编辑
from dotenv import load_dotenv
import os

load_dotenv()
api_key = os.getenv("DEEPSEEK_API_KEY")

二、LLM 客户端初始化：兼容性设计

python
编辑
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1"
)

深度解析：

• OpenAI 类是 SDK 提供的统一入口，无论后端是 GPT、DeepSeek 还是本地 Ollama，接口一致。
• base_url 允许切换不同厂商的 OpenAI 兼容端点，实现“一次开发，多端部署”。
• DeepSeek 的 deepseek-reasoner 模型特别支持 reasoning_content 字段（显示思维链），适合教学与调试。

三、模块化核心：消息发送函数设计

python
编辑
def send_message(messages, tools=None, model="deepseek-reasoner"):
    return client.chat.completions.create(
        model=model,
        messages=messages,
        tools=tools,
        tool_choice="auto" if tools else None,
        temperature=0.3,  # 降低随机性，提升工具调用稳定性
        max_tokens=1000
    )

关键参数详解：

四、工具定义：OpenAI Tool Schema 深度解读

python
编辑
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的当前天气，仅支持中国大陆城市",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "城市拼音或中文名，如'beijing'或'北京'"
                    }
                },
                "required": ["location"]
            }
        }
    }
]

Schema 各字段含义：

• type: 目前仅支持 "function"，未来可能扩展 "retrieval"、"code_interpreter" 等。
• name: 函数标识符，必须与本地 Python 函数名一致。
• description: 极其重要！  LLM 依靠此描述判断何时调用该工具。应清晰、具体、带约束。
• parameters: 使用 JSON Schema 描述输入格式，确保 LLM 生成合法参数。

五、本地函数实现：健壮性与错误处理

python
编辑
import requests
import json

def get_weather(location: str) -> str:
    url = "https://api.seniverse.com/v3/weather/now.json"
    params = {
        "key": os.getenv("SENIVERSE_KEY"),
        "location": location,
        "language": "zh-Hans"
    }
    
    try:
        resp = requests.get(url, params=params, timeout=8)
        resp.raise_for_status()  # 抛出 HTTP 错误
        
        data = resp.json()
        if not data.get("results"):
            return f"未找到城市 '{location}' 的天气信息，请检查名称是否正确。"
            
        r = data["results"][0]
        city = r["location"]["name"]
        now = r["now"]
        return f"{city}当前天气：{now['text']}，气温 {now['temperature']}℃"
        
    except requests.exceptions.Timeout:
        return "天气服务响应超时，请稍后再试。"
    except requests.exceptions.RequestException as e:
        return f"网络请求失败：{str(e)}"
    except (KeyError, json.JSONDecodeError):
        return "天气服务返回格式异常。"
    except Exception as e:
        return f"未知错误：{repr(e)}"

错误处理层次：

1. 网络层：超时、DNS 失败、SSL 错误 → requests.exceptions
2. HTTP 层：404、500 → resp.raise_for_status()
3. 数据层：JSON 解析失败、字段缺失 → KeyError, JSONDecodeError
4. 兜底层：捕获所有异常，避免程序崩溃

六、完整工具调用流程：两阶段交互详解

这是整个系统的核心逻辑，分为两个阶段：

阶段 1：LLM 决策是否调用工具

python
编辑
messages = [{"role": "user", "content": "上海明天会下雨吗？"}]
response = send_message(messages, tools=tools)
msg = response.choices[0].message

此时，msg 可能包含：

• content: 一段文字（无需工具）
• tool_calls: 一个列表，每个元素包含 id, function.name, function.arguments

阶段 2：执行工具并回传结果

python
编辑
if msg.tool_calls:
    messages.append(msg)  # 保存 LLM 的决策
    
    for tool_call in msg.tool_calls:
        func_name = tool_call.function.name
        args = json.loads(tool_call.function.arguments)  # 注意：arguments 是 JSON 字符串！
        
        # 动态调用函数（可用字典映射替代 if-else）
        available_functions = {
            "get_weather": get_weather,
            "get_closing_price": get_closing_price
        }
        func = available_functions.get(func_name)
        result = func(**args) if func else "未知工具"
        
        # 将结果作为 'tool' 消息加入对话历史
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "name": func_name,
            "content": result
        })
    
    # 阶段 3：LLM 整合结果生成最终回答
    final_resp = send_message(messages)
    print(final_resp.choices[0].message.content)

关键细节：

• tool_call_id：用于关联“请求”与“响应”，确保多工具调用时不混淆。
• role="tool"：这是 OpenAI 协议规定的特殊角色，LLM 会识别此消息为函数返回值。
• 对话历史必须完整保留：包括 system、user、assistant、tool 消息，否则 LLM 无法理解上下文。

七、扩展：支持多工具与动态注册

为支持未来新增工具（如查快递、翻译），可设计注册机制：

python
编辑
class ToolRegistry:
    def __init__(self):
        self.functions = {}
        self.schemas = []
    
    def register(self, func, schema):
        name = func.__name__
        self.functions[name] = func
        self.schemas.append(schema)
        return func

registry = ToolRegistry()

# 注册天气工具
@registry.register
def get_weather(location: str) -> str:
    # ... 实现同上 ...

weather_schema = { /* 同上 */ }
registry.register(get_weather, weather_schema)

# 调用时
tools = registry.schemas
func = registry.functions[tool_call.function.name]

八、性能与安全考量

8.1 性能优化

• 缓存：对高频查询（如茅台股价）加内存缓存（functools.lru_cache）
• 异步：使用 httpx.AsyncClient + async/await 提升并发能力
• 限流：在函数内加计数器，防止恶意用户刷接口

8.2 安全防护

• 输入校验：对 location 做白名单或正则过滤，防止 SSRF（如 location=）
• 沙箱执行：若工具涉及代码执行（如 code_interpreter），务必在隔离环境中运行
• 日志审计：记录所有工具调用，便于追踪与分析

九、调试技巧：如何排查工具调用失败？

1. 打印完整 messages：确认对话历史格式正确
2. 检查 arguments 是否合法 JSON：常见错误是 LLM 生成了非 JSON 字符串
3. 验证 function name 是否匹配：大小写敏感！
4. 查看模型是否支持 tool calling：并非所有模型都支持（如 gpt-3.5-turbo-instruct 不支持）

示例调试代码：

python
编辑
print("LLM 返回:", msg)
if msg.tool_calls:
    for call in msg.tool_calls:
        print("欲调用函数:", call.function.name)
        print("参数 (raw):", call.function.arguments)
        print("参数 (parsed):", json.loads(call.function.arguments))

十、总结与展望

核心收获