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

多轮对话示例（基于 chat/completions）

import requests
import json

chat_history = []
url = "http://localhost:8080/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
    answer = chat_with_model(user_input)
    print(f"助手：{answer}\n")

带有对话记忆功能测试

import requests
import json
import re

# 初始化对话历史（包含系统提示，引导模型记上下文）
chat_history = [{"role": "system", "content": "你是一个有帮助的助手，必须记住之前的对话内容，基于上下文回答用户问题。"}]

# 你的服务实际地址
url = "http://localhost:8080/chat/completions"
headers = {"Content-Type": "application/json"}

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

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:
            choice = result["choices"][0]
            if "message" in choice and "content" in choice["message"]:
                raw_answer = choice["message"]["content"]
                answer = clean_pad_content(raw_answer)
                # 关键：将助手回复加入历史，下次请求会带上
                chat_history.append({"role": "assistant", "content": answer})
                return answer
            else:
                return f"返回格式异常：缺少 message/content 字段"
        else:
            return f"返回格式异常：缺少 choices 字段"
    except requests.exceptions.ConnectionError:
        return "连接失败：请检查本地服务是否在 8080 端口运行"
    except requests.exceptions.Timeout:
        return "请求超时：模型响应过慢"
    except Exception as e:
        return f"调用失败：{str(e)}"

# 多轮对话测试（重点测试上下文记忆）
print("开始多轮对话（输入'退出'结束）：")
print("提示：先发送 '我的名字是李四'，再发送 '我叫什么名字' 测试记忆功能\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")

函数工具调用测试

import requests
import json
import re
from datetime import datetime

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

# 工具 2：加法计算
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：计算两个数字的加法，需要参数 a 和 b（均为数字）如果需要调用工具，请严格按照以下 JSON 格式返回（仅返回 JSON，不要加其他内容）： {"name": "工具名", "parameters": {"参数名": 参数值}} 如果不需要调用工具，直接回答用户问题即可，不要返回 JSON 格式。"""}]

# 本地 LLM 服务地址
url = "http://localhost:8080/chat/completions"
headers = {"Content-Type": "application/json"}

# ====================== 3. 工具调用相关辅助函数 ======================
def clean_pad_content(content):
    """过滤模型返回的 [PAD...] 垃圾字符"""
    return re.sub(r'\[PAD\d+\]', '', content).strip()

def parse_tool_call(content):
    """解析模型返回的内容，提取工具调用指令（JSON 格式）"""
    try:
        # 提取 JSON 部分（兼容模型返回时可能带的多余文字）
        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} 的工具，可用工具：{list(tool_registry.keys())}"
    tool_info = tool_registry[tool_name]
    tool_func = tool_info["function"]
    tool_params = tool_info["parameters"]
    # 验证必填参数
    missing_params = []
    for param_name, param_info in tool_params.items():
        if param_info.get("required") and param_name not in parameters:
            missing_params.append(param_name)
    if missing_params:
        return f"错误：调用 {tool_name} 缺少必填参数：{', '.join(missing_params)}"
    # 转换参数类型（比如字符串转数字）
    try:
        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])
    except ValueError as e:
        return f"错误：参数类型转换失败 - {str(e)}"
    # 执行工具函数
    try:
        result = tool_func(**parameters)
        return f"工具调用成功（{tool_name}）：{result}"
    except Exception as e:
        return f"错误：执行 {tool_name} 失败 - {str(e)}"

# ====================== 4. 核心对话函数（支持工具调用） ======================
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 and "message" in result["choices"][0]:
            raw_answer = result["choices"][0]["message"]["content"]
            clean_answer = clean_pad_content(raw_answer)
        else:
            return f"返回格式异常：{json.dumps(result, ensure_ascii=False)[:300]}"
        # 解析是否包含工具调用指令
        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_response.raise_for_status()
            second_result = second_response.json()
            # 解析第二次调用的结果
            if "choices" in second_result and len(second_result["choices"]) > 0 and "message" in second_result["choices"][0]:
                final_answer = clean_pad_content(second_result["choices"][0]["message"]["content"])
                chat_history.append({"role": "assistant", "content": final_answer})
                return final_answer
            else:
                return f"工具调用后二次请求异常：{json.dumps(second_result, ensure_ascii=False)[:300]}"
        else:
            # 无需调用工具，直接返回模型回答
            chat_history.append({"role": "assistant", "content": clean_answer})
            return clean_answer
    except requests.exceptions.ConnectionError:
        return "连接失败：请检查本地服务是否在 8080 端口运行"
    except requests.exceptions.Timeout:
        return "请求超时：模型响应过慢"
    except Exception as e:
        return f"调用失败：{str(e)}"

# ====================== 5. 多轮对话测试（含工具调用） ======================
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")