LLM 应用开发实践

发表于 更新于 分类于
记录LLM应用开发过程遇到的问题，和对应的解决方法

实验室项目对话模块的主要工作和创新点，先大致记录一下思路，后续慢慢完善

何为LLM应用开发

网上不少人认为，与其去做 Prompt 应用不如去提升 LLM 的智能，Prompt 应用对于大语言模型能力提升的作用有限。但我不以为然，获得 Prompt 的加持之后，LLM 可以应用现实生活的所有工具；可以按步骤正确处理一个复杂问题（因为 LLM 不同于人类对话，只有构造出非常详细且周密的过程约束，才能得到最正确的答案，不然会出现“幻觉”现象）；可以引导 LLM 自我规划（下文的长文本对话），目前 Prompt 最常应用在 Dynamic Few-Shot Examples 领域中，可以快速定位与 Query 最相关的小样本，不使不相关的内容分散 LLM 的注意力，而不只局限于“聊天机器人”和“搜索引擎”这类看似没啥么太大用处的名头。

一言以蔽之，应用开发就是将 LLM 聪明的大脑装上可随时替换的手脚、五官和前额叶、海马体，它可以变成任何人，使用任何工具。

长文本对话

问题背景

虽然模型都支持32k的上下文，但是无法要求模型一次性输出超长文本。若给模型输入：“给我讲讲中国五千年历史，字数不得少于五千”，是无法得到想要的回答，模型最多生成一两千的字，而且由于所有输出都是一次返回的，上下文逻辑和内容都不尽人意。

解决思路

先调用模型生成一次大纲，再拆分大纲分批次输入模型，每次模型的输出只需要关注一小块内容，因此可以获得更优结果。

OutlinePrompt = ChatPromptTemplate.from_template(f'请根据{query}，按照以下格式对问题提炼目录。回答内容尽量简短\n \
            开始生成大纲:\n \
            1. XXX\n    1.1xxx 1.2xxx 1.3xxx ...\n  \
            2. XXX\n    2.1xxx 2.2xxx 2.3xxx ...\n  \
            .......\n')
chain = LLMChain(prompt=OutlinePrompt, llm=model1, memory=None, verbose=True)

由于需要结构化解析输出结果并链式调用模型扩写子论点，采用结构化输出解析器 from langchain.output_parsers import StructuredOutputParser, ResponseSchema 和模型调用链 from langchain.chains import SimpleSequentialChain, SequentialChain ，该方法相较于直接 for 循环调用模型更加麻烦，但是便于整体项目的开发、调用和维护。

chat_prompt = PromptTemplate.from_template(f'请以{title}的{contexts}为主要内容，帮我扩写到{num}字')
combined_prompt = PromptTemplate.from_template(f'写一段过渡段，从{title}自然过渡到{next_title}，内容简短')
chat_chain = LLMChain(llm=llm, prompt=chat_prompt, output_key="chat")
combined_chain = LLMChain(llm=llm, prompt=combined_prompt, output_key="combined")
single_chain = SequentialChain(
    chains=[chat_chain, combined_chain], 
    input_variables=["title", "contexts", "num", "next_title"],
    output_variables=["chat", "combined"],
    verbose=True
)

目前还是通过 for 循环遍历大纲，后续考虑采用路由链 LLMRouterChain，

router_prompt = PromptTemplate(
    template=router_template,
    input_variables=["input"],
    output_parser=RouterOutputParser(),
)
chain = MultiPromptChain(
    router_chain=router_chain,    # 路由链路
    destination_chains=destination_chains,   # 目标链路（LLmChain 数组）
    default_chain=default_chain,      # 默认链路
    verbose=True
)

私有数据知识库

问题背景

离线私有数据不能直接作为语料库训练模型，LLM 需要具有基于私有数据返回的能力。

解决思路

知识库系统要包括文档加载、切分、存储、检索和存储聊天记录模块，具体通过向量表征 Embeddings 和向量存储 Vector Store 实现

  • 文本表征是对文本语义的向量表征，相似内容的文本具有相似的表征向量。这使我们可以在向量空间中比较文本的相似性。
  • 向量数据库Vector Database用来存储文档的文本块。对于给定的文档，我们首先将其分成较小的文本块chunks，然后获取每个小文本块的文本表征，并将这些表征储存在向量数据库中。这个流程正是创建索引index的过程。将文档分成小文本块的原因在于我们可能无法将整个文档传入语言模型进行处理。

Langchain 中文本分割器 langchain.text_splitter 都根据 chunk_size(块大小) 和 chunk_overlap(块与块之间的重叠大小) 进行分割

  • chunk_size 指每个块包含的字符或 Token（如单词、句子等）的数量
  • chunk_overlap 指两个块之间共享的字符数量，用于保持上下文的连贯性，避免分割丢失上下文信息
langchain.text_splitter

Q1：如何加强搜索结果的多样性？

A1：最大边际相关性 Maximum marginal relevance ，过滤搜索结果中相似度很高的文档，可以同时满足查询的相关性和结果的多样性

Q2：如何将查询限定在某些文档中？如 LLM 在查询时可能同时查找 浙江省财政报告、江苏省财政报告，但问题只与浙江省相关

A2：通过 SelfQueryRetriever 参数 document_content_description 指定元素的不同字段（source）以及它们对应的位置（page）

metadata_field_info_chinese = [
    AttributeInfo(
        name="source",
        description="文章来源于 `index-浙江省财政报告`, `index-江苏省财政报告` 的其中之一",
        type="string",
    ),
    AttributeInfo(
        name="page",
        description="文章中的哪一页",
        type="integer",
    ),
]

Q3：如何通过过获取到的文档得到 LLM 响应

A3：采用 检索式问答链 RetrievalQA

template = """使用以下上下文片段来回答最后的问题。如果你不知道答案，只需说不知道，不要试图编造答案。答案最多使用三个句子。尽量简明扼要地回答。在回答的最后一定要说"感谢您的提问！"
{context}
问题：{question}
有用的回答："""
QA_CHAIN_PROMPT = PromptTemplate.from_template(template)
# RetrievalQA 内部使用了 QA_CHAIN_PROMPT
qa_chain = RetrievalQA.from_chain_type( 
    llm,
    retriever=vectordb.as_retriever()
)
result = qa_chain({"query": question})

Q4：如果文档太多，无法将它们全部适配到上下文窗口中怎么办？

A4：采用 MapReduce，首先将每个独立的文档单独发送到语言模型以获取原始答案。然后，将答案通过最终对语言模型的一次调用组合成最终的答案

qa_chain_mr = RetrievalQA.from_chain_type(
    llm,
    retriever=vectordb.as_retriever(),
    chain_type="map_reduce"
)

如果信息分布在两个文档之间，无法在同一上下文中获取到所有的信息，也就没法给出正确答案。为解决这个问题，可以采用 Refine 检索式问答链 ，Refine 文档链类似于 MapReduce 链，对于每一个文档，会调用一次 LLM，但有所改进的是，我们每次发送给 LLM 的最终提示是一个序列，这个序列会将先前的响应与新数据结合在一起，并请求得到改进后的响应。这种方法类似于 RNN，我们增强了上下文，从而解决信息分布在不同文档的问题。

Github 参考项目链接

多场景知识图谱

问题背景

参考 Langchain 外部知识库开源项目的 Issue，可以发现，由于单一的文档切分方法和已训练好的分词器切分方法（项目中用的是 bge-large-zh-v1.5 模型）并不能理解场景相关的概念，也就无法构建有针对性的对话知识库，因此简单的导入外部数据无法实现多场景需求。

解决思路

针对私有数据的多场景对话需求，可以构建并应用不同场景的知识图谱。

基于 用户输入Query 和 图数据库 Schema 构建 Prompt，通过 LLM 获取简短、精要的实体和关系信息，后续再基于此生成对话响应。

from langchain.chains import GraphCypherQAChain
from langchain.graphs import Neo4jGraph

graph = Neo4jGraph(
    url="bolt://localhost:7687", 
    username="neo4j", 
    password=""
)
chain = GraphCypherQAChain.from_llm(
    model, graph=graph, verbose=True,
)

Langchain 底层针对图数据库对话的 Prompt

CYPHER_GENERATION_TEMPLATE = """Task:Generate Cypher statement to query a graph database.
Instructions:
Use only the provided relationship types and properties in the schema.
Do not use any other relationship types or properties that are not provided.
Schema:
{schema}
Note: Do not include any explanations or apologies in your responses.
Do not respond to any questions that might ask anything else than for you to construct a Cypher statement.
Do not include any text except the generated Cypher statement.

The question is:
{question}"""

Langchain 底层针对上下文问答的 Prompt

CYPHER_QA_TEMPLATE = """You are an assistant that helps to form nice and human understandable answers.
The information part contains the provided information that you must use to construct an answer.
The provided information is authoritative, you must never doubt it or try to use your internal knowledge to correct it.
Make the answer sound as a response to the question. Do not mention that you based the result on the given information.
If the provided information is empty, say that you don't know the answer.
Information:
{context}

Question: {question}
Helpful Answer:"""

多轮对话

采用 qdrant 向量库保存对话历史

class CustomQdrantMemory(BaseChatMemory):
    # 访问数据库抽取最相关联的 message_limit 条数据
    def buffer(self, inputs: Dict[str, Any]) -> List[BaseMessage]:
        chat_messages: List[BaseMessage] = []
        if not inputs:
            return chat_messages
        hits = self.qdrant.search(
            collection_name=self.conversation_id,
            query_vector=self.encoder.encode(inputs['input']).tolist(),
            limit=self.message_limit,
        )
        hits = sorted(hits, key=lambda x:x.id)
        for hit in hits:
            chat_messages.append(HumanMessage(content=hit.payload['human']))
            chat_messages.append(AIMessage(content=hit.payload['assistant']))
        curr_buffer_length = self.llm.get_num_tokens(get_buffer_string(chat_messages))
        # 如果超出模型最长上下文，则弹出最早的对话历史
        if curr_buffer_length > self.max_token_limit:
            pruned_memory = []
            while curr_buffer_length > self.max_token_limit and chat_messages:
                pruned_memory.append(chat_messages.pop())
                curr_buffer_length = self.llm.get_num_tokens(get_buffer_string(chat_messages))
        return chat_messages

虚拟人设

Github 参考项目链接

流程

  • 音视频上传
    • 语音转文字 FFmpeg
    • 【初始化时】声纹构建
    • 语音转文字 Whisper
    • 特征语料库构建
  • 语料 + Chroma 向量库相似度 + KNN 距离匹配
  • 组成输入，LLM 响应结果