开始
│
├─ 显存 < 8GB?
│  ├─ 是 → 选择 7B 量化模型 (Qwen3-7B, CodeQwen-7B)
│  └─ 否 → 继续
│
├─ 显存 8-16GB?
│  ├─ 是 → 选择 13B/14B 量化模型 (Qwen3-14B, Llama-3.3-8B)
│  └─ 否 → 继续
│
├─ 显存 16-32GB?
│  ├─ 是 → 选择 70B 4-bit 或 13B 全精度
│  └─ 否 → 继续
│
└─ 显存 > 32GB?
   └─ 选择 70B 全精度或多模型并行部署

# Modelfile 示例：针对代码生成优化的配置
FROM qwen3:14b

# 系统提示词：定义AI助手的角色和行为
SYSTEM """You are an expert software engineer. 
When writing code, follow these principles:
1. Write clean, readable code with meaningful variable names
2. Add comments for complex logic
3. Consider edge cases and error handling
4. Follow language-specific best practices"""

# 参数调优
PARAMETER temperature 0.2        # 低温度以获得更确定的输出
PARAMETER top_p 0.9              # 适度的多样性
PARAMETER top_k 40               # 限制候选词数量
PARAMETER num_ctx 32768          # 32K上下文窗口
PARAMETER num_predict 2048       # 最大生成长度
PARAMETER repeat_penalty 1.1     # 适度抑制重复

# GPU层卸载：将所有层都放在GPU上
PARAMETER num_gpu 999

import requests
import concurrent.futures

def generate(prompt):
    response = requests.post('http://localhost:11434/api/generate', json={
        'model': 'qwen3:14b',
        'prompt': prompt,
        'stream': False
    })
    return response.json()['response']

# 并行处理多个请求
prompts = [
    "Explain Python decorators",
    "Explain Python generators",
    "Explain Python context managers"
]

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(generate, prompts))

import os
from pathlib import Path
from langchain_community.document_loaders import TextLoader, PDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.embeddings import OllamaEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.llms import Ollama
from langchain.chains import RetrievalQA

class LocalRAGSystem:
    def __init__(self, model_name="qwen3:14b", embed_model="bge-m3"):
        self.embeddings = OllamaEmbeddings(model=embed_model)
        self.llm = Ollama(model=model_name, temperature=0.1)
        self.vectorstore = None
        self.qa_chain = None
        
    def ingest_documents(self, docs_path: str):
        """将文档目录中的所有文件加载到向量数据库"""
        documents = []
        
        for file_path in Path(docs_path).rglob('*'):
            if file_path.suffix in ['.txt', '.md', '.py', '.js']:
                loader = TextLoader(str(file_path), encoding='utf-8')
                documents.extend(loader.load())
            elif file_path.suffix == '.pdf':
                loader = PDFLoader(str(file_path))
                documents.extend(loader.load())
        
        # 文档分块
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            separators=["

", "
", ". ", " ", ""]
        )
        chunks = text_splitter.split_documents(documents)
        
        # 创建向量数据库
        self.vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory="./chroma_db"
        )
        
        # 构建RAG链
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=self.vectorstore.as_retriever(
                search_kwargs={"k": 5}
            ),
            return_source_documents=True
        )
        
        print(f"成功加载 {len(documents)} 个文档，切分为 {len(chunks)} 个块")
    
    def query(self, question: str) -> dict:
        """查询知识库"""
        if not self.qa_chain:
            raise ValueError("请先调用 ingest_documents 加载文档")
        
        result = self.qa_chain.invoke({"query": question})
        return {
            "answer": result["result"],
            "sources": [doc.metadata for doc in result["source_documents"]]
        }

# 使用示例
rag = LocalRAGSystem()
rag.ingest_documents("./my_knowledge_base/")

result = rag.query("什么是MCP协议？")
print(f"回答：{result['answer']}")
print(f"参考来源：{result['sources']}")

def update_document(self, doc_id: str, new_content: str, metadata: dict):
    """更新或添加单个文档"""
    # 删除旧版本（如果存在）
    self.vectorstore.delete(ids=[doc_id])
    
    # 添加新版本
    self.vectorstore.add_texts(
        texts=[new_content],
        metadatas=[metadata],
        ids=[doc_id]
    )
    
    # 持久化
    self.vectorstore.persist()

{
  "models": [
    {
      "title": "Local Qwen3",
      "provider": "ollama",
      "model": "qwen3:14b"
    }
  ],
  "tabAutocompleteModel": {
    "title": "Local Code Model",
    "provider": "ollama",
    "model": "codestral:22b"
  }
}

# 本地自动化脚本示例：RSS摘要生成
import feedparser
from local_rag import LocalRAGSystem

rag = LocalRAGSystem()

feeds = [
    "https://news.ycombinator.com/rss",
    "https://dev.to/feed"
]

for feed_url in feeds:
    feed = feedparser.parse(feed_url)
    for entry in feed.entries[:5]:
        summary = rag.query(
            f"请用中文总结以下文章的要点（限制在100字内）：{entry.summary}"
        )
        print(f"[{entry.title}] {summary['answer']}")

import time
import statistics

def benchmark_model(model_name, test_prompts, warmup=3, runs=10):
    """基准测试本地模型性能"""
    # 预热
    for _ in range(warmup):
        requests.post('http://localhost:11434/api/generate', json={
            'model': model_name,
            'prompt': test_prompts[0],
            'stream': False
        })
    
    latencies = []
    for prompt in test_prompts:
        run_times = []
        for _ in range(runs):
            start = time.time()
            requests.post('http://localhost:11434/api/generate', json={
                'model': model_name,
                'prompt': prompt,
                'stream': False
            })
            run_times.append(time.time() - start)
        latencies.append(statistics.median(run_times))
    
    return {
        'model': model_name,
        'avg_latency': statistics.mean(latencies),
        'min_latency': min(latencies),
        'max_latency': max(latencies)
    }

# 验证Ollama模型哈希
ollama show qwen3:14b --modelfile
# 检查模型来源和配置