Agent开发实战-调用本地模型

observer？

437人浏览 · 2026-04-10 18:04:53

observer？ · 2026-04-10 18:04:53 发布

安装方式（邪修版）

小龙虾自动安装，我用的是腾讯的Qclaw，直接给他指令，他就会自动下载并且安装。在安装过程中遇到的各种问题，他也会自动去解决。甚至生成速度过慢他也会想办法去解决。一个字：夯。

本地调用

本地调用需要使用llama.cpp，需要贵贱命令行格式，并且对输出格式进行处理。

我在这里安装了三个本地模型。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
本地 LLM GPU 调用脚本
支持 Qwen 2.5 7B 和 Hermes 3 8B，通过 llama.cpp 调用 RTX 5060 加速
"""

import subprocess
import os
import sys
import tempfile
import argparse
import re
import time

# ==================== 模型配置 ====================
MODELS = {
    "qwen": {
        "path":  r"D:\models\qwen2.5-7b\qwen2.5-7b-instruct-q4_k_m.gguf",
        "name":  "Qwen 2.5 7B Q4_K_M",
        "format": "chatml",
        "system": "你是一个有帮助的AI助手，请用中文回答。",
    },
    "hermes": {
        "path":  r"D:\models\hermes3-8b\Hermes-3-Llama-3.1-8B.Q4_K_M.gguf",
        "name":  "Hermes 3 Llama 3.1 8B Q4_K_M",
        "format": "chatml",
        "system": "You are a helpful assistant. Respond in Chinese when asked in Chinese.",
    },
    "qwopus": {
        "path":  r"D:\models\qwopus3.5-9b\Qwen3.5-9B.Q4_K_M.gguf",
        "name":  "Qwopus 3.5 9B v3 Q4_K_M (带思考)",
        "format": "chatml",
        "system": "你是一个有帮助的AI助手，请用中文回答。",
        "thinking": True,  # 支持深度思考模式
    },
}

LLAMA_CLI    = r"D:\software\llama-cpp-b8741\llama-cli.exe"
GPU_LAYERS   = 99
CPU_THREADS  = 6
CONTEXT_SIZE = 4096

DEFAULT_TEMP    = 0.7
DEFAULT_TOP_P  = 0.9
DEFAULT_MAX    = 512

# ==================== 格式构建 ====================

def build_prompt_chatml(messages: list[dict], system: str = "") -> str:
    """ChatML 格式（Hermes / Llama 3 通用）"""
    parts = []
    if system:
        parts.append(f"<|im_start|>system\n{system}<|im_end|>")
    for msg in messages:
        parts.append(f"<|im_start|>{msg['role']}\n{msg['content']}<|im_end|>")
    parts.append("<|im_start|>assistant\n")
    return "\n".join(parts)


def build_prompt_qwen(messages: list[dict], system: str = "") -> str:
    """Qwen ChatML 格式"""
    return build_prompt_chatml(messages, system)


# ==================== 核心调用 ====================

def chat(
    prompt: str,
    model_key: str = "qwen",
    temperature: float = DEFAULT_TEMP,
    max_tokens: int = DEFAULT_MAX,
    history: list[dict] | None = None,
) -> str:
    """
    与本地模型对话

    Args:
        prompt:     用户输入
        model_key:  模型 key ("qwen" 或 "hermes")
        temperature: 温度
        max_tokens:  最大生成 token 数
        history:    对话历史
    """
    cfg = MODELS.get(model_key, MODELS["qwen"])
    messages = list(history) if history else []
    messages.append({"role": "user", "content": prompt})

    formatted = build_prompt_chatml(messages, cfg["system"])

    with tempfile.NamedTemporaryFile(
        mode="w", suffix=".txt", delete=False, encoding="utf-8"
    ) as f:
        f.write(formatted)
        prompt_file = f.name

    try:
        cmd = [
            LLAMA_CLI,
            "-m", cfg["path"],
            "-f", prompt_file,
            "-n", str(max_tokens),
            "-ngl", str(GPU_LAYERS),
            "-t", str(CPU_THREADS),
            "--temp", str(temperature),
            "--top-p", str(DEFAULT_TOP_P),
            "-c", str(CONTEXT_SIZE),
            "--no-display-prompt",
            "--single-turn",      # 单轮生成后退出
            "--log-disable",      # 禁用日志输出
        ]

        env = os.environ.copy()
        env["KMP_DUPLICATE_LIB_OK"] = "TRUE"

        t0 = time.time()
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="ignore",
            env=env,
            timeout=120,
        )
        elapsed = time.time() - t0

        text = _extract(result.stdout + result.stderr)
        stats = _extract_perf(result.stdout + result.stderr)
        stats["elapsed"] = round(elapsed, 2)

        if stats.get("eval_speed"):
            print(f"  ⚡ {stats['eval_speed']} tok/s | {elapsed:.1f}s", file=sys.stderr)

        return text
    finally:
        os.unlink(prompt_file)


# ==================== 辅助函数 ====================

_LOG_KEYWORDS = [
    "llama_model_loader:", "llama_model_load:", "llm_load_",
    "llama_perf_", "ggml_cuda_init:", "GGML_CUDA",
    "common_init_from_params:", "system_info:",
    "sampler chain:", "sampler seed:", "sampler params:",
    "generate:", "repeat_last_n", "dry_multiplier", "dry_penalty",
    "top_k:", "top_p:", "min_p:", "typical_p:", "mirostat",
    "penalty_prompt:", "penalty_freq:", "penalty_present:",
    "VMM:", "build      :", "model      :", "modalities :",
    "available commands:", "/exit", "/regen", "/clear", "/read", "/glob",
    "Loading model", "Exiting...", "load_backend:",
    "offloaded", "llama_context:", "▄", "██",
    "kv self size", "graph nodes", "graph splits",
    "Prompt:", "Generation:",
]


def _is_log(line: str) -> bool:
    stripped = line.strip()
    if not stripped:
        return True
    # 进度点行 / banner 图案
    if len(stripped) < 3 and set(stripped.replace(" ", "")) <= {".▄▀█"}:
        return True
    # llama banner 图案行
    if set(stripped.replace(" ", "")) <= {"▄", "▀", "█"}:
        return True
    return any(kw in line for kw in _LOG_KEYWORDS)


def _extract(output: str) -> str:
    lines = output.split("\n")
    parts = [l for l in lines if not _is_log(l)]
    text = "\n".join(parts).strip()
    # 过滤 prompt 显示行
    text = re.sub(r"^>\s*<\|im_start\|>.*$", "", text, flags=re.MULTILINE)
    text = re.sub(r"^\s*(user|assistant|system)\s*\n", "", text, flags=re.I)
    # 只保留 assistant 之后的回答（去掉回显的 system/user prompt）
    m = re.search(r"<\|im_start\|>assistant\s*\n", text)
    if m:
        text = text[m.end():]
    text = text.replace("[end of text]", "").strip()
    # 提取思考内容
    thinking = ""
    m = re.search(r"\[Start thinking\](.*?)\[End thinking\]", text, re.DOTALL)
    if m:
        thinking = m.group(1).strip()
    # 提取最终回答（思考部分之后）
    if "[End thinking]" in text:
        answer = text.split("[End thinking]", 1)[-1].strip()
    else:
        answer = text
    # 如果有思考内容，格式化输出
    if thinking:
        return f"💭 思考:\n{thinking}\n\n📝 回答:\n{answer}"
    return answer


def _extract_perf(output: str) -> dict:
    stats = {}
    # 旧格式: eval time = xxx ms / N runs (xx ms per token, yy tokens per second)
    m = re.search(r"eval time\s+=\s+[\d.]+ ms\s*/\s+\d+ runs\s+\(\s*[\d.]+ ms per token,\s*([\d.]+)\s*tokens per s", output)
    if m:
        stats["eval_speed"] = round(float(m.group(1)), 1)
    # 新格式: [ Prompt: xxx t/s | Generation: yyy t/s ]
    m = re.search(r"Generation:\s*([\d.]+)\s*t/s", output)
    if m:
        stats["eval_speed"] = round(float(m.group(1)), 1)
    m = re.search(r"Prompt:\s*([\d.]+)\s*t/s", output)
    if m:
        stats["prompt_speed"] = round(float(m.group(1)), 1)
    m = re.search(r"offloaded\s+(\d+)/(\d+) layers to GPU", output)
    if m:
        stats["gpu_layers"] = f"{m.group(1)}/{m.group(2)}"
    return stats


# ==================== 交互模式 ====================

def interactive(model_key: str = "qwen"):
    cfg = MODELS.get(model_key, MODELS["qwen"])
    print("=" * 50)
    print(f"  {cfg['name']} · GPU 模式 · RTX 5060")
    print("  输入 'quit' 退出, 'clear' 清空历史")
    print("=" * 50)
    print()

    history = []
    while True:
        try:
            user_input = input("你: ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n再见！")
            break

        if not user_input:
            continue
        if user_input.lower() in ("quit", "exit"):
            print("再见！")
            break
        if user_input.lower() == "clear":
            history = []
            print("✅ 历史已清空\n")
            continue

        print("助手: ", end="", flush=True)
        t0 = time.time()
        reply = chat(user_input, model_key=model_key, history=history)
        print(reply)
        print(f"  ({time.time() - t0:.1f}s)\n")

        history += [{"role": "user", "content": user_input},
                    {"role": "assistant", "content": reply}]
        if len(history) > 20:
            history = history[-16:]


# ==================== 入口 ====================

def main():
    parser = argparse.ArgumentParser(
        description="本地 LLM GPU 调用工具 (Qwen / Hermes)",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  python local_llm.py -m qwen    -p "你好"
  python local_llm.py -m hermes  -p "你好"
  python local_llm.py -m qwopus  -p "1+1等于几？"
  python local_llm.py -m qwen    -i
  python local_llm.py -m qwopus  -i
  python local_llm.py -m qwen    -p "写个快排" --code
        """,
    )
    parser.add_argument("-m", "--model", choices=["qwen", "hermes", "qwopus"], default="qwen",
                        help="选择模型 (默认 qwen)")
    parser.add_argument("-p", "--prompt", type=str, help="单次提问")
    parser.add_argument("-i", "--interactive", action="store_true", help="交互模式")
    parser.add_argument("-n", "--max-tokens", type=int, default=DEFAULT_MAX, help="最大 token 数")
    parser.add_argument("--temp", type=float, default=DEFAULT_TEMP, help="温度")
    parser.add_argument("--code", action="store_true", help="编程模式 (低温)")

    args = parser.parse_args()

    cfg = MODELS[args.model]
    if not os.path.exists(cfg["path"]):
        print(f"❌ 模型文件不存在: {cfg['path']}")
        sys.exit(1)

    temp = 0.2 if args.code else args.temp

    if args.interactive:
        interactive(args.model)
    elif args.prompt:
        print(f"模型: {cfg['name']}")
        print(f"温度: {temp}")
        print()
        reply = chat(args.prompt, model_key=args.model, temperature=temp,
                     max_tokens=args.max_tokens)
        print(reply)
    else:
        parser.print_help()


if __name__ == "__main__":
    main()

架构升级

在真实的企业级架构中，大模型是作为常驻服务 (Daemon Service) 运行在显存里的。有以下两种方案改造：

方案 A：使用 `llama-cpp-python`（Python 原生绑定）

这是目前最主流的本地开发方式。模型只在类初始化时加载一次，之后的调用全是在内存中直接前向传播。

# pip install llama-cpp-python
from llama_cpp import Llama

# 初始化时加载一次模型（常驻显存）
llm = Llama(model_path="qwen2.5-7b...gguf", n_gpu_layers=-1, n_ctx=4096)

# 调用时瞬间响应
output = llm(
      "你想问的问题", 
      max_tokens=512, 
      echo=False
)

方案 B：启动 `llama-server`（微服务架构，强烈推荐！）

你下载的 llama.cpp 工具包里，除了 llama-cli.exe，一定还有一个 llama-server.exe。
你可以用它在后台启动一个 兼容 OpenAI API 格式 的本地服务器：

# 在终端后台运行这个（只加载一次模型）
llama-server.exe -m D:\models\qwen2.5-7b\qwen2.5-7b-instruct-q4_k_m.gguf -ngl 99 -c 4096 --port 8080

然后你的 Python 代码就可以彻底脱离繁重的子进程管理，直接像调用 ChatGPT 一样调用你的本地模型：

import requests

def chat_api(prompt):
    res = requests.post(
        "http://127.0.0.1:8080/v1/chat/completions",
        json={
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.7
        }
    )
    return res.json()["choices"][0]["message"]["content"]

选用方案 B 的巨大优势： 几乎所有的 Agent 框架（如 LangChain、LangGraph、AutoGen）都默认支持 OpenAI API 格式。你只要把 API Base 指向你的本地 8080 端口，你就可以无缝使用所有业界最顶级的框架！

API 常驻微服务

采用 llama-server 后，你的显卡将化身为一台私有云服务器。模型权重（4GB+）只会在第一次启动时加载到显存中，之后的每一次问答都会瞬间响应，再也不需要经历漫长的冷启动。

更重要的是：llama-server 提供的是完全兼容 OpenAI 标准的 API 接口。这意味着你后续使用 LangChain、LangGraph 时，可以像调用 ChatGPT 一样直接调用你的本地 Qwen！

第一步：创建后台常驻服务 (启动你的“私有云大脑”)

不要每次都在终端里敲长长的命令，我们在你的项目根目录下建一个批处理脚本。

在你的项目文件夹下，新建一个文本文件，命名为 start_llm_server.bat。
将以下命令复制进去并保存：

@echo off
title Qwen 2.5 7B API Server
echo 正在启动本地大模型 API 服务...

D:\software\llama-cpp-bin\llama-server.exe -m D:\models\qwen2.5-7b\qwen2.5-7b-instruct-q4_k_m.gguf -c 4096 -ngl 99 --port 8080 --host 127.0.0.1

pause

实操： 双击运行这个 .bat 文件。你会看到命令行飞速闪过加载日志，最后停留在 HTTP server listening on 127.0.0.1:8080。让这个黑窗口一直开着，这就是你的“常驻大脑”。

第二步：编写标准化 API 客户端 (替换你之前的代码)

既然服务端完全兼容 OpenAI，我们就直接使用工业界最标准的 openai 官方 Python 库来调用它。

安装依赖：

pip install openai

新建文件 llm_engine.py：
这个类将专门负责和你的后台大模型通信，代码极其简洁优雅：

from openai import OpenAI
import time

class LocalLLMEngine:
    def __init__(self, base_url="http://127.0.0.1:8080/v1"):
        """
        初始化本地 API 客户端
        base_url: 指向你刚才启动的 llama-server
        """
        print(f"[*] 正在连接本地大模型服务: {base_url}")
        self.client = OpenAI(
            base_url=base_url,
            api_key="sk-no-key-required" # 本地模型不需要真实秘钥
        )
        
    def chat(self, prompt, system_msg="你是一个专业的AI助手。", temperature=0.7, stream=True):
        """
        发起对话 (支持流式输出，打字机效果)
        """
        messages = [
            {"role": "system", "content": system_msg},
            {"role": "user", "content": prompt}
        ]
        
        # 记录耗时
        t0 = time.time()
        
        # 发起 API 请求
        response = self.client.chat.completions.create(
            model="qwen2.5-7b", # 这里的名字可以随便写，因为本地只有一个模型
            messages=messages,
            temperature=temperature,
            stream=stream # 开启流式输出
        )
        
        print("\n🤖 Qwen: ", end="", flush=True)
        full_reply = ""
        
        if stream:
            for chunk in response:
                if chunk.choices[0].delta.content:
                    content = chunk.choices[0].delta.content
                    print(content, end="", flush=True)
                    full_reply += content
            print(f"\n\n[⏱️ 耗时: {time.time() - t0:.2f} 秒]")
            return full_reply
        else:
            ans = response.choices[0].message.content
            print(ans)
            print(f"\n[⏱️ 耗时: {time.time() - t0:.2f} 秒]")
            return ans

# 测试一下瞬间响应的快感：
if __name__ == "__main__":
    llm = LocalLLMEngine()
    llm.chat("你好！请用一句话介绍一下紫微斗数。")