Windows 下 llama.cpp 编译与 Qwen 模型本地部署实战

Python 客户端集成

为了更灵活地处理业务逻辑，推荐使用 Python 编写调用脚本。以下示例展示了基础对话、多轮记忆及工具函数调用的完整实现。

基础对话与多轮记忆

import requests
import json

chat_history = []
url = "http://localhost:11433/chat/completions"
headers = {"Content-Type": "application/json"}

def chat_with_model(prompt):
    # 添加当前用户消息到历史
    chat_history.append({"role": "user", "content": prompt})
    
    data = {
        "model": "qwen.gguf",
        "messages": chat_history,
        "temperature": 0.7,
        "max_tokens": 512
    }
    
    try:
        response = requests.post(url, headers=headers, data=json.dumps(data), timeout=60)
        response.raise_for_status()
        result = response.json()
        answer = result["choices"][0]["message"]["content"]
        
        # 将助手回答加入历史，保持上下文连贯
        chat_history.append({"role": "assistant", "content": answer})
        return answer
    except Exception as e:
        return f"调用失败：{e}"

# 多轮对话测试
print("开始多轮对话（输入'退出'结束）：")
while True:
    user_input = input("你：")
    if user_input == "退出":
        break
    if not user_input.strip():
        print("助手：请输入有效内容！\n")
        continue
    answer = chat_with_model(user_input)
    print(f"助手：{answer}\n")

工具函数调用（Function Calling）

对于需要外部数据或计算的场景，可以启用工具调用功能。以下代码演示了如何定义工具集并让模型自动决策调用。

import requests
import json
import re
from datetime import datetime

# ====================== 1. 定义可用工具集 ======================
def get_current_time():
    """获取当前的本地时间"""
    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    return f"当前时间为：{current_time}"

def calculate_add(a: float, b: float):
    """计算两个数的加法结果"""
    return f"{a} + {b} = {a + b}"

# 工具注册表（核心：映射工具名到函数和描述）
tool_registry = {
    "get_current_time": {
        "function": get_current_time,
        "description": "获取当前的本地时间，无需参数",
        "parameters": {}
    },
    "calculate_add": {
        "function": calculate_add,
        "description": "计算两个数字的加法，需要参数 a 和 b",
        "parameters": {
            "a": {"type": "float", "required": True, "description": "加数 1"},
            "b": {"type": "float", "required": True, "description": "加数 2"}
        }
    }
}

# ====================== 2. 初始化对话历史和基础配置 ======================
chat_history = [{"role": "system", "content": """你是一个有帮助的助手，必须记住之前的对话内容，基于上下文回答用户问题。
你可以调用以下工具来辅助回答：
1. get_current_time：获取当前的本地时间
2. calculate_add：计算两个数字的加法
如果需要调用工具，请严格按照 JSON 格式返回。"""}]

url = "http://localhost:11433/chat/completions"
headers = {"Content-Type": "application/json"}

def clean_pad_content(content):
    """过滤模型返回的 [PAD...] 垃圾字符"""
    return re.sub(r'\[PAD\d+\]', '', content).strip()

def parse_tool_call(content):
    """解析模型返回的内容，提取工具调用指令"""
    try:
        json_match = re.search(r'\{[\s\S]*\}', content)
        if not json_match:
            return None
        tool_call = json.loads(json_match.group())
        if "name" in tool_call and "parameters" in tool_call:
            return tool_call
        return None
    except (json.JSONDecodeError, Exception):
        return None

def execute_tool(tool_call):
    """执行工具调用，返回执行结果"""
    tool_name = tool_call["name"]
    parameters = tool_call.get("parameters", {})
    
    if tool_name not in tool_registry:
        return f"错误：不存在名为 {tool_name} 的工具"
    
    tool_info = tool_registry[tool_name]
    tool_func = tool_info["function"]
    tool_params = tool_info["parameters"]
    
    # 参数类型转换
    for param_name, param_info in tool_params.items():
        if param_name in parameters:
            param_type = param_info.get("type", "str")
            if param_type == "float":
                parameters[param_name] = float(parameters[param_name])
            elif param_type == "int":
                parameters[param_name] = int(parameters[param_name])
    
    try:
        result = tool_func(**parameters)
        return f"工具调用成功（{tool_name}）：{result}"
    except Exception as e:
        return f"错误：执行 {tool_name} 失败 - {str(e)}"

# ====================== 3. 核心对话函数 ======================
def chat_with_model(prompt):
    global chat_history
    chat_history.append({"role": "user", "content": prompt})
    
    data = {
        "model": "qwen.gguf",
        "messages": chat_history,
        "temperature": 0.7,
        "max_tokens": 512,
        "stream": False,
        "stop": ["[PAD"]
    }
    
    try:
        response = requests.post(url, headers=headers, data=json.dumps(data), timeout=60)
        response.raise_for_status()
        result = response.json()
        
        if "choices" in result and len(result["choices"]) > 0:
            raw_answer = result["choices"][0]["message"]["content"]
            clean_answer = clean_pad_content(raw_answer)
        else:
            return f"返回格式异常"
        
        # 解析是否包含工具调用指令
        tool_call = parse_tool_call(clean_answer)
        if tool_call:
            print(f"📢 检测到工具调用：{json.dumps(tool_call, ensure_ascii=False)}")
            tool_result = execute_tool(tool_call)
            print(f"🔧 工具执行结果：{tool_result}")
            
            # 将工具执行结果加入对话历史
            chat_history.append({"role": "assistant", "content": f"工具调用结果：{tool_result}"})
            
            # 再次请求生成最终回答
            second_response = requests.post(url, headers=headers, data=json.dumps(data), timeout=60)
            second_result = second_response.json()
            final_answer = clean_pad_content(second_result["choices"][0]["message"]["content"])
            chat_history.append({"role": "assistant", "content": final_answer})
            return final_answer
        else:
            chat_history.append({"role": "assistant", "content": clean_answer})
            return clean_answer
    except requests.exceptions.ConnectionError:
        return "连接失败：请检查本地服务是否在 11433 端口运行"
    except requests.exceptions.Timeout:
        return "请求超时：模型响应过慢"
    except Exception as e:
        return f"调用失败：{str(e)}"

# ====================== 4. 测试入口 ======================
if __name__ == "__main__":
    print("开始多轮对话（输入'退出'结束）：")
    print("📌 测试工具调用示例：")
    print(" 1. 现在几点了？")
    print(" 2. 计算 123+456 等于多少？")
    print(" 3. 我的名字是李四，我叫什么？\n")
    
    while True:
        user_input = input("你：")
        if user_input.strip() == "退出":
            break
        if not user_input.strip():
            print("助手：请输入有效内容！\n")
            continue
        answer = chat_with_model(user_input)
        print(f"助手：{answer}\n")

通过以上步骤，你可以在 Windows 本地完成从编译到应用的全流程部署。如果遇到显存不足的问题，可以尝试加载更低精度的量化模型（如 Q4_K_M），或在启动时增加 --n-gpu-layers 参数来控制 GPU 层数分配。