核心能力：大模型识别用户意图，自动选择调用自定义 Python 函数（工具：查天气、学历验证），拿到工具结果后再交给大模型整理成自然语言回答。

环境依赖：需要本地运行 Ollama，拉取 qwen3.5:9b 模型；本地启动 Redis；安装依赖

bash

pip install openai requests redis

一、导入模块 & 初始化 Redis 客户端

python

运行

import json
import os
import random
import requests
from openai import OpenAI
from redis import Redis

redis_client = Redis(
    port=6379,
    host="127.0.0.1",
    decode_responses=True
)
  1. json：解析工具返回字符串、序列化字典；
  2. requests：调用外部学历查询 http 接口；
  3. openai.OpenAI：兼容 OpenAI 接口，对接 Ollama，Ollama 提供兼容 OpenAI 的/v1接口；
  4. Redis：做缓存，同一个验证码不要重复请求第三方学历接口；
  5. decode_responses=True：redis 取出数据直接返回字符串，不返回 bytes 字节。

⚠️ 前提：本地 redis 服务必须启动 redis-server，否则直接抛连接报错。

二、定义两个工具函数（可供大模型调用）

1）模拟天气工具 get_current_weather

python

运行

def get_current_weather(arguments):
    weather_conditions = ["晴天", "多云", "雨天"]
    random_weather = random.choice(weather_conditions)
    location = arguments["location"]
    return f"{location}今天是{random_weather}。"
  • 入参 arguments：是字典，大模型输出的 json 参数，例如 {"location":"北京"}
  • 逻辑：随机返回天气字符串，纯模拟，没有真实 http 请求。

2）学历验证工具 academic_credential_verification

python

运行

def academic_credential_verification(arguments):
    vcode = arguments["vcode"]
    key = f"boss:llm:academic_credential_verification:{vcode}"
    redis_verification_data = redis_client.get(key)
    if redis_verification_data is None:
        # redis没缓存，调用第三方API
        BASE_URL = "https://www.apimy.cn/api/xxw/bgcx"
        payload = {"key": "MY_KEY_ztYvBsjT8NdGhlhiVck3RIrTNKvxEa", "vcode": arguments["vcode"]}
        headers = {"Content-Type": "application/json"}
        response = requests.post(BASE_URL, json=payload, headers=headers, timeout=30)
        response.raise_for_status()
        data = response.json()
        redis_client.set(key, json.dumps(data, ensure_ascii=False))
        return json.dumps(data, ensure_ascii=False)
    else:
        # redis命中缓存，直接返回缓存字符串，不再请求第三方接口
        return redis_verification_data

业务逻辑：

  1. 从参数取出验证码vcode；
  2. 构造 redis key，查询缓存；
  3. 缓存未命中：post 请求第三方学历校验接口，拿到返回 json，存入 redis 缓存，返回 json 字符串；
  4. 缓存命中：直接读取 redis 返回，减少第三方接口调用；
  • response.raise_for_status()：http 状态码 4xx/5xx 直接抛出异常；
  • json.dumps(...,ensure_ascii=False)：中文不转义 unicode。

注意：第三方 api 密钥硬编码在代码里，生产环境建议放到环境变量，不要写死。

三、tools 工具描述数组（Function Calling 元数据）

python

运行

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "当你想查询指定城市的天气时非常有用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "城市或县区，比如北京市、杭州市、余杭区等。",
                    }
                },
                "required": ["location"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "academic_credential_verification",
            "description": "当你想查询学历或者验证学历时非常有用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "vcode": {
                        "type": "string",
                        "description": "学历验证码",
                    }
                },
                "required": ["vcode"],
            },
        },
    }
]

这一段是传给大模型的 Schema，不是 Python 函数本身！

  • name：函数名字，大模型要调用哪个函数就写这个名字；
  • description：告诉大模型这个工具是干什么的，写的越清楚大模型越会正确选择工具；
  • parameters：JSON Schema，定义入参字段名、类型、描述、哪些参数必填；
  • required 必填参数列表；

大模型不会直接执行你的 python 函数，它只会输出一段 json：告诉我调用哪个函数，参数是什么；真正执行函数的是我们自己的 Python 程序。这是 Function Calling 核心！

四、初始化 OpenAI 客户端（对接本地 Ollama）

python

运行

client = OpenAI(
    api_key="sk-ws-H.ERMEXXD.mE3s.MEQCIDgda4zpgMUgFVv04fGbqygseUALajLCmOH6K3GKz5nYAiA2JMNBBq4_P3pB2GsCHF2BY7S-0gVg2Ov7g4xp7wBeFg",
    base_url="http://127.0.0.1:11434/v1",
)
messages = []
  • Ollama 开启 OpenAI 兼容接口地址就是 http://127.0.0.1:11434/v1；
  • api_key 在这里Ollama 不校验，随便写一个字符串占位即可；
  • messages = []：对话上下文数组，保存历史对话，role 区分：user、assistant、tool。

封装请求函数

python

运行

def get_ai_response(messages):
    completion = client.chat.completions.create(
        model="qwen3.5:9b",
        messages=messages,
        temperature=0.75,
        tools=tools
    )
    return completion
  • model="qwen3.5:9b"：本地 ollama 运行的模型名称；
  • temperature=0.75：创造性，越高回答越发散；
  • tools=tools：把上面工具 schema 传给大模型，开启 function‑calling 能力； 返回的 completion 是 OpenAI 格式响应对象。

五、主业务流程

python

运行

# 用户提问
user_message = {"role": "user", "content": "帮我查询一下学历, 验证码是:AZL698MXS2X56QB5"}
messages.append(user_message)

# 第一次请求大模型
completion = get_ai_response(messages)
print(completion.model_dump_json())
# 把大模型assistant的响应加入上下文
messages.append(completion.choices[0].message)
  1. 构造用户消息加入对话上下文；
  2. 第一次调用大模型，此时大模型看到用户要查学历，结合 tools 描述，会输出tool_calls，而不是直接输出自然语言；
  3. model_dump_json()把返回对象打印为完整 json 方便调试；
  4. 将 assistant 消息追加到 messages 上下文。

判断是否要调用工具

python

运行

if completion.choices[0].message.tool_calls is None:
    print("不需要调用工具")
    print(completion.choices[0].message.content)

else:
    print("需要调用工具")
    tool_calls = completion.choices[0].message.tool_calls
    for tool_call in tool_calls:
        tool_id = tool_call.id
        func_name = tool_call.function.name
        func_arguments = tool_call.function.arguments
        print(f"大模型告诉程序要调用这个工具:{func_name},参数是:{func_arguments}")
        
        # 函数映射表：大模型返回函数名字字符串 → 真实Python函数对象
        function_mapping = {
            "get_current_weather": get_current_weather,
            "academic_credential_verification": academic_credential_verification
        }
        print(type(func_arguments))
        
        # ⚠️重点：大模型返回arguments是JSON字符串，必须json.loads转字典再传入函数
        tool_result = function_mapping[func_name](json.loads(func_arguments))
        print(f"工具返回的结果是:{tool_result}")
        
        # 构造tool角色消息，放回对话上下文
        tool_message = {
            "content": tool_result,
            "role": "tool",
            "tool_call_id": tool_id
        }
        messages.append(tool_message)

    # 第二次请求大模型：把工具返回结果塞回去，让大模型整理成人类可读回答
    completion = get_ai_response(messages)
    print(completion.model_dump_json())
    print(f"最终的结果是:{completion.choices[0].message.content}")

完整 Function Calling 执行流程拆解（这个是核心）

  1. 第一轮请求 LLM：用户问查学历，LLM 识别意图，返回tool_calls（函数名 + 参数字符串），不输出最终答案。
  2. 程序接管：
    • 解析tool_calls；
    • function_mapping做名字映射，拿到真实 python 函数；
    • json.loads(func_arguments)：把大模型输出的 json 字符串转 python 字典，传给工具函数；
    • 执行自定义工具函数（走 redis 缓存、调用第三方 http 接口）拿到原始工具返回结果；
  3. 构造 tool 消息：固定 role=tool，带上tool_call_id（必须和前面 tool_call 的 id 一一对应，大模型才能匹配），追加到 messages；
  4. 第二轮请求 LLM：把工具原始数据喂给大模型，大模型基于工具返回原始数据整理成通顺的自然语言输出，给用户看最终结果。

messages 完整流转： user消息 → assistant(tool_calls消息) → tool(工具返回结果消息) → assistant(最终自然语言回答)

关键坑点说明（代码里容易踩的）

  1. func_arguments 是JSON 字符串，不能直接传进函数，必须json.loads()转字典；
  2. tool_call_id必须前后对应，tool 消息的 tool_call_id 必须等于前面 assistant tool_calls 的 id，否则 ollama/qwen 会报错；
  3. messages 要完整保存全部轮次，不能丢消息；
  4. Ollama 不是所有模型都很好支持 Function Calling，qwen 系列效果好；
  5. Redis 不启动直接报错连接拒绝；
  6. 第三方接口异常没有 try‑except，线上需要增加异常捕获。

执行样例流程

输入：帮我查询一下学历, 验证码是:AZL698MXS2X56QB5

  1. LLM 输出 tool_call，调用academic_credential_verification，参数{"vcode":"AZL698MXS2X56QB5"}
  2. Python 执行函数，redis 无缓存，请求第三方学历接口，写入 redis 缓存；
  3. 将接口原始 json 结果以 tool 消息丢回上下文；
  4. LLM 读取接口原始 json，整理成中文自然语言输出最终回答。
# Ollama # 后端
Logo
亚马逊云科技技术品牌专区

更多推荐

cover

DeepSeek API涨价500%！Agent框架同步开源

avatar 亚马逊云科技技术品牌专区
cover

Agent Run 的状态机：别把智能体交给 LLM 自己管

avatar 亚马逊云科技技术品牌专区
cover

【python】【FastAPI 进阶】《FastAPI 异步内核剖析：中间件链与 SQLAlchemy 异步驱动的全链路非阻塞实践》

avatar 亚马逊云科技技术品牌专区