

本教程为 DeepSeek R1 本地部署教程  
使用 Ollama 进行模型部署，Open Webui 是实现可视化页面聊天

# 部署教程
## 安装 Docker
前往 [Docker 官网](https://www.docker.com/) 下载并安装适合你操作系统的 Docker 版本并安装
![](images/dockerhome.png)

## 安装 Ollama
1. 安装 Ollama  
前往官网 [Ollama](https://ollama.com/) 选择适合你的版本并下载 支持 Mac、Windows、Linux
![](images/ollamahome.png)

2. 前往 Ollama 模型页面
访问 DeepSeek R1 模型页面 [DeepSeek R1](https://ollama.com/library/deepseek-r1)  
根据自己的电脑配置选择合适的模型并使用命令行运行
![](images/aa0ea1beccba22c41d34b10f4678f387.png)

3. 运行 Ollama  
我这边选择的是 DeepSeek R1 1.5b 版本，终端运行命令如下  
```shell
ollama run deepseek-r1:1.5b
```
运行后会自动开始下载模型  
![](images/d753199b36418e56e105fd17d625161a.png)

4. 运行完成  
![](images/image.png)

## 安装 Open Webui
1. 安装 Open Webui
如果本机安装了 Ollama，那么可以使用以下命令安装  

```shell
docker run -d -p 3000:8080 --add-host=host.docker.internal:host-gateway -v open-webui:/app/backend/data --name open-webui --restart always ghcr.io/open-webui/open-webui:main
```

如果想要细节探索可以前往 github [Open Webui](https://github.com/open-webui/open-webui) 查看  

运行之后可以在浏览器中输入 `http://localhost:3000` 访问 Open Webui 页面
第一次运行可能会需要拉取镜像，等待一段时间成功启动之后即可看到页面
![](images/54efa7335d9b2b326a624918c0d27012.png)

2. 访问 Open Webui
观察 Docker 容器运行情况，可以看到 Open Webui 已经运行成功
![](images/e354aab22c0c9d010f6d770975e90cd1.png)
访问 `http://localhost:3000` 即可看到 Open Webui 页面
他会自动连接到 Ollama 服务，可以直接使用 DeepSeek R1 模型进行聊天
![](images/201f0b881d0fb707feef1c491ae8192a.png)

# 总结
到此为止，我们已经完成了 DeepSeek R1 的本地部罋，可以在本地环境中使用 DeepSeek R1 进行聊天了

---
**如有问题可以联系我的邮箱交流和学习**
```
jzh1264204425@163.com
```




本地部署 DeepSeek R1 教程


# 原始Dockerfile

在使用 Node.js 20的官方基础镜像打包出来的镜像体积有1.2G

```
# 使用 Node.js 20的官方基础镜像
FROM node:20.11.1

# 设置工作目录
WORKDIR /app

# 将 package.json 和 yarn.lock 复制到容器中
COPY package.json yarn.lock ./

# 设置npm镜像
RUN npm config set registry https://registry.npmmirror.com

# 安装yarn和项目依赖
# RUN npm install -g yarn

# 设置Yarn使用淘宝镜像
RUN yarn config set registry 'https://registry.npmmirror.com'

# 安装项目依赖
RUN yarn install

# 将项目文件复制到容器中
COPY . .

# 构建项目
RUN yarn build

# 容器启动时运行
CMD ["yarn", "start"]
```

在使用 Node.js 20的官方基础镜像打包出来的镜像体积有1.2G

# 新版Dockerfile

1. 先使用node:20.11.1-alpine构建应用

3. 在使用node:20.11.1-alpine复制build后的文件

5. 最后打包的体积为200M

```
# 第一阶段：构建应用
FROM node:20.11.1-alpine AS build_image

WORKDIR /app
COPY package.json yarn.lock ./
RUN yarn install
COPY . .
RUN yarn build

# 第二阶段：运行应用
FROM node:20.11.1-alpine

WORKDIR /app
COPY --from=build_image /app/.next ./.next
COPY --from=build_image /app/node_modules ./node_modules
COPY --from=build_image /app/package.json ./package.json
COPY --from=build_image /app/public ./public

# Next.js官方推荐使用自己的服务器进行生产环境部署
EXPOSE 3000
CMD ["node_modules/.bin/next", "start", "-p", "3000"]
```

这样就实现了打包体积从 1.2G -> 200MB


Docker Node 环境打包 减少打包体积


# 问题出现

今天在写了个中间件，判断权限问题之后接口一直返回:

```
{
    "code": "invalid_param",
    "message": "Invalid JSON data: EOF"
}
```

# 问题排查

造成原因是 MiddleWare中我读取了Request Body中的参数来进行鉴权但是

> Go语言的http.Request.Body是一个io.ReadCloser接口类型，它允许你从HTTP请求体中读取数据。Body的设计和使用方式在某些场景下会有特定的设计考虑。理解这个设计背后的原因可以帮助你更好地编写健壮的中间件和处理函数。
> 
> - 一次性读取的设计 http.Request.Body本质上是一个流（stream），它允许从客户端发送的数据中顺序读取字节。由于流的性质，Body的内容只能被读取一次。当你从Body中读取数据后，流的光标已经移动到末尾，再次读取将返回EOF（即流结束）。这是因为Body在设计上就是为了减少内存占用，并且符合流式处理的惯用模式（流式处理就是一次读取一点数据，而不是把所有数据都放到内存中）。

\# 解决方案

1. 提前存入变量

```
 bodyBytes, err := io.ReadAll(c.Request.Body)
```

2. 在使用完之后

```
 // reset context reader to the beginning
c.Request.Body = io.NopCloser(bytes.NewReader(bodyBytes))
```


【Go语言】Gin框架下，body参数只能读取一次解决方案


苹果公司在其2024年全球开发者大会（WWDC24）上介绍的Apple Intelligence系统的一部分，特别是关于其基础模型的。Apple Intelligence是一个集成在iOS 18、iPadOS 18和macOS Sequoia中的个人智能系统。这个系统由多个高度能生成模型组成，这些模型专门针对用户的日常任务，并且能够实时适应当前的活动。

这些模型包括一个约30亿参数的设备上语言模型和一个更大的基于服务器的语言模型，后者可在Private Cloud Compute上使用，运行在苹果硅服务器上。这些基础模型已经被微调，以支持用户体验，如文本写作和精炼、通知的优先排序和总结、为与家人和朋友的对话创建有趣的图像，以及简化应用程序之间的交互的应用程序内操作。

苹果公司强调了其负责任的AI开发重点，包括保护用户隐私、代表用户、谨慎设计和保护隐私等原则。这些原则贯穿于支持Apple Intelligence的架构中，连接了特性和工具与专门的模型，并扫描输入和输出，以确保每个特性能够负责任地运行。

在训练这些基础模型时，苹果使用了其AXLearn框架，这是一个在2023年发布的开源项目，它建立在JAX和XLA之上，允许在各种训练硬件和云平台上高效、可扩展地训练模型。苹果还采用了数据并行、张量并行、序列并行和完全分片数据并行（FSDP）等技术来扩展训练。

此外，苹果公司还介绍了其在模型优化方面的创新技术，包括分组查询注意力、共享输入和输出词汇嵌入表、低位量化、LoRA适配器框架、Talaria工具等，以提高设备和服务器模型的速度和效率。

在模型适应方面，苹果使用适配器——小型神经网络模块，可以插入预训练模型的不同层中，以针对特定任务微调模型。通过仅微调适配器层，原始预训练模型的参数保持不变，同时适配器层被定制以支持特定任务。

在性能和评估方面，苹果公司专注于提供能够使用户在苹果产品上进行通信、工作、表达自己和完成任务的生成模型。他们通过人类评估来基准测试模型，发现这些结果与产品中的用户体验高度相关。苹果还使用了IFEval基准测试来比较模型遵循指令的能力，并在内部摘要和写作基准测试中评估了模型的写作能力。

最后，文章提到了苹果公司对于安全性的持续评估，包括手动和自动的红队测试，以及与内部和外部团队的合作，以继续评估模型的安全性。


AI总结：Apple’s On-Device and Server Foundation Models


```
import torch
import torch.nn as nn

# 定义Transformer模型
class Transformer(nn.Module):
    def __init__(self, input_dim, output_dim, hidden_dim, num_layers):
        super(Transformer, self).__init__()

        # 编码器和解码器的初始化
        self.encoder = nn.TransformerEncoderLayer(input_dim, hidden_dim, num_layers)
        self.decoder = nn.TransformerDecoderLayer(output_dim, hidden_dim, num_layers)

    def forward(self, src, tgt):
        # 编码器的前向传播
        enc_output = self.encoder(src)

        # 解码器的前向传播
        dec_output = self.decoder(tgt, enc_output)

        return dec_output

# 创建Transformer模型实例
input_dim = 100
output_dim = 200
hidden_dim = 256
num_layers = 4
model = Transformer(input_dim, output_dim, hidden_dim, num_layers)

# 定义输入和目标数据
src = torch.randn(50, input_dim)
tgt = torch.randn(60, output_dim)

# 进行前向传播
output = model(src, tgt)
```


简单案例Transformer-翻译任务简单案例Transformer-


使用了Https，大大提升了大量网络io请求时的速度

```
def get_majors_school_list():
    with open("./data/majors_list.json", "r") as f:
        majors_list = json.load(f)
    for major in majors_list:
        major_code = major.get("major_code")
        major_name = major.get("major_name")
        way = major.get("way")
        url = f""
        headers = {
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.77 Safari/537.36"
        }
        response = requests.get(url, headers=headers)
        result = response.json().get("data").get("results")
        major["school_list"] = result
        # 按专业名称保存
        with open(f"./data/majors_school_list/{major_code}_{major_name}_{way}.json", "w") as f:
            f.write(json.dumps(majors_list, ensure_ascii=False))
```

```
import json
import aiohttp
import asyncio

async def fetch_major_data(session, major):
    major_code = major.get("major_code")
    major_name = major.get("major_name")
    way = major.get("way")
    url = f""
    headers = {
        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.77 Safari/537.36"
    }
    async with session.get(url, headers=headers) as response:
        result = await response.json()
        major["school_list"] = result.get("data").get("results")
        # 按专业名称保存
        with open(f"./data/majors_school_list/{major_code}_{major_name}_{way}.json", "w") as f:
            f.write(json.dumps(major, ensure_ascii=False))

async def get_majors_school_list():
    with open("./data/majors_list.json", "r") as f:
        majors_list = json.load(f)
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_major_data(session, major) for major in majors_list]
        await asyncio.gather(*tasks)

# asyncio.run 是 Python 3.7+ 中的新方式来运行异步程序
asyncio.run(get_majors_school_list())
```

### 导入模块

pythonCopy code

`import json import aiohttp import asyncio`

- `json`：用于处理 JSON 数据。

- `aiohttp`：用于异步 HTTP 网络请求。

- `asyncio`：Python 的异步I/O框架，用于编写并发代码。

### 异步函数定义

pythonCopy code

`async def fetch_major_data(session, major):`

- `async def`：定义一个异步函数。这允许函数内部使用 `await` 表达式，并使函数执行非阻塞。

### 构建请求 URL 和头部

pythonCopy code

`major_code = major.get("major_code") major_name = major.get("major_name") way = major.get("way") url = f"" headers = { "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.77 Safari/537.36" }`

- 提取专业信息，构建 API 请求的 URL 和 headers。

### 发起异步网络请求

pythonCopy code

`async with session.get(url, headers=headers) as response:`

- `async with`：上下文管理器，用于异步操作。这里用于执行异步的 HTTP GET 请求。

- `session.get`：`aiohttp.ClientSession` 对象的 `get` 方法，用于发起 GET 请求。它是异步的，返回一个 response 对象。

### 处理响应数据

pythonCopy code

`result = await response.json() major["school_list"] = result.get("data").get("results")`

- `await`：等待异步操作完成。这里等待的是 `response.json()`，它异步地将响应内容转换为 JSON 对象。

- 解析 JSON 数据，并将解析后的数据保存到 `major` 字典中。

### 保存数据到文件

pythonCopy code

`with open(f"./data/majors_school_list/{major_code}_{major_name}_{way}.json", "w") as f: f.write(json.dumps(major, ensure_ascii=False))`

- 将包含学校列表的 `major` 字典保存为 JSON 文件。

### 驱动整个异步流程

pythonCopy code

`async def get_majors_school_list(): with open("./data/majors_list.json", "r") as f: majors_list = json.load(f) async with aiohttp.ClientSession() as session: tasks = [fetch_major_data(session, major) for major in majors_list] await asyncio.gather(*tasks)`

- 定义另一个异步函数来读取专业列表，然后为列表中的每个专业创建一个异步任务。

- `aiohttp.ClientSession()`：创建一个 HTTP 会话，用于发起请求。

- `asyncio.gather`：并发运行所有的异步任务。

### 执行异步主函数

pythonCopy code

`asyncio.run(get_majors_school_list())`

- `asyncio.run`：运行最高级别的入口点 "main" 函数，这里是 `get_majors_school_list()`。这是 Python 3.7+ 中推荐的方式来运行异步程序。

这段代码展示了如何使用 Python 的异步编程来高效地处理网络I/O密集型任务。通过并发地发起网络请求，程序无需在每个请求上顺序等待，从而大大提高了效率。


HTTPX异步，提速150倍


在Diffusers的0.17.0版本之后支持直接调用来自CivitAI的Lora

# 1\. 创建Pipeline

```
import differs 
import torch 

# 创建PipeLine
safetensors_path = "model.safetensors"

 pipeline = differs.StableDiffusionPipeline.from_single_file( 
    safetensors_path, 
    torch_dtype = torch.float16, # 对于 CUDA。
 )
```

# 2\. 将LoRA下载到本地

```
lora_safetensors_path = "test.safetensors"

 pipeline.load_lora_weights(lora_safetensors_path)
```

# 3.设置权重

使用 pipeline.fuse\_lora(lora\_scale = 0.7)  
若要导入多个LoRA设置多个权重则可以

```
lora_dirs = [ "lora1.safetensors" , "lora2.safetensors" , ...] 
lora_scales = [ 0.5 , 0.75 , ...] 

ldir, lsc in  zip (lora_dirs, lora_scales): 
    # 迭代添加新的 LoRA。
    pipeline.load_lora_weights(ldir) 
    # 并相应地缩放它们。
    pipeline.fuse_lora(lora_scale = lsc)
```

# 4.卸载LoRA

```
1.unload_lora_weights()
2.disable_lora()
但是实测下来发现会有问题，建议是卸载LoRA之后新建一个pipeline
```


在Diffusers中调用CivitAI Lora


## aiohttp

● 成熟度：aiohttp是较早提供异步HTTP客户端和服务器功能的Python库之一。它完全基于asyncio，在社区中广泛使用，有丰富的文档和社区支持。  
● 客户端会话：aiohttp强调使用ClientSession作为发起请求的主要方式，这是因为ClientSession支持连接复用、cookies存储等。在aiohttp中，创建一个ClientSession实例，然后使用这个会话发起GET、POST等请求是常见的模式。

```
import aiohttp
import asyncio

async def fetch_with_aiohttp(endpoint, headers, data):
    async with aiohttp.ClientSession() as session:
        async with session.post(endpoint, headers=headers, json=data) as response:
            print(response.status)
            print(await response.json())
```

## httpx

● 现代和简洁：httpx是一个相对较新的库，它的API设计得更加现代和简洁。httpx旨在提供requests库的异步版本，同时保持与requests相似的API设计，使得对于熟悉requests的用户更加容易上手。  
● 同步和异步：httpx同时支持同步和异步请求，这意味着你可以在同一个库中使用相同的API进行同步和异步HTTP操作。

```
import httpx
import asyncio

async def fetch_with_httpx(url, headers, data):
    async with httpx.AsyncClient() as client:
        response = await client.post(url, headers=headers, json=data)
        print(response.status_code)
        print(response.json())
```


Python的两种异步HTTP请求


演示代码为调用tools的功能  
这边只做演示使用

# Fastapi接口构造

```
# fastapi接口
@app.post("/v1/chat/completions", status_code=status.HTTP_200_OK)
async def create_chat_completion(request: ChatCompletionRequest):
    """
    生成聊天
    :param request:
    :return:
    """
    query = request.messages[-1].content
    print(query)
    prev_messages = request.messages[:-1]
    if len(prev_messages) and prev_messages[0].role == Role.SYSTEM:
        system = prev_messages.pop(0).content
    else:
        system = None

    system = system.replace('\n', '')
    messages = []
    messages.append(
        {
            "role": "system",
            "content": system
        },
    )
    messages.append(
        {
            "role": "user",
            "content": query
        }
    )
    return EventSourceResponse(tools_select(messages))
```

# 功能实现

## 调用的tools\_select

该函数的功能主要是开一个新的线程来fetch\_data()，用于获取返回的信息，因为zhipu的sdk是同步操作，所以我们要同步执行，异步获取返回信息

```
async def tools_select(messages):
    """
    数据获取（同步操作）在一个单独的线程中进行，不会阻塞主事件循环。
    数据的发送（异步操作）通过异步生成器逐个进行，每当有数据可用时就立即处理和发送。
    :param messages:
    :return:
    """
    # Queue 提供了阻塞（如 get()、put()）和非阻塞（如 get_nowait()、put_nowait()）的方法来处理队列元素。
    # 阻塞操作会等待直到队列中有可用的数据或者有空间来放置新的数据。
    q = queue.Queue()  # queue.Queue() 是 Python 中的一个标准库，用于在多个线程之间安全地交换信息或数据

    # 创建并启动数据获取线程
    thread = threading.Thread(target=fetch_data, args=(q, messages))
    thread.start()

    # 异步处理队列中的数据
    while True:
        content = await asyncio.to_thread(q.get) # 从队列中获取数据
        if content is None:  # 使用 None 作为结束信号
            break

        content = content.replace('\n', '').strip()
        content = re.sub(r'\s+', ' ', content)
        yield content

    thread.join()  # 确保线程完成
```

## fetch\_data()函数的实现

使用Queue来发送消息，由于需要判断是否调用工具这边又封装了一个函数process\_async\_generator()

```
def fetch_data(q, messages):
    response = client.chat.completions.create(
        model="GLM-3-Turbo",
        messages=messages,
        tools=tools,
        stream=True,
    )
    for chunk in response:
        print(chunk.choices[0].delta)
        print(chunk.choices[0].delta.tool_calls)
        if chunk.choices[0].delta.tool_calls:
            messages.append(chunk.choices[0].delta.model_dump())
            # parse_function_call(chunk, messages)
            asyncio.run(process_async_generator(q, chunk, messages))
        else:
            q.put(chunk.choices[0].delta.content)
    q.put(None)  # 发送结束信号
```

## process\_async\_generator()函数的实现

该函数主要是用来处理parse\_function\_call工具调用的功能，同样通过异步操作实现了流式的返回

```
async def parse_function_call(model_response: ChatCompletionChunk, messages):
    # 处理函数调用结果，根据模型返回参数，调用对应的函数。
    # 调用函数返回结果后构造tool message，再次调用模型，将函数结果输入模型
    # 模型会将函数调用结果以自然语言格式返回给用户。
    if model_response.choices[0].delta.tool_calls:
        tool_call = model_response.choices[0].delta.tool_calls[0]
        args = tool_call.function.arguments
        print("args", args)
        # str转dict
        # args = json.loads(args)
        function_result = {}
        if tool_call.function.name == "search_score":
            function_result = search_score(**json.loads(args))
        messages.append({
            "role": "system",
            "content": f"请把查询出来的所有内容以Markdown的格式都返回给用户,包括学校的联系方式地址等,如果没有返回内容就正常回答",
        })
        messages.append({
            "role": "tool",
            "content": f"{json.dumps(function_result)}",
            "tool_call_id": tool_call.id
        })
        response = client.chat.completions.create(
            model="GLM-3-Turbo",
            # model='glm-4',
            messages=messages,
            tools=tools,
            stream=True,
            temperature=0.8
        )
        for chunk in response:
            content = chunk.choices[0].delta.content
            yield content


async def process_async_generator(q, chunk, messages):
    async for result in parse_function_call(chunk, messages):
        q.put(result)

```

![](images/image-1024x494.png)


智谱zhipuai python SDK，使用实现fastapi实现流式传输


## 发现问题：

昨晚突然发现公司电脑远程桌面连不上，准备第二天排查一下，结果第二天电脑重启之后远程桌面仍然连不上，上本机查看发现系统引导损坏，修复完毕后进入系统，发现文件后缀名全被篡改，一开始以为是硬盘出现问题，导致文件损坏，具体情况类似于下图。

![](images/image-7-1024x355.png)

并且文件夹中会带有一封信，类似于解锁需要支付加密货币。

![](images/image-8-1024x517.png)

## 排查

初步分析原因可能有以下三点：

1.使用的是Windows10系统比较老的版本，可能存在漏洞被攻击

2.主机上是存在https服务的，有对外开放的端口所以存在被攻击的可能

3.公网IP暴露，导致被攻击

## 解决方案

1.其实如果不想花钱的话，只有重装

2.做好数据备份，这台机子上的服务，代码都有备份所以恢复起来难度并不大

3.以此为戒强调备份的重要性，如果包含企业重要数据那将会得不偿失


记录一次Win10被勒索病毒攻击


```
import random
import time

from moviepy.audio.AudioClip import CompositeAudioClip
from moviepy.config import change_settings
from moviepy.editor import VideoFileClip, AudioFileClip, clips_array
from moviepy.video.VideoClip import TextClip, ColorClip
from moviepy.video.compositing.CompositeVideoClip import CompositeVideoClip

# change_settings({"IMAGEMAGICK_BINARY": r"D:\Program Files\ImageMagick-7.1.1-Q16-HDRI\magick.exe"})

# 加载网络视频
video_url = ""
video_clip = VideoFileClip(video_url, audio=True)
start = random.randint(0, int(video_clip.duration - 15))
cropped_video = video_clip.subclip(start, start + 15)
cropped_video.write_videofile('qiniu_video_test.mp4', codec="libx264", audio_codec="aac", threads=6)

# 加载视频和背景音乐
video_path = r'./video_test.mp4'  # 输入视频文件路径
bgm_path = r'./xuanni.m4a'  # 背景音乐文件路径

video_clip = VideoFileClip(video_path, audio=True)
bgm_clip = AudioFileClip(bgm_path)

# # video_clip裁剪一半
# video_clip = video_clip.subclip(0, video_clip.duration / 2)
# video_clip.write_videofile('video_test_half.mp4', codec="libx264", audio_codec="aac", threads=6)
# exit()

# random随机裁剪15s片段
start = random.randint(0, int(video_clip.duration - 15))
cropped_video = video_clip.subclip(start, start + 15)
# cropped_video = video_clip.subclip(10, 30)

# 裁剪背景音乐，使其与视频长度匹配，并调整音量
# 同样使用random随机裁剪
bgm_clip = bgm_clip.subclip(random.randint(0, int(bgm_clip.duration - 15)), random.randint(15, int(bgm_clip.duration)))
bgm_clip = bgm_clip.volumex(2)  # 调整音量

# 将背景音乐与原视频的音频混合在一起
video_with_bgm = cropped_video.set_audio(CompositeAudioClip([cropped_video.audio, bgm_clip]))

# 设置混合后音频的持续时间，以确保与视频持续时间相匹配
video_with_bgm = video_with_bgm.set_duration(cropped_video.duration)

# 调整视频大小为竖屏
target_width = 1080
target_height = 1920

# 创建一个背景视频，大小为1080x1920，创建黑色背景
background_color = (0, 0, 0)
background_clip = ColorClip(size=(target_width, target_height), color=background_color)

# 将视频放置在背景视频中间
video_with_bgm = video_with_bgm.set_position('center')

# 将视频和背景视频合并
video_with_background = CompositeVideoClip([background_clip, video_with_bgm])
video_with_background.duration = video_with_bgm.duration

# 创建字幕
text = "这是我的字幕"  # 要添加的字幕文本
fontsize = 100  # 字幕字体大小
font_color = 'white'  # 字幕字体颜色
# bg_color = 'black'  # 字幕背景颜色（可选）
duration = video_with_background.duration  # 字幕持续时间，与视频长度相同

# 创建 TextClip 对象来生成字幕
subtitle = TextClip(text, fontsize=fontsize, color=font_color, size=video_clip.size, font=r"../font/OPPOSans-R.ttf")
subtitle = subtitle.set_duration(duration)
# 计算视频长宽
w, h = video_clip.size
# 设置字幕的位置 为视频中间偏上
# subtitle = subtitle.set_position(((w - subtitle.w) / 2, (h - subtitle.h) / 4))
subtitle = subtitle.set_position(("center", "top"))  # 设置字幕的位置
# subtitle = subtitle.set_position((45, 150))  # 设置字幕的位置

# 将字幕添加到视频
video_with_subtitle = CompositeVideoClip([video_with_background, subtitle])
video_with_subtitle.duration = video_with_background.duration

# 导出最终视频
output_path = 'output_video.mp4'  # 输出视频文件路径
video_with_subtitle.write_videofile(output_path, codec="libx264", audio_codec="aac", threads=6)
```


MoviePy常用操作


本文提供Python中使用moviepy的ImageMagick问题的解决教程

# 报错问题

问题为win中运行moviepy会报错ImageMagick找不到的问题，即使配置了环境变量，也仍然会报错。  
下图是已经给了权限的：

![](images/image-4-1024x434.png)

  
然而系统其实已经安装完成了ImageMagick

![](images/image-5-1024x106.png)

# 解决方案

1.下载安装[ImageMagick](https://imagemagick.org/script/download.php#windows)无脑下一步就行，安装过程中最后一步会自动添加至环境变量  
2.在代码中添加一行

```
from moviepy.config import change_settings


change_settings({"IMAGEMAGICK_BINARY": r"D:\Program Files\ImageMagick-7.1.1-Q16-HDRI\magick.exe"})
```

这样就大功告成了

![](images/image-6-1024x247.png)


windows环境 Python中使用moviepy的ImageMagick配置


# LLaMA-Factory介绍

LLaMA-Factory是一个基于Gradio的可视化界面，可以导入模型训练以及训练模型，对新手十分友好，且可以可视化训练过程。

# 项目部署

首先git clone LLaMA-Factory仓库

```bash
git clone https://github.com/hiyouga/LLaMA-Factory.git
```

使用conda创建所需要环境

```
conda create -n llamaFactory python=3.10
```

```
# 安装所需依赖
pip install -r requirements.txt
```

# 项目启动

由于我使用的wsl2可能和部分读者不一样，推荐使用linux ubuntu

```
# 启动LLaMA board
python src/train_web.py
```

![](images/image-1-1024x618.png)

以上就是启动页面，注意红框部分

- 选择模型名称
- 选择完模型之后模型路径会默认为Hugging Face模型标识符，我推荐下载到本地

当选择完之后我们就可以直接进行模型的加载

![](images/image-2-1024x599.png)

点击chat，选择加载模型，等待模型加载完毕，即可弹出对话框

# 准备数据集

数据集需要特定的格式  
在项目的data文件夹下操作  
项目默认的example中存在两个比较重要的文件分别为：dataset\_info.json以及examples.json

```
dataset_info.json
{
  "你的数据集名字": {
    "file_name": "文件名.json",
    "file_sha1": "文件sha1607f94a7f581341e59685aef32f531095232cf23"
  },
  "alpaca_zh": {
    "file_name": "alpaca_data_zh_51k.json",
    "file_sha1": "e655af3db557a4197f7b0cf92e1986b08fae6311"
  }
}
```

```
examples.json
其中instruction，output是必须的，其他的可选
[
  {
    "instruction": "听起来很不错。人工智能可能在哪些方面面临挑战呢？",
    "input": "",
    "output": "人工智能面临的挑战包括数据隐私、安全和道德方面的问题，以及影响就业机会的自动化等问题。",
    "history": [
      ["你好，你能帮我解答一个问题吗？", "当然，请问有什么问题？"],
      ["我想了解人工智能的未来发展方向，你有什么想法吗？", "人工智能在未来的发展方向可能包括更强大的机器学习算法，更先进的自然语言处理技术，以及更加智能的机器人。"]
    ]
  },
  {
    "instruction": "好的，谢谢你！",
    "input": "",
    "output": "不客气，有其他需要帮忙的地方可以继续问我。",
    "history": [
      ["你好，能告诉我今天天气怎么样吗？", "当然可以，请问您所在的城市是哪里？"],
      ["我在纽约。", "纽约今天晴间多云，气温最高约26摄氏度，最低约18摄氏度，记得注意保暖喔。"]
    ]
  }
]
```

```
获取文件的sha1
sha1sum example.json
```

## 开始训练

当我们准备好数据集之后

![](images/image-3-1024x585.png)

选择我们的数据集，点击开始训练，便可以可视化看到我们训练的过程了。


使用LLaMA-Factory训练ChatGLM3B


需要OKLink的APIKEY自行申请

将脚本放置于挖Quark的目录

```
import time
import os

import requests

wallet_address = '生成的钱包地址'
ok_access_key = 'OKLink的APIKEY'

proxy = {
    'http': 'http://192.168.0.69:7890',
    'https': 'http://192.168.0.69:7890'
}


def getTransactionList(address):
    url = f"https://www.oklink.com/api/v5/explorer/address/transaction-list?chainShortName=btc&address={address}"
    headers = {
        'Ok-Access-Key': ok_access_key
    }
    response = requests.request("GET", url, headers=headers, proxies=proxy)
    # 提取订单列表
    transactionList = response.json()['data'][0]['transactionLists']
    # 获取订单id
    transactionIdList = []
    for transaction in transactionList:
        transactionIdList.append(transaction['txId'])
    return transactionIdList


def fetch_gas():
    url = 'https://www.oklink.com/api/v5/explorer/blockchain/fee?chainShortName=btc'
    headers = {
        'Ok-Access-Key': ok_access_key
    }
    response = requests.request("GET", url, headers=headers, proxies=proxy)
    # 提取gas
    bestTransactionFeeSat = response.json()['data'][0]['bestTransactionFeeSat']
    # 计算聪/字节
    satsbyte = int(bestTransactionFeeSat) / 1000

    return satsbyte + 5


if __name__ == '__main__':
    old_order = getTransactionList(wallet_address)
    transactionIdList = getTransactionList(wallet_address)
    # 每十分钟检查一次有没有新订单
    while True:
        transactionIdList = getTransactionList(wallet_address)
        # 比较是否有新订单
        if transactionIdList != old_order:
            print('有新订单')
            low_gas = fetch_gas()
            # 输入cmd命令
            os.system(f'yarn cli mint-dft quark --satsbyte {low_gas}')

        else:
            print('没有新订单')
        time.sleep(600)
```


Quark自动化检测脚本


```
import os

output_path = r'./data_erji_utf8'

def gbk_to_utf8(path):
    """
    将GBK编码的文件转换为UTF-8编码
    :param path: 文件路径
    :return: None
    """
    # 获取文件列表
    file_list = os.listdir(path)
    for file_name in file_list:
        # 拼接文件路径
        file_path = os.path.join(path, file_name)
        print(file_path)
        # 判断是否为文件
        if os.path.isfile(file_path):
            # 打开文件
            with open(file_path, 'rb') as f:
                # 读取文件内容
                gbk_content = f.read()
            utf8_content = gbk_content.decode('gbk').encode('utf-8')
            # 保存文件
            with open(os.path.join(output_path, file_name), 'wb') as f:
                f.write(utf8_content)


if __name__ == '__main__':
    gbk_to_utf8('./data_erji')
```


实用脚本-编码转换


### 文档结构

- **\\documentclass{}**: 定义文档类型。如 **\\documentclass\[12pt\]{article}** 表示创建一个12点字体的文章。

- **\\usepackage{}**: 导入宏包。如 **\\usepackage{graphicx}** 导入用于处理图像的宏包。

- **\\begin{document}** 和 **\\end{document}**: 文档内容的开始和结束。

- **\\title{}**, **\\author{}**, **\\date{}**: 定义标题、作者和日期。

- **\\maketitle**: 在文档中生成标题。

- **\\tableofcontents**: 自动生成目录。

### 章节和段落

- **\\section{}**, **\\subsection{}**, **\\subsubsection{}**: 创建章节、子章节和子子章节。

- **\\paragraph{}**, **\\subparagraph{}**: 创建段落和子段落。

- **\\newline** 或 **\\\\**: 新的一行。

- **\\newpage**: 新的一页。

### 文本格式化

- **\\textbf{}**: 加粗文本。

- **\\textit{}**: 斜体文本。

- **\\underline{}**: 下划线文本。

- **\\footnote{}**: 创建脚注。

### 列表

- **\\begin{itemize}** 和 **\\end{itemize}**: 无序列表。

- **\\begin{enumerate}** 和 **\\end{enumerate}**: 有序列表。

- **\\item**: 列表项。

### 图表

- **\\begin{figure}** 和 **\\end{figure}**: 图片环境。

- **\\caption{}**: 图片标题。

- **\\label{}**: 为图片、表格等添加标签，用于交叉引用。

- **\\begin{table}** 和 **\\end{table}**: 表格环境。

- **\\begin{tabular}** 和 **\\end{tabular}**: 定义表格格式。

### 数学公式

- **$...$**: 行内数学模式。

- **\\\[...\\\]** 或 **\\begin{equation}** 和 **\\end{equation}**: 单独的数学公式。

- **\\frac{}{}**: 分数。

- **\\sum**, **\\int**: 求和和积分。

### 参考文献

- **\\bibliographystyle{}**: 定义参考文献样式。

- **\\bibliography{}**: 引用 **.bib** 文件中的参考文献。

### 其他

- **\\begin{verbatim}** 和 **\\end{verbatim}**: 原样输出文本，忽略 LaTeX 命令。

- **\\url{}**: 插入网址。

- **%**: 注释，不会出现在最终文档中。


LaTeX 语法


```
import re
import os
import requests
import tqdm

header = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/66.0.3359.139 Safari/537.36'}


def getImg(url, idx, path, search):
    img = requests.get(url, headers=header)
    file = open(path + search + str(idx) + '.jpg', 'wb')
    file.write(img.content)
    file.close()


search = input("请输入搜索内容：")
number = int(input("请输入需求数量："))
path = r'dataset/train/others/'
if not os.path.exists(path):
    os.makedirs(path)

bar = tqdm.tqdm(total=number)
page = 0
while (True):
    if number == 0:
        break
    url = 'https://image.baidu.com/search/acjson'
    params = {
        "tn": "resultjson_com",
        "logid": "11555092689241190059",
        "ipn": "rj",
        "ct": "201326592",
        "is": "",
        "fp": "result",
        "queryWord": search,
        "cl": "2",
        "lm": "-1",
        "ie": "utf-8",
        "oe": "utf-8",
        "adpicid": "",
        "st": "-1",
        "z": "",
        "ic": "0",
        "hd": "",
        "latest": "",
        "copyright": "",
        "word": search,
        "s": "",
        "se": "",
        "tab": "",
        "width": "",
        "height": "",
        "face": "0",
        "istype": "2",
        "qc": "",
        "nc": "1",
        "fr": "",
        "expermode": "",
        "force": "",
        "pn": str(60 * page),
        "rn": number,
        "gsm": "1e",
        "1617626956685": ""
    }
    try:
        result = requests.get(url, headers=header, params=params).json()
    except Exception as e:
        print(e)
        continue
    url_list = []
    for data in result['data'][:-1]:
        url_list.append(data['thumbURL'])
    for i in range(len(url_list)):
        getImg(url_list[i], 60 * page + i, path, search)
        bar.update(1)
        number -= 1
        if number == 0:
            break
    page += 1
print("\nfinish!")
```


实用脚本-百度图片爬取


# 前言

购买了一个N100的小主机，准备打算来做软路由的，可是如果软路由的话就受限于openvrt，想做All IN ONE的话又怕很多东西不适配，最终选择Windows Server 2022的系统 + Hyper-V 虚拟化，但是在安装的过程中遇到了许多坑，特地分享一下。

# 系统安装

采用的镜像是NEXT TELL YOU的，这里就不多说了，大家可以自行前往下载，[NEXT TELL YOU](https://next.itellyou.cn/)。

## 网卡驱动下载

首先建议大家使用店家给的驱动，如果没有的话可以去Intel官网下载选择对应型号

## 注意事项

Windows Server 2022 的【无线 LAN 服务】默认是关闭的。开启【无线 LAN 服务】步骤如下：

1. 打开【服务器管理器】，在【仪表板】中点击【添加角色和功能】

3. 在弹出窗口中跳过【开始之前】，在【安装类型】中选择【基于角色或基于功能的安装】，在【服务器选择】中选择当前服务器

5. 跳过【服务器角色】，在【功能】中选择【无线LAN服务】，点击【下一步】

![](images/image-1-1024x734.png)

4.在【确认】中，选择【安装】，安装好后重启就可以看到驱动安装成功了

![](images/image-2-1024x722.png)

5.如果网卡还是禁用状态，win+r打开【服务】，将【WLAN AutoConfig】运行起来

![](images/image-3-1024x740.png)

## 提示：server（服务器） 系统安装（无线、声卡等）驱动程序时提示“这个Inf中的服务安装段落无效”解决方式

windows服务器版 windows server 时安装不上显卡及wlan等驱动时，其实是windows server为了提高稳定性，默认不安装或关闭了相关组件服务所致，添加图片上的两个功能。

![](images/image-4-1024x589.png)

![](images/image-5-1024x538.png)


N100 WindowsServer2022 固件安装 网卡驱动


```
import os
from PIL import Image
from tqdm import tqdm

def check_images(root_dir):
    broken_files = []
    for subdir, dirs, files in os.walk(root_dir):
        for file in tqdm(files):
            if file.lower().endswith(('.png', '.jpg', '.jpeg', '.tiff', '.bmp', '.gif')):
                file_path = os.path.join(subdir, file)
                try:
                    img = Image.open(file_path)  # 打开图像
                    img.verify()  # 验证图像完整性
                except (IOError, SyntaxError) as e:
                    print(f'损坏的文件: {file_path}')
                    broken_files.append(file_path)

    return broken_files

# 设置你的数据集目录
dataset_directory = 'path/to/your/dataset'  # 替换为你的数据集路径
broken_files = check_images(dataset_directory)

# 可选：将损坏的文件列表保存到文本文件中
with open('broken_files.txt', 'w') as file:
    for item in broken_files:
        file.write("%s\n" % item)

print(f'检查完成。发现 {len(broken_files)} 个损坏的文件。')
```

1. `for subdir, dirs, files in os.walk(root_dir):` - 这行代码使用 `os.walk` 函数遍历 `root_dir` 指定的根目录。`os.walk` 会生成一个三元组 `(subdir, dirs, files)`，其中：
    - `subdir` 是当前正在遍历的子目录的路径。
    
    - `dirs` 是该子目录中所有子目录的名称列表（这个例子中未使用）。
    
    - `files` 是该子目录中所有文件的名称列表。

3. `for file in tqdm(files):` - 这个循环遍历 `files` 列表中的每个文件名。`tqdm` 是一个进度条库，用于显示处理进度。

5. `if file.lower().endswith(('.png', '.jpg', '.jpeg', '.tiff', '.bmp', '.gif')):` - 这个条件判断当前文件名（转换为小写）是否以常见的图像文件扩展名结束。这确保了只有图像文件被进一步处理。

7. `file_path = os.path.join(subdir, file)` - 这行代码将子目录的路径和文件名组合成完整的文件路径。


实用脚本--批量检测图片完整性


在早期我们还需要不断地拼凑最大limits来获取交易数据

但是币安已经提供了下载链接，理论上我们只需要清洗一下数据就ok了，现在获取数据已经不用那么麻烦了,一分钟获取了一年的，效率还是很高的

![](images/%7BB6E79FE7-A100-4084-919B-60564478A7AC%7D.png)

```
import os
import zipfile
import pandas as pd
import requests

BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

def get_binance_kline(symbol, time_interval, date, file_path):
    """
    获取k线数据
    :param symbol:
    :param time_interval: 1s, 1m, 5m, 15m, 30m, 1h, 2h, 4h, 6h, 8h, 12h, 1d
    :param date: xxxx-xx 2021-01
    :param file_path:
    :return:
    """
    url = f"https://data.binance.vision/data/spot/monthly/klines/{symbol.upper()}/{time_interval}/{symbol.upper()}-{time_interval}-{date}.zip"
    response = requests.get(url)

    # 检查文件夹是否存在
    if not os.path.exists(file_path):
        os.makedirs(file_path)

    with open(f"{file_path}/{symbol.upper()}-{time_interval}-{date}.zip", "wb") as f:
        f.write(response.content)


def unzip_file(zip_src, dst_dir):
    """
    解压缩文件
    :param zip_src: 压缩文件路径
    :param dst_dir: 解压缩路径
    :return:
    """
    r = zipfile.is_zipfile(zip_src)
    if r:
        fz = zipfile.ZipFile(zip_src, 'r')
        for file in fz.namelist():
            fz.extract(file, dst_dir)
    else:
        print('This is not zip')


# 处理数据
def clear_data(timestamp):
    for file in os.listdir(f"{BASE_DIR}/data/binance/{timestamp}"):
        if file.endswith(".zip"):
            unzip_file(f"{BASE_DIR}/data/binance/{timestamp}/{file}", f"{BASE_DIR}/data/binance/{timestamp}")
            os.remove(f"{BASE_DIR}/data/binance/{timestamp}/{file}")

    # 合并csv
    for file in os.listdir(f"{BASE_DIR}/data/binance/{timestamp}"):
        if file.endswith(".csv"):
            os.system(f"cat {BASE_DIR}/data/binance/{timestamp}/{file} >> {BASE_DIR}/data/binance/{timestamp}/all.csv")
            os.remove(f"{BASE_DIR}/data/binance/{timestamp}/{file}")

    df = pd.read_csv(f"{BASE_DIR}/data/binance/{timestamp}/all.csv", names=[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11],
                     header=None)
    if df.empty:
        pass
    df.rename(columns={0: 'MTS', 1: 'open', 2: 'high', 3: 'low',
                       4: 'close', 5: 'volume'}, inplace=True)
    df['candle_begin_time'] = pd.to_datetime(df['MTS'], unit='ms')
    df = df[['candle_begin_time', 'open', 'high', 'low', 'close', 'volume']].copy()
    # 去重、排序
    df.drop_duplicates(subset=['candle_begin_time'], keep='last', inplace=True)
    df.sort_values('candle_begin_time', inplace=True)
    df.reset_index(drop=True, inplace=True)

    df.to_csv(f"{BASE_DIR}/data/binance/{timestamp}/all.csv", index=False)
    file_path = f"{BASE_DIR}/data/binance/{timestamp}/all.csv"

    return file_path

if __name__ == '__main__':
    symbol = "BTCUSDT"
    time_interval = "5m"
    start_time = "2021-01"
    end_time = "2022-02"

    # 获取当前时间戳做为文件夹名
    current_timestamp = int(time.time())

    # 遍历月份
    start_year = int(start_time.split("-")[0])
    start_month = int(start_time.split("-")[1])
    end_year = int(end_time.split("-")[0])
    end_month = int(end_time.split("-")[1])
    for year in range(start_year, end_year + 1):
        for month in range(1, 13):
            if year == start_year and month < start_month:
                continue
            if year == end_year and month > end_month:
                continue
            date = f"{year}-{month:02d}"
            logger.info(f"symbol:{symbol},date: {date}")
            get_binance_kline(symbol, time_interval, date, f"{BASE_DIR}/data/binance/{current_timestamp}")

    # 清洗数据
    file_path = clear_data(current_timestamp)

    return flie_path
```


一分钟快速获取交易对数据（Binance）


# 1.具体思路

  LLM分析PDF的话一种是可以通过PDF转文字，让大语言模型进行分析，但是这种方法存在局限性，如果PDF文字内容过多话，就涉及到最大Token的问题，即使是本地部署的大语言模型，也会遇到Token限制，或者是显存溢出问题。

  这个时候就可以用到LangChain中的Chroma来构建我们的向量知识库，具体流程如下：

PyPDFLoader(加载PDF文档)-->文本分割-->向量化-->构造向量库

# 2.实现方法

1.定义我们的文件路径并加载

```
from langchain.document_loaders import PyPDFLoader, UnstructuredFileLoader
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import Chroma

loader = PyPDFLoader(file_path)
docs = loader.load()
```

2.文本分割  
**chunk\_size**:为分割大小  
**chunk\_overlap**: 是为了避免完整的句子被分割导致LLM理解错误

```
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=200)
docs = text_splitter.split_documents(docs)
```

3.向量化并构造向量数据库  
我这边使用的Embdding模型是shibing624/text2vec-base-multilingual

```
embedding = HuggingFaceEmbeddings(model_name=config.EMBEDDING_MODEL_PATH)
# 构造向量库+conversation_id
persist_directory = os.path.join(config.KNOWLEDGE_FILE_PATH, kb_name)

# 创建向量数据库
vectordb = Chroma.from_documents(
    documents=docs,
    embedding=embedding,
    persist_directory=persist_directory
)
print("vectordb:", vectordb._collection.count())
vectordb.persist() # 向量持久话
```

4.用户提问，检索对应的片段，交给LLM处理回答  
我们可以根据用户的Query来进行匹配文章内容，进而构造prompt来进行提问

```
embedding = HuggingFaceEmbeddings(model_name=config.EMBEDDING_MODEL_PATH)
vectordb = Chroma(persist_directory=persist_directory, embedding_function=embedding)
print(vectordb._collection.count())
logger.info("Load database building finished. %s", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))

# docs = vectordb.similarity_search(query, k=3)
docs = vectordb.similarity_search(query, k=5)

page = list(set([docs.metadata['page'] for docs in docs]))
page.sort()

context = [docs.page_content for docs in docs]
prompt = f"已知PDF内容：\n{context}\n根据已知信息回答问题：\n{query}\n所有的回答都根据已知信息"
```

**GitHub已经开源：**  
[https://github.com/1264204425/PdfReader-LangChian-LLM](https://github.com/1264204425/PdfReader-LangChian-LLM)  
代码写的较烂轻喷

![](images/llm01.png)

![](images/llm02.png)


使用LangChain实现大语言模型对话解析PDF


### Linux 命令参考手册（Markdown格式）

#### 目录操作

```
markdownCopy code
- `rm -rf mydir`：删除名为'mydir'的目录。
- `cd mydir`：进入名为'mydir'的目录。
- `cd -`：回到上一级目录。
- `cd ..`：回到父目录。
- `cd ~`：回到根目录。
- `mv tools tool`：将名为'tools'的目录重命名为'tool'。
- `ln -s tool bac`：为名为'tool'的目录创建名为'bac'的符号链接。
- `cp -a tool /home/vpser/www`：将'tool'目录下的所有文件复制到'/home/vpser/www'目录下。
```

#### 文件操作

```
markdownCopy code
- `rm go.tar`：删除名为'go.tar'的文件。
- `find mt.cgi`：查找文件名为'mt.cgi'的文件。
- `df -h`：查看磁盘剩余空间。
```

#### 解压缩

```
markdownCopy code
- `tar xvf wordpress.tar`：解压tar格式的文件。
- `tar -tvf myfile.tar`：查看tar文件中包含的文件。
- `tar cf toole.tar tool`：将'tool'目录打包为'toole.tar'文件。
- `tar cfz vpser.tar.gz tool`：将'tool'目录打包并压缩为'vpser.tar.gz'文件。
- `tar jcvf /var/bak/www.tar.bz2 /var/www/`：创建'.tar.bz2'文件，压缩率高。
- `tar xjf www.tar.bz2`：解压'tar.bz2'格式文件。
- `gzip -d ge.tar.gz`：解压'.tar.gz'文件为'.tar'文件。
- `unzip phpbb.zip`：解压zip文件。
```

#### 下载

```
markdownCopy code
- `wget http://soft.vpser.net/web/nginx/nginx-0.8.0.tar.gz`：从远程服务器下载文件到本地服务器。
- `wget -c http://soft.vpser.net/web/nginx/nginx-0.8.0.tar.gz`：继续下载上次未完成的文件。
```

#### 进程管理

```
markdownCopy code
- `ps -aux`：查看进程状态。
- `ps -aux | grep nginx`：在所有进程中查找nginx的进程。
- `kill 1234`：结束进程ID为1234的进程。
- `killall nginx`：通过程序名结束所有名为nginx的进程。
```

#### Vim操作

```
markdownCopy code
- 移动类命令：如 `h/j/k/l`，`w`，`b`，`e` 等。
- 跳转类命令：如 `"`，`CTRL-O`，`CTRL-I` 等。
- 查找类命令：如 `/`，`?`，`n` 等。
- 编辑类命令：如 `i`，`x`，`.`，`u`，`CTRL-R`，`p` 等。
- 退出保存命令：如 `:q`，`:q!`，`ZZ`，`:e!` 等。
```


Linux常用操作


## 基础框架

```
import logging
import os
import uvicorn
from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware

env = os.environ
# app = FastAPI(docs_url=None, redoc_url=None)  # 用于禁用swagger文档
app = FastAPI()

# 日志配置
logger = logging.getLogger()
logger.setLevel(logging.INFO)
formatter = logging.Formatter(
    "%(asctime)s - %(module)s - %(funcName)s - line:%(lineno)d - %(levelname)s - %(message)s"
)
# 输出到控制台
to_console = logging.StreamHandler()
to_console.setFormatter(formatter)
logger.addHandler(to_console)

# 输出到文件中
to_file = logging.FileHandler(filename="log.txt")
to_file.setFormatter(formatter)
logger.addHandler(to_file)

# 配置CORS
origins = [
    "*"
]
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# def

if __name__ == '__main__':
    # 启动服务
    host = env.get("HOST") if env.get("HOST") is not None else "0.0.0.0"
    port = int(env.get("PORT")) if env.get("PORT") is not None else 8000
    uvicorn.run(app='api_service:app', host=host, port=port, reload=True)
```

## Json请求

```
class ChatPdfRequest(BaseModel):
    conversation_id: str
    query: str
    history: List[Tuple[str, str]]


@app.post("/ChatPdf", tags=["Chat"], summary="与llm模型对话")
async def chatpdf(
        ChatPdfRequest: ChatPdfRequest = Body(..., content_type="application/json", description="对话信息")):
```

## 日志写入

```
# 日志记录器
logger = logging.getLogger()
 
# 设置日志级别，只有大于等于这个级别的日志才能输出
logger.setLevel(logging.ERROR) # 有DEBUG INFO WARRING ERROR CRITICAL
 
# 设置日志格式
formatter = logging.Formatter(
    "%(asctime)s - %(module)s - %(funcName)s - line:%(lineno)d - %(levelname)s - %(message)s"
)
 
# 输出到控制台
to_console = logging.StreamHandler()
to_console.setFormatter(formatter)
logger.addHandler(to_console)
 
# 输出到文件中
to_file = logging.FileHandler(filename="log.txt")
to_file.setFormatter(formatter)
logger.addHandler(to_file)
```

## 更好的日志写入

```
def getLogger(name, file_name, use_formatter=True):
    logger = logging.getLogger(name)
    logger.setLevel(logging.INFO)
    console_handler = logging.StreamHandler(sys.stdout)
    formatter = logging.Formatter('%(asctime)s    %(message)s')
    console_handler.setFormatter(formatter)
    console_handler.setLevel(logging.INFO)
    logger.addHandler(console_handler)
    if file_name:
        handler = logging.FileHandler(file_name, encoding='utf8')
        handler.setLevel(logging.INFO)
        if use_formatter:
            formatter = logging.Formatter('%(asctime)s - %(name)s - %(message)s')
            handler.setFormatter(formatter)
        logger.addHandler(handler)
    return logger

logger = getLogger('ChatGLM', 'chatlog.log')
```


FastAPi操作


# 安装

官网：[https://www.anaconda.com/download](https://www.anaconda.com/download)

点下一步安装安装

# 基本操作

```
conda create -n env_name python=xx版本
```

```
conda activate env_name
```

```
pip install --upgrade pip
pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple
```

```
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main
conda config --set show_channel_urls yes
```

# 创建指定文件夹环境

1.创建的指定文件夹

首先，创建一个无名称的环境：

```
bash复制代码conda create --prefix E:\condaenv\Anaconda3\envs\pytorch
```

然后，激活该环境：

```
bash复制代码conda activate E:\condaenv\Anaconda3\envs\pytorch
```

最后，使用--name参数给环境命名：

```
bash复制代码conda env update --name pytorch --file C:\Users\12642\Desktop\ymlcunfang\pytorch.yml
```

# MAC出现问题

在mac中有可能出现命令丢失的情况这时候需要

```
source activate
```


👋 你好我是 Joe

最新文章

本地部署 DeepSeek R1 教程

Docker Node 环境打包减少打包体积

【Go语言】Gin框架下，body参数只能读取一次解决方案

AI总结：Apple’s On-Device and Server Foundation Models

简单案例Transformer-翻译任务简单案例Transformer-

👋 你好 我是 Joe

最新文章

本地部署 DeepSeek R1 教程

Docker Node 环境打包 减少打包体积

【Go语言】Gin框架下，body参数只能读取一次解决方案

AI总结：Apple’s On-Device and Server Foundation Models

简单案例Transformer-翻译任务简单案例Transformer-

👋 你好我是 Joe

Docker Node 环境打包减少打包体积