LLM 大语言模型实战 (九)-深入掌握大模型应用开发框架 LangChain

2024-09-25 17:29:43 ⋅ 614 ⋅ 0 ⋅ 0

一、LangChain 简介

LangChain 是一个用于开发由大型语言模型（LLM）支持的应用程序的框架。

LangChain核心模块：

  • Models: 从不同的LLM，Chat模型中选择
  • Prompts: 管理LLM的输入
  • Chains: 将LLM和其他模块相结合
  • Retrieval: 访问外部数据
  • Memory: 历史的对话信息
  • Agents: 智能体用来访问其他工具（函数）

二、Agents

file

三、实战

1、建立向量库

使用 OpenAI 的api来做词向量嵌入 OpenAIEmbeddings：

# from langchain.embeddings import OpenAIEmbeddings
# pip install python-dotenv
from pathlib import Path
from loguru import logger
from dotenv import dotenv_values
import tyro

from langchain_community.document_loaders import (
    TextLoader,
    CSVLoader,
    BSHTMLLoader,
    UnstructuredMarkdownLoader,
)

from langchain.text_splitter import CharacterTextSplitter

# ---- langchain==0.1.2 旧方法
# from langchain.embeddings import OpenAIEmbeddings
#from langchain.embeddings import ErnieEmbeddings  # 方法已过时
#from langchain.vectorstores import FAISS # 方法已过时

# langchain == 0.3.1 版本需要单独安装
# pip install -U langchain-community
from langchain_community.vectorstores import FAISS
# from langchain_community.embeddings import ErnieEmbeddings
# from langchain_community.embeddings import QianfanEmbeddingsEndpoint
# pip install openai   # 使用OpenAI必须安装
# from langchain_community.embeddings import OpenAIEmbeddings
from langchain_openai import OpenAIEmbeddings

# 使用开源库的Embedding
# pip install sentence-transformers
from langchain_community.embeddings import HuggingFaceEmbeddings

env_config = dotenv_values("../.env")

"""
doc1 : [1, 2, 3, , ...., 10, 12]
doc2: [8, 9, 10, ... 20]
"""
text_splitter = CharacterTextSplitter(
    chunk_size=400,
    chunk_overlap=200,
)

BASE_PATH = Path("../data")

def load_csv():  # -> List[Document]
    # load csv 的办法
    csv_loader = CSVLoader(
        file_path=BASE_PATH / "info.csv",
        metadata_columns=[
            "user_id",
        ],
        csv_args={
            "delimiter": ",",
            "quotechar": '"',
            "fieldnames": [
                "user_id",
                "name",
                "birthday",
                "interest",
                "personal_website",
            ],
        },
    )

    csv_docs = csv_loader.load()

    logger.info(len(csv_docs))
    logger.info(type(csv_docs))
    logger.info(csv_docs[1].page_content)
    return csv_docs

def load_html():
    # load html
    html_loader = BSHTMLLoader(file_path=BASE_PATH / "query.html")
    html_docs = html_loader.load()

    logger.info(len(html_docs))
    logger.info(html_docs[0].page_content)
    return html_docs

def load_md():
    # load md
    md_loader = UnstructuredMarkdownLoader(BASE_PATH / "info.md")
    md_docs = md_loader.load()

    logger.info(md_docs)
    return md_docs

def load_text():
    ## load text
    text_loader = TextLoader(BASE_PATH / "text.txt")
    text_docs = text_loader.load()

    logger.info(text_docs)
    return text_docs

def load_local_data():
    text_docs = load_text()
    # md_docs = load_md()
    # html_docs = load_html()
    # csv_docs = load_csv()

    # 上面一共有有几种 docs
    # all_docs = [text_docs, md_docs, html_docs, csv_docs]
    all_docs = [
        text_docs,
    ]
    # 可以看到 document 是一样的；
    for item in all_docs:
        logger.info(f"type is is: {type(item[0])}")
    # doc item 很多
    # all_docs_item = [*text_docs, *md_docs, *html_docs, *csv_docs]
    all_docs_item = [
        *text_docs,
    ]
    # 进行 chunk 化;
    small_docs = text_splitter.transform_documents(all_docs_item)
    logger.info(len(small_docs))
    logger.info(small_docs[0])
    return small_docs

def prepare_db():
    small_docs = load_local_data()
    # embedding model

    emb_model = OpenAIEmbeddings(
        api_key=env_config.get("API_KEY"),
        base_url=env_config.get("BASE_URL"),
    )

    print("----------- small docs ----------")
    print(small_docs)

    logger.info("保存文件到本地向量库...")

    # 使用 FAISS 创建索引
    db = FAISS.from_documents(small_docs, emb_model)

    # 保存索引到本地
    db.save_local("../data/db_indexv2")

    logger.info("保存文件到本地向量库End...")

def load_db() -> FAISS:
   # embedding model

   # -------- 这些Embedding 都不能用 Begin -----------
   # emb_model = ErnieEmbeddings(api_key=env_config.get("API_KEY"), secret_key=env_config.get("secret_key"))
   # emb_model = HuggingFaceEmbeddings(model_name="jinaai/jina-embeddings-v3")
   # emb_model = HuggingFaceEmbeddings(model_name="EleutherAI/gpt-j-6B")
   # -------- 这些Embedding 都不能用 End -----------

    emb_model = OpenAIEmbeddings(
        api_key=env_config.get("API_KEY"),
        base_url=env_config.get("BASE_URL"),
    )

    print("api_key:", env_config.get("API_KEY"))

    logger.info("embedding model OPENAI.")

    # 加载索引
    # pip install faiss-cpu
    # pip install tiktoken
    db = FAISS.load_local("../data/db_indexv2", emb_model, allow_dangerous_deserialization=True)

    logger.info("-------db info --------")
    print(db)
    return db

def main(pre_db: bool = False):
    # 是否是提前准备 data base
    # pre_db = True
    if pre_db:
        prepare_db()
    else:
        db = load_db()
        # 进行检索
        query = "你好"
        res = db.similarity_search(query, k=3)
        logger.info(res)
        # 进行检索
        query = "你好"
        res = db.similarity_search_with_score(query, k=3)
        logger.info(res)

        # 进行检索
        query = "RAG本质"
        res = db.similarity_search_with_score(query, k=3)
        logger.info(res)
        # 进行检索
    # tyro, loguru

if __name__ == "__main__":
    # main()
    tyro.cli(main)
    # python prepare_data.py --pre-db

.env 配置文件:

#openai
API_KEY=sk-27Rbxxx244
BASE_URL=https://open.xiaojingai.com/v1

打印输出：

/Users/kaiyi/miniconda3/envs/llm/bin/python /langchain-local-qa-course/src/prepare_data.py
api_key: sk-27Rb7QfL8Hxxx74244
2024-09-28 14:52:21.222 | INFO     | __main__:load_db:172 - embedding model OPENAI.
<langchain_community.vectorstores.faiss.FAISS object at 0x7fc9a8319730>
2024-09-28 14:52:21.320 | INFO     | __main__:load_db:179 - -------db info --------
2024-09-28 14:52:25.770 | INFO     | __main__:main:194 - [Document(metadata={'source': '../data/text.txt'}, page_content='通用语言模型通过微调就可以完成几类常见任务，比如分析情绪和识别命名实体。这些任务不需要额外的背景知识就可以完成。\n\n最近，基于检索器的方法越来越流行，\n\n\n要完成更复杂和知识密集型的任务，可以基于语言模型构建一个系统，访问外部知识源来做到。这样的实现与事实更加一性，生成的答案更可靠，还有助于缓解“幻觉”问题。')]
2024-09-28 14:52:26.240 | INFO     | __main__:main:198 - [(Document(metadata={'source': '../data/text.txt'}, page_content='通用语言模型通过微调就可以完成\n\n\n要完成更复杂和知识密集型的任务，可以基于语言模型构建一个系统，访问外部知识源来做到。这样的实现与事实更加一性，生成的答案更可靠，还有助于缓解“幻觉”问题。'), 0.5583868)]
2024-09-28 14:52:26.710 | INFO     | __main__:main:203 - [(Document(metadata={'source': '../data/text.txt'}, page_content='RAG的本质是让模型获取正确的Context(上下文)，利用ICL (In Context Learning)的能力，输出正确的响应。它综合利用了固化在模型权重中的参数化知识和存在外部存储中的非参数化知识(知识库、数据库等)。RAG分为两阶段：使用编码模型（如 BM25、DPR、ColBERT 等）这样 RAG 更加适应事实会随时间变化的情况。这非常有用，因为 LLM 的参数化知识是静态的。RAG 让语言模型不用重新训练就能够获取最新的信息，基于检索生成产生可靠的输出。'), 0.35194033)]

Process finished with exit code 0

四、BM25

BM25 在业界用的非常多，个人认为没有办法被向量召回完全替代的，至少可以作为补充

使用 BM25Retriever 的主要优势在于它可以在初步筛选阶段提供更高的精确度，尤其是在处理短文本或查询时。BM25算法基于词频和逆文档频率（TF-IDF），适合处理信息检索中的经典查询需求，可以有效捕捉到文档中的重要关键词。而Embedding方法则通过语义相似性进行匹配，适合处理更复杂的上下文。

结合这两种方法，可以利用BM25的精确性和Embedding的语义理解能力，形成一个强大的检索系统。通过EnsembleRetriever，可以更好地平衡不同类型查询的表现，最终提升检索效果。这样，用户在检索时能获得更全面、更相关的结果。

file

def get_retriver() -> EnsembleRetriever:
    text_loader = TextLoader(
        "/Users/kaiyi/Work/AI/immoc/源码+PDF课件/源码+PDF课件/源码/llm/langchain_code/langchain-local-qa-course/data/text.txt"
    )
    text_docs = text_loader.load()

    vectorstore = FAISS.from_documents(
        text_docs,
        OpenAIEmbeddings(
            api_key=ENV_CONFIG.get("API_KEY"), base_url=ENV_CONFIG.get("BASE_URL")
        ),
    )

        # 使用 BM25 提高精确度
    bm25_retriever = BM25Retriever.from_documents(documents=text_docs, k=1)
    emb_retriever = vectorstore.as_retriever(search_kwargs={"k": 1})
    ensemble_retriever = EnsembleRetriever(
        retrievers=[bm25_retriever, emb_retriever], weights=[0.5, 0.5]
    )
    return ensemble_retriever

五、部署大模型服务

两行代码实现一个大模型的在线服务，部署一个后端的服务：

# uvicorn , nginx 相关的东西部署后端服务；
# Batch , 异步等
# -------- 相关依赖 --------
#  pip install rank_bm25
# pip install sse_starlette

from loguru import logger
from dotenv import dotenv_values

from langchain.retrievers import BM25Retriever, EnsembleRetriever
from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings
from langchain_community.document_loaders import TextLoader

from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain.vectorstores import FAISS

# 添加的
from langserve import add_routes
from fastapi import FastAPI

app = FastAPI(
    title="LangChain Server",
    version="1.0",
    description="A simple api server using Langchain's Runnable interfaces",
)

logger.info("init app succ")

ENV_CONFIG = dotenv_values("../.env")

def get_retriver() -> EnsembleRetriever:
    text_loader = TextLoader(
        "/Users/kaiyi/Work/AI/immoc/源码+PDF课件/源码+PDF课件/源码/llm/langchain_code/langchain-local-qa-course/data/text.txt"
    )
    text_docs = text_loader.load()

    vectorstore = FAISS.from_documents(
        text_docs,
        OpenAIEmbeddings(
            api_key=ENV_CONFIG.get("API_KEY"), base_url=ENV_CONFIG.get("BASE_URL")
        ),
    )
    bm25_retriever = BM25Retriever.from_documents(documents=text_docs, k=1)
    emb_retriever = vectorstore.as_retriever(search_kwargs={"k": 1})
    ensemble_retriever = EnsembleRetriever(
        retrievers=[bm25_retriever, emb_retriever], weights=[0.5, 0.5]
    )
    return ensemble_retriever

def build_chain():
    # template = """Answer the question based only on the following context:
    # {context}

    # Question: {question}
    # """
    template = """Answer the question based on the following context.

    {context}

    Question: {question}

    If context is not useful. answer the question directly and Do not refer to context
    """

    prompt = ChatPromptTemplate.from_template(template)

    model = ChatOpenAI(
        api_key=ENV_CONFIG.get("API_KEY"), base_url=ENV_CONFIG.get("BASE_URL")
    )
    retriver = get_retriver()
    chain = (
        {"context": retriver, "question": RunnablePassthrough()}
        | prompt
        | model
        | StrOutputParser()
    )
    return chain

def build_joke_chain():
    template = """帮我写一个关于{topic}的笑话"""
    prompt = ChatPromptTemplate.from_template(template)

    model = ChatOpenAI(
        api_key=ENV_CONFIG.get("API_KEY"), base_url=ENV_CONFIG.get("BASE_URL")
    )
    joke_chain = prompt | model | StrOutputParser()
    return joke_chain

def api():
    import uvicorn

    qa_chain = build_chain()
    add_routes(app, qa_chain, path="/qa")

    joke_chain = build_joke_chain()
    add_routes(app, joke_chain, path="/joke")

    uvicorn.run(app, host="localhost", port=8005)

def test():
    qa_chain = build_chain()
    logger.info(qa_chain.invoke("RAG的本质是什么？"))

    joke_chain = build_joke_chain()
    logger.info(joke_chain.invoke({"topic": "cat"}))

def main():
    # test()
    api()

if __name__ == "__main__":
    main()

启动服务：
file

部署的服务界面：
file

LangChain -面试挖掘的点

面试的过程：

  • 算法是实验科学:
    • 至少现在大家都不太理解大模型
    • 结果无法保证（温度--> 0)

  • 面试:（做项目->公司）

    • 项目增色，有特点（闪光点）
    • 因为内部数据稀少，需要利用公域数据， 所以结合了 wikipidea，构建了一个更大规模的本地知识库
    • 检索方式
      • 因为数据太多，优化检索方式
      • BM 25 + Emb
        • BGE, text2vec(shibing624_text2vec-base-chinese 中文更友好）, m3e
        • Lost in middle -> 重排序（遇到了什么问题，发现了，找文档）
      • 生成 Prompt
      • SFT (项目上线一段时间后，有了数据，然后微调，retrain 了一下）/ICL/CoT
        GPT4 改写 -> 更接近 GPT

    面试：你为什么这么做？这么做带来了什么效果？回复需要合理性；

    file

    搭建本地私有知识库流程：

    • 数据来源：
      • 爬虫爬取
      • 本地知识库（Hive，hdfs, 文本，关系型数据）；切分粒度
    • 检索方式
      • BM 25 + Emb
      • EMB尝试用 BGE，text2vec, M3E
      • 阈值设定，Lost in Middle -> 重排
      • Query Rewrite, Query pretrain
    • 生成答案
      • CoT (Chain of Thout)
      • ICL (In Context Learning)
        • 动态检索
        • 数量
        • RAG 反思

    QA

    • 数据太多？

      • 数据压缩（txt -> zip x)
      • 大文本 -> 大模型总结内容 （内容压缩）

    • 数据预处理导致数据质量不高？

      • 不是问题，算法工程师把数据处理好；
        • 更精细化
      • 产品形态上让用户帮你打标 -> 更高质量

    • 对生成的答案，增加来自于知识库的哪一条记录，就像bing的chat 那样，这个怎么实现：

      • 这个在RAG的 notebook 里面 Chain2 返回 docs 记录
      • 让大模型生成

    • 怎么样用国产大模型比如千问来结合 Langchain 加载本地文档

    • 切分文档的时候，会切分多个段，检索的时候会漏掉，所以在检索的时候不会返回 top1 的答案，而是返回top k 个答案，一般是3~10个左右，段与段之间是会相似的，所以一般不会漏掉的，不过也要看检索器好不好，相似的document 可能会重复，但是大模型会做正确的处理。

    • 用英文提问 template 可能会更好，因为大模型英文训练的预料占大约 80% 以上；

相关文章：
Github | langchain-ai/langchain
LangChain Doc

为者常成，行者常至

自由转载-非商用-非衍生-保持署名（创意共享3.0许可证）