本地部署

本地部署教程

1. 使用

1.1. 添加文件

文件名：finance.txt

【贷款利率政策】
当前贷款利率由央行基准利率决定，并受到市场因素影响。对于个人住房贷款，银行一般采用 LPR（贷款市场报价利率）+ 加点的方式确定最终利率。企业贷款利率则根据信用评级、行业风险等进行调整。

【贷款审批条件】
银行对贷款申请人的信用评分、收入水平、资产状况进行综合评估。一般要求：
1. 信用评分在 600 分以上，无重大逾期记录；
2. 个人住房贷款首付款比例至少 20%-30%；
3. 企业贷款需提供财务报表，部分情况需抵押物。

【还款方式】
贷款还款方式主要有两种：等额本息与等额本金。等额本息前期压力较小，但总利息较高；等额本金前期还款较多，但利息总支出较少。借款人可根据自身经济情况选择适合的方式。

【提前还款政策】
大部分银行允许提前还款，但可能收取违约金。提前还款是否划算，需考虑贷款利率、剩余期限等因素。一般来说，如果贷款剩余年限较长，提前还款可减少利息支出。

【企业贷款优惠政策】
部分政府支持行业（如科技创新、绿色能源）可获得低息贷款支持。中小企业可申请信用贷款或政府贴息贷款，以降低融资成本。

1.2. 运行

1.2.1. 传统方式

import os
import requests
import chromadb
from chromadb.utils import embedding_functions
from chromadb.config import Settings
# 🔹 定义 ChromaDB 存储路径
DB_DIR = "./chroma_db"

# 🔹 定义 Ollama API 地址
OLLAMA_EMBEDDING_URL = "http://127.0.0.1:11434/api/embeddings"

# 🔹 定义本地模型
OLLAMA_MODEL = "deepseek-r1:latest"

# 1️⃣ **加载 linux.txt 并切分文本**
def load_and_split_text(file_path, chunk_size=100, overlap=20):
    """ 读取文本文件并进行分块 """
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            text = f.read()

        chunks = []
        for i in range(0, len(text), chunk_size - overlap):
            chunks.append(text[i : i + chunk_size])

        print(f"✅ 成功加载 {file_path}，共分割 {len(chunks)} 个块")
        return chunks
    except Exception as e:
        print(f"❌ 读取 {file_path} 失败: {e}")
        return []

# 2️⃣ **调用 Ollama API 获取文本向量**
def get_embedding(text):
    """ 通过 Ollama API 获取文本的向量表示 """
    try:
        data = {"model": "nomic-embed-text", "prompt": text}
        response = requests.post(OLLAMA_EMBEDDING_URL, json=data, timeout=10)

        if response.status_code == 200:
            return response.json().get("embedding", [])
        else:
            print(f"❌ 向量化失败，状态码: {response.status_code}, 响应: {response.text}")
            return None
    except requests.exceptions.RequestException as e:
        print(f"❌ 请求 Ollama 失败: {e}")
        return None

# 3️⃣ **存入 ChromaDB**
def store_in_chroma(text_chunks):
    """ 将文本向量存入 ChromaDB 持久化数据库 """
    try:
        client = chromadb.PersistentClient(path=DB_DIR)
        collection = client.get_or_create_collection(name="linux")
        for i, chunk in enumerate(text_chunks):
            embedding = get_embedding(chunk)
            if embedding:
                collection.add(
                    ids=[f"doc_{i}"],  # 设定唯一 ID
                    embeddings=[embedding],
                    documents=[chunk]
                )

        print("✅ 数据已成功存入 ChromaDB！")
    except Exception as e:
        print(f"❌ 存入 ChromaDB 失败: {e}")

# 4️⃣ **查询 ChromaDB**
def query_chroma(query_text, top_k=3):
    """ 从 ChromaDB 查询相关文本 """
    try:
        client = chromadb.PersistentClient(path=DB_DIR)
        collection = client.get_collection(name="linux")
        query_vector = get_embedding(query_text)
        if query_vector:
            results = collection.query(
                query_embeddings=[query_vector],
                n_results=top_k
            )
            return results.get("documents", [[]])[0]
        else:
            return []
    except Exception as e:
        print(f"❌ 查询 ChromaDB 失败: {e}")
        return []

# 5️⃣ **让 deepseek-r1:1.5b 回答问题**
def ask_local_model(context, question):
    """ 让本地 deepseek-r1:1.5b 模型回答问题 """
    try:
        ollama_url = "http://127.0.0.1:11434/api/generate"
        data = {
            "model": OLLAMA_MODEL,
            "prompt": f"根据以下内容回答问题：\n{context}\n\n问题：{question}",
            "stream": False
        }
        print(data)
        print("===========================")

        response = requests.post(ollama_url, json=data, timeout=40)
        if response.status_code == 200:
            return response.json().get("response", "⚠️ 无法生成回答")
        else:
            print(f"❌ deepseek-r1:1.5b生成回答失败，状态码: {response.status_code}, 响应: {response.text}")
            return "⚠️ deepseek-r1:1.5b生成回答失败"
    except requests.exceptions.RequestException as e:
        print(f"❌ 请求 deepseek-r1:1.5b 失败: {e}")
        return "⚠️ deepseek-r1:1.5b 生成回答失败"

# **运行流程**
if __name__ == "__main__":
    try:
        # 1️⃣ 读取 linux.txt 并存入 ChromaDB
        linux_file = "linux.txt"
        if os.path.exists(linux_file):
            text_chunks = load_and_split_text(linux_file)
            if text_chunks:
                store_in_chroma(text_chunks)
        else:
            print(f"❌ 文件 {linux_file} 不存在！")

        # 2️⃣ 处理查询
        question = "Linux常用命令有哪些？"
        retrieved_docs = query_chroma(question)
        print("\n🔹 **retrieved_docs结果** 🔹\n", retrieved_docs)

        if retrieved_docs:
            context = "\n".join(retrieved_docs)
            answer = ask_local_model(context, question)
            print("\n🔹 **最终回答** 🔹\n", answer)
        else:
            print("⚠️ 没有找到相关信息！")

    except Exception as e:
        print(f"❌ 运行主流程时发生错误: {e}")

1.2.2. Langchain 方式

import os
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain_community.embeddings import OllamaEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.chat_models import ChatOllama
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 🔹 定义 ChromaDB 存储路径
DB_DIR = "./chroma_db"

# 🔹 使用 Ollama 本地模型
OLLAMA_MODEL = "deepseek-r1:latest"
EMBEDDING_MODEL = "nomic-embed-text"

# 1️⃣ **加载 linux.txt 并切分文本**
def load_and_split_text(file_path, chunk_size=200, overlap=20):
    try:
        loader = TextLoader(file_path, encoding="utf-8")
        documents = loader.load()

        text_splitter = CharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=overlap)
        doc_splits = text_splitter.split_documents(documents)

        print(f"✅ 成功加载 {file_path}，共分割 {len(doc_splits)} 个块")
        return doc_splits
    except Exception as e:
        print(f"❌ 读取 {file_path} 失败: {e}")
        return []

# 2️⃣ **使用 LangChain 进行向量化**
def store_in_chroma(doc_splits):
    try:
        embedding_function = OllamaEmbeddings(model=EMBEDDING_MODEL)
        vectorstore = Chroma.from_documents(doc_splits, embedding_function, persist_directory=DB_DIR)
        print("✅ 数据已成功存入 ChromaDB！")
        return vectorstore
    except Exception as e:
        print(f"❌ 存入 ChromaDB 失败: {e}")
        return None

# 3️⃣ **查询 ChromaDB**
def query_chroma(query_text, vectorstore, top_k=3):
    try:
        retriever = vectorstore.as_retriever(search_kwargs={"k": top_k})
        retrieved_docs = retriever.get_relevant_documents(query_text)
        return [doc.page_content for doc in retrieved_docs]
    except Exception as e:
        print(f"❌ 查询 ChromaDB 失败: {e}")
        return []

# 4️⃣ **使用 LangChain 让 deepseek-r1:1.5b 回答**
def ask_local_model(context, question):
    try:
        model_local = ChatOllama(model=OLLAMA_MODEL)

        template = """基于以下内容回答问题：
        {context}
        问题：{question}"""
        prompt = ChatPromptTemplate.from_template(template)

        chain = (
            {"context": RunnablePassthrough(), "question": RunnablePassthrough()}
            | prompt
            | model_local
            | StrOutputParser()
        )

        return chain.invoke({"context": context, "question": question})
    except Exception as e:
        print(f"❌ deepseek-r1:1.5b 生成回答失败: {e}")
        return "⚠️ 生成回答失败"

# **运行流程**
if __name__ == "__main__":
    try:
        # 1️⃣ 读取 linux.txt 并存入 ChromaDB
        linux_file = "linux.txt"
        if os.path.exists(linux_file):
            doc_splits = load_and_split_text(linux_file)
            if doc_splits:
                vectorstore = store_in_chroma(doc_splits)
        else:
            print(f"❌ 文件 {linux_file} 不存在！")
            vectorstore = None

        # 2️⃣ 处理查询
        if vectorstore:
            question = "Linux常用命令有哪些？"
            retrieved_docs = query_chroma(question, vectorstore)
            print("\n🔹 **retrieved_docs 结果** 🔹\n", retrieved_docs)

            if retrieved_docs:
                context = "\n".join(retrieved_docs)
                answer = ask_local_model(context, question)
                print("\n🔹 **最终回答** 🔹\n", answer)
            else:
                print("⚠️ 没有找到相关信息！")

    except Exception as e:
        print(f"❌ 运行主流程时发生错误: {e}")