1.ollama

ollama模型内外网都是win环境，所以就是模型的打包导入和python环境的打包导入。

1.1.模型

1.1.1下载

首先ollama下载并启动模型

ollama pull deepseek-r1:32b #1.ollama模型仓库下载模型到本地C:\Users\<用户名>\.ollama\models\blobs里面【ollama默认的模型存放地址，如果想修改，可以修改设置环境变量OLLAMA_MODELS来更改】
ollama serve #2.启动ollama服务
ollama run deepseek-r1:32b #3.启动模型命令行交互（ctrl+d退出）

1.1.2.打包

我内外网都是用的是ollama，外网pull下来的模型文件，通过gguf和Modelfile拿到内网使用。
1.ollama下载的模型文件在C:\Users<用户名>.ollama\models\blobs里面【ollama默认的模型存放地址，如果想修改，可以修改设置环境变量OLLAMA_MODELS来更改】，最大的那个就是模型，复制出来，改名为deepseek-r1.gguf

ollama list #2.查看模型文件叫什么(deepseek-r1:32b)
ollama show deepseek-r1:32b --modelfile > Modelfile  #3.得到模型的Modelfile文件

4.修改Modelfile文件里面的模型路径，例如这里是./deepseek-r1.gguf

1.1.3.部署

1.将这两个文件复制到内网的空文件夹

ollama create deepseek-r1 -f Modelfile #2.将这个gguf模型文件就解到了c盘的cache目录下
ollama run deepseek-r1 #3.此时就可以启动这个模型了

如果模型启动后在局域网下访问不到，修改环境变量OLLAMA_HOST=0.0.0.0

1.2.python环境

环境我内网都使用的是anconda创建的虚拟环境，外网环境打包成压缩包，在内网直接解压使用。

1.2.1.打包

conda activate chatglm#1.激活我外网使用的anconda环境
conda install -c conda-forge conda-pack #2.使用conda-forge通道下载conda-pack包，后续打包使用
conda pack -n chatglm -o environment.tar.gz #3.将这个环境下的包全部打到environment.tar.gz压缩包里面

1.2.2.部署

mkdir v2 #1.内网电脑新建文件夹
tar -xzf environment.tar.gz -C v2 #2.将打包的压缩包解压到v2目录下
.\v2\Scripts\activate #3.使用conda激活这个v2环境

1.2.3.模型封装

使用openai接口包装模型地调用方式

from langchain.llms.base import LLM
from langchain_core.messages.ai import AIMessage
from openai import OpenAI
from flask import Flask, request, jsonify, stream_with_context
import re
import json
import time
from sm import session_manager 
class MYLLM(LLM):
    client:object = None
    def __init__(self):
        super().__init__()
        self.client = OpenAI(
    base_url="http://192.168.3.126:11434/v1",  
    api_key="ollama"
)
    
    @property
    def _llm_type(self):
        return "deepseek-r1:8b"
    
    def invoke(self,user_id,user_input):
        if not isinstance(user_input,list):#prompt,否则就是history
            history=[{"role":"user","content":user_input}]
        else:
            history=user_input
        response = self.client.chat.completions.create(
            model="deepseek-r1:8b",#"deepseek-r1",
            messages=history, 
        )
        result = response.choices[0].message.content
        session_manager.update_history(user_id, result=result)
        return AIMessage(content=result).content
        
    def stream(self,user_id,user_input):
        if not isinstance(user_input,list):#prompt,否则就是history
            history=[{"role":"user","content":user_input}]
        else:
            history=user_input
        result=""
        response = self.client.chat.completions.create(
            model="deepseek-r1:8b",
            messages=history,
            stream=True,


        )
        for chunk in response:
            if hasattr(chunk.choices[0].delta,"content"):#yield f"data: {i} yes your right\n\n"
                content_piece=chunk.choices[0].delta.content
                if content_piece:
                    result+=content_piece
                    yield (json.dumps({"response":content_piece,"done":False})+"\n")#.encode('utf-8')
        # print("stream_output:"+out)
        session_manager.update_history(user_id, result=result)
        yield json.dumps({"response":"","done":True})#.encode('utf-8')

2.llama.cpp

llama.cpp内外都是linux环境，打包系统环境包，python环境包和llama.cpp和模型文件。

2.1.安装系统环境

外网安装

sudo apt update
sudo apt install -y build-essential git cmake wget python3 python3-pip  libcurl4-openssl-dev
安装cuda驱动

2.2.安装python环境

conda create -n llama python==3.11.9
conda activate llama
pip install -U transformers
pip3 install torch torchvision --index-url https://download.pytorch.org/whl/cu129
pip install mistral_common
pip install sentencepiece

2.3.下载编译llama

git clone https://github.com/ggml-org/llama.cpp.git
cd llama.cpp
mkdir build
cd build
cmake ..
cmake --build . --config Release

2.4.运行测试

2.4.1.下载模型

git clone https://www.modelscope.cn/deepseek-ai/DeepSeek-R1-Distill-Qwen-7B.git

2.4.2.模型转换GGUF格式

使用llama.cpp把hf格式转换为 GGUF格式，也可以直接下载guff格式

python3 convert_hf_to_gguf.py ../DeepSeek-R1-Distill-Qwen-7B --outfile ../deepseek.gguf

2.4.3.运行推理

./build/bin/llama-cli -m ../deepseek.gguf -p "写一段Python代码，打印Hello World"      
./build/bin/llama-cli -m ../deepseek.gguf -i    #可交互模式
./build/bin/llama-cli -m ../deepseek.gguf -c 4096 -i     #带上下文窗口（例如 4096）带记忆交互模式
./build/bin/llama-server -m ../deepseek.gguf --port 8080 --host 0.0.0.0     #server 模式进行API 部署

2.4.4.测试调用

curl -X POST http://localhost:8080/v1/chat/completions -H "Content-Type: application/json" -d '{"model": "deepseek","messages": [{"role": "system", "content": "You are a helpful assistant."},{"role": "user", "content": "用C语言写一个快速排序"}],"max_tokens": 200}'

2.5.打包部署

打包系统所需包

mkdir ~/curl_pkg && cd ~/curl_pkg
apt-get download libcurl4-openssl-dev libcurl4 libnghttp2-14 librtmp1 libssh2-1 libidn2-0 libpsl5 libunistring2 zlib1g
tar -czvf curl_pkgs.tar.gz *.deb

在内网环境安装系统包

tar -xzvf curl_pkgs.tar.gz
sudo dpkg -i *.deb
sudo apt --fix-broken install

打包安装python环境win可以参考前面的ollama，麒麟的话可以参考第二篇。

2.6.模型封装

根据业务自己封装模型封装

import requests
import json
from functools import lru_cache
from sm import session_manager 
from langchain_core.messages.ai import AIMessage
class MYLLM:
    def __init__(self):
        self.base_url_un = 'http://192.168.1.10:8001/v1/chat/completions'
        self.base_url = 'http://192.168.1.10:8010/v1/chat/completions'
        self.base_model_un = 'un_deep_think_model'
        self.base_model = 'deep_think_model'
        self.topp=0.9
        self.temperature=0.1

    def invoke(self,user_id,user_input):
        if not isinstance(user_input,list):#prompt,否则就是history
            history=[{"role":"user","content":user_input}]
        else:
            history=user_input
        response = requests.post(self.base_url_un, json={
                    "temperature": self.temperature,
                    "model": self.base_model_un,
                    "topp": self.topp,
                    "messages": history
                })

        if response.status_code == 200:
            result = response.json()
            result = result.get("choices", [{}])[0].get("message", {}).get("content")
            session_manager.update_history(user_id, result=result)
            return AIMessage(content=result).content
    def stream(self, user_id, user_input):
        if not isinstance(user_input, list):  # 如果是单个prompt，转换为history格式
            history = [{"role": "user", "content": user_input}]
        else:
            history = user_input

        result = ""
        response = requests.post(
            self.base_url,
            json={
                "temperature": self.temperature,
                "model": self.base_model,
                "topp": self.topp,
                "messages": history,
                "stream": True,
            },
            stream=True  # 关键：启用流式传输
        )

        if response.status_code == 200:
            decoder = json.JSONDecoder()  # 用于解析JSON片段
            buffer = ""  # 用于累积不完整的JSON数据

            for line in response.iter_lines():  # 直接逐行处理流式响应
                if line:  # 跳过空行
                    line = line.decode('utf-8')
                    if line.startswith('data: '):  # 处理数据行
                        data = line[6:]  # 去掉"data: "前缀
                        if data == "[DONE]":  # 响应结束标记
                            break
                        buffer += data  # 累积可能的多行JSON
                        try:
                            # 尝试解析完整的JSON对象
                            while buffer:
                                obj, idx = decoder.raw_decode(buffer)
                                buffer = buffer[idx:].lstrip()  # 清理缓冲区
                                content_piece = obj.get("choices", [{}])[0].get("delta", {}).get("content")
                                if content_piece:
                                    result += content_piece
                                    yield json.dumps({"response": content_piece, "done": False}) + "\n"
                        except json.JSONDecodeError:
                            pass  # 数据不完整，继续累积
                    # else:  # 其他控制信息（如错误）可自行处理

            session_manager.update_history(user_id, result=result)
            yield json.dumps({"response": "", "done": True})
        else:
            # 处理错误响应（根据实际需求添加）
            yield json.dumps({"response": "请求失败", "done": True})
    def stream_unthink(self, user_id, user_input):
        if not isinstance(user_input, list):  # 如果是单个prompt，转换为history格式
            history = [{"role": "user", "content": user_input}]
        else:
            history = user_input

        result = ""
        response = requests.post(
            self.base_url_un,
            json={
                "temperature": self.temperature,
                "model": self.base_model_un,
                "topp": self.topp,
                "messages": history,
                "stream": True,
            },
            stream=True  # 关键：启用流式传输
        )

        if response.status_code == 200:
            decoder = json.JSONDecoder()  # 用于解析JSON片段
            buffer = ""  # 用于累积不完整的JSON数据

            for line in response.iter_lines():  # 直接逐行处理流式响应
                if line:  # 跳过空行
                    line = line.decode('utf-8')
                    if line.startswith('data: '):  # 处理数据行
                        data = line[6:]  # 去掉"data: "前缀
                        if data == "[DONE]":  # 响应结束标记
                            break
                        buffer += data  # 累积可能的多行JSON
                        try:
                            # 尝试解析完整的JSON对象
                            while buffer:
                                obj, idx = decoder.raw_decode(buffer)
                                buffer = buffer[idx:].lstrip()  # 清理缓冲区
                                content_piece = obj.get("choices", [{}])[0].get("delta", {}).get("content")
                                if content_piece:
                                    result += content_piece
                                    yield json.dumps({"response": content_piece, "done": False}) + "\n"
                        except json.JSONDecodeError:
                            pass  # 数据不完整，继续累积
                    # else:  # 其他控制信息（如错误）可自行处理

            session_manager.update_history(user_id, result=result)
            yield json.dumps({"response": "", "done": True})
        else:
            # 处理错误响应（根据实际需求添加）
            yield json.dumps({"response": "请求失败", "done": True})

# if __name__ == "__main__":
#     model_caller = MYLLM()

#     messages = [
#         {"role": "user", "content": "你好，我是用户。"}
#     ]

#     # response = model_caller.invoke("1",messages)
#     response2 = model_caller.stream("2",messages)
    # print(response)
    # print(response2)

参考博文
[1].https://www.zouht.com/3835.html
[2].https://blog.csdn.net/ATTK_Time/article/details/146184307
[3].https://zhuanlan.zhihu.com/p/682585600
[4].https://zhuanlan.zhihu.com/p/1895533879319836316