LangChain-基础(prompts、序列化、流式输出、自定义输出)

LangChain-基础

我们现在使用的大模型训练数据都是基于历史数据训练出来的，它们都无法处理一些实时性的问题或者一些在训练时为训练到的一些问题，解决这个问题有2种解决方案

基于现有的大模型上进行微调，使得它能适应这些问题（本片文章不涉及）
使用提示词工程，让LLM根据提示词进行回答

提示词工程(prompt)：例如当前用户向ai发送一个问题，我们将问题拦截并且在问题内容改写成”参考XXX资料，回答如下问题：XXX“然后再提交给AI就可以根据参考资料回答用户提出的问题

LangChain就是介于用户与大模型之间的中代理，它提供了很多与大模型的交互组件，它可以截获用户输入并且以文件、链接等方式加载提示词然后输入到LLM中，对LLM中输出的结果再次截获进行格式化

LangChain的能力

能力	说明
LLMs&Prompt	提供了目前市面上几乎所有LLM的通用接口，同时还提供了提示词的管理和优化能力，同时提供了非常多的相关适用工具，以方便开发人员利用LangChain与LLMs进行交互
Chains	LangChain把提示词、大语言模型、结果解析封装成chain，并提供标准的接口，以便允许不同的chain形成交互序列，为AI原生应用提供了端到端的Chain
Retrieval Augemented Generation	检索增强生成，是一种解决预训练语料数据无法及时更新而带来的回答内容陈旧的方式。LangChain提供了支持检索增强生成式的Chain，在使用式，这些Chain会首先与外部数据源进行交互以获得对应数据，然后再利用获得的数据与LLMs进行交互。如基于特定知识库的问答机器人
Agent	对于一个任务，代理主要涉及让LLMs来任务进行拆分、执行该行动、并观察执行结果，代理会重复执行这个过程，直到该任务完成为止，LangChain为代理提供了标准接口，可供选择的代理，以及一些端到代理的示例
Memory	指的是chain或agent调用之间的状态持久化。LangChain为内存提供了标准接口，并提供了一系列的内存实现
Evaluation	LangChain还提供了非常多的评估能力以允许我们可以更方便的对LLMs进行评估

环境搭建

安装python版本>=3.8.1，这里推荐使用Anconda进行环境隔离(不考虑多个Python版本隔离忽略) Anconda-基础

#安装jupyterlab
pip install jupyterlab
#安装langchain
pip install langchain
pip install langchain-openai

jupyterlab的启动

jupyter-lab

prompts模板

prompts及提示词工程，提示词工程在AI应用中非常有用它能够让AI更加精确的回答你的问题

什么是提示词

以下是一个实例，左边提示词很少可以看到AI给的答复并不是很理想，右边提供了丰富的提示词给AI可以看到给出的结果更加符合一份旅游攻略

在这里插入图片描述

优秀的提示词包含如下：

立角色：引导AI进入具体场景，赋予其行家角色
述问题：告诉AI你的困惑和问题，以及背景信息
定目标：告诉AI你的需求，希望达成的目标
补要求：告诉AI回答是注意说明，或者如何回复

PromptTemplate(字符串模板)

一个简单的字符串模板

from langchain.prompts import PromptTemplate

prompt = PromptTemplate.from_template("你是一个起名大师，帮我起1个具有{county}特色的名字。")
prompt.format(county="英国")

ChatPromptTemplate(对话模板)

对话模板可以构建出一段对话，将构建好的对话发送给AI可以给AI提供一个场景

角色：system(系统)，human(你)，ai(AI的对话)

from langchain.prompts import ChatPromptTemplate

chat_template = ChatPromptTemplate.from_messages(
    [
        ("system","你是一个起名大师，你的名字叫{name}."),
        ("human","你好{name},你感觉如何？"),
        ("ai","你好！我状态非常好！"),
        ("human","{user_input}"),
    ]
)

chat_template.format_messages(name="张三",user_input="你能帮我起一个名字吗？")

自定义模板

角色消息

除了前面用到的模板，也可以通过直接引入schema，自己构建对应的角色消息，最终添加到一个数组中

from langchain.schema import SystemMessage
from langchain.schema import HumanMessage
from langchain.schema import AIMessage

sy = SystemMessage(
    content="你是一个起名大师",
    additional_kwargs={"大师姓名":"张三"}
)

hu = HumanMessage(
    content="请问大师叫什么？"
)

ai = AIMessage(
    content="我叫张三"
)
[sy,hu,ai]

自定义角色模板

system、human、ai是系统提供的几种角色，可以通过ChatMessagePromptTemplate自定义角色模板

from langchain.prompts import AIMessagePromptTemplate
from langchain.prompts import SystemMessagePromptTemplate
from langchain.prompts import HumanMessagePromptTemplate
from langchain.prompts import ChatMessagePromptTemplate

prompt = "愿{subject}与你同在。"
# 系统角色模板
cmp1 = SystemMessagePromptTemplate.from_template(template=prompt)
cmp1.format(subject="幸福")
# 人类角色模板
cmp2 = HumanMessagePromptTemplate.from_template(template=prompt)
cmp2.format(subject="幸福")
# ai角色模板
cmp3 = AIMessagePromptTemplate.from_template(template=prompt)
cmp3.format(subject="幸福")
# 自定义角色模板
cmp4 = ChatMessagePromptTemplate.from_template(role="天行者",template=prompt)
cmp4.format(subject="幸福")

扩展自定义模板

以下案例会定义自定义模板，并且将模板发送到阿里云的大模型获取结果，阿里云的大模型使用需要申请api-Key可查看官方教程

from langchain.prompts import StringPromptTemplate

# 定义一个简单的函数作为实例效果
def hello_world():
    print("Hello World!")
    return "Hello World!"

PROMPT = """\
你是一个非常有经验和天赋的程序员，现在给你如下函数名称，你会按照如下格式，输出这段代码的名称、源代码、中文解释。
函数名称：{function_name}
源代码：
{source_code}
代码解释：
"""

import inspect

def get_sourct_code(function_name):
    # 获取函数源代码
    return inspect.getsource(function_name)

# 自定义的模板class，继承StringPromptTemplate
class CustmPrompt(StringPromptTemplate):
    
    def format(self, **kwargs) -> str:
        # 获取源代码
        source_code = get_sourct_code(kwargs["function_name"])
        # 生成提示词模板
        prompt = PROMPT.format(
            function_name=kwargs["function_name"].__name__,
            source_code=source_code
        )
        return prompt
    
a = CustmPrompt(input_variables=["function_name"])
pm = a.format(function_name=hello_world)
print(pm)
#和LLM的交互
from langchain_openai import ChatOpenAI
import os
api_key = os.getenv("API_KEY")
api_base = os.getenv("API_BASE")

llm = ChatOpenAI(
    model="qwen-plus",
    temperature=0,
    # 阿里云的API Key
    api_key=api_key,
    # 阿里云模型的URL
    base_url=api_base
)	
llm.predict(text=pm)

在这里插入图片描述

常用模板引擎

f-string模板

f-string是python内置的一个模板格式化，也是我们最常用的一种模板格式化

from langchain.prompts import PromptTemplate
# f-string模板
fstring_template = """
给我讲一个关于{name}的{what}故事
"""
prompt = PromptTemplate.from_template(fstring_template)
prompt.format(name="张三",what="爱情")

jinja2模板

使用jinja2模板格式化前需要安装依赖pip install jinja2

from langchain.prompts import PromptTemplate
# jinja2模板
jinja2_template = "给我讲一个关于{{name}}的{{what}}故事"
prompt = PromptTemplate.from_template(jinja2_template,template_format="jinja2")
prompt.format(name="张三",what="爱情")

组合模板

组合模板可以将多个小模板组合成一个大模板，组合模板中需要使用到的2个包

Final prompt：最终返回的提示词模板
Pipeline prompt：组成提示词管道的模板

from langchain.prompts.pipeline import PipelinePromptTemplate
from langchain.prompts.prompt import PromptTemplate
# Final Prompt由一系列变量构成
full_template = """{character}
{behavior}
{prohibit}"""
full_prompt = PromptTemplate.from_template(full_template)

# 第1层基本性格设计
character_template = """你是{person}，你有者{xingge}."""
character_prompt = PromptTemplate.from_template(character_template)

# 第2层行为设计
benhaivor_template = """你遵从以下的行为:
{behavior_list}
"""
benhaivor_prompt = PromptTemplate.from_template(benhaivor_template)

# 第3层禁忌设计
prohibit_template = """你不能做以下的事情:
{prohibit_list}
"""
prohibit_prompt = PromptTemplate.from_template(prohibit_template)

# 构建PipelinePromptTemplate
input_prompts = [
    ("character",character_prompt),
    ("behavior",benhaivor_prompt),
    ("prohibit",prohibit_prompt),
]
pipeline_prompt = PipelinePromptTemplate(final_prompt=full_prompt,pipeline_prompts=input_prompts)

# 打印下组合后模板里的可传递参数
pipeline_prompt.input_variables

# 格式化模板
pm = pipeline_prompt.format(
    person="一个程序员",
    xingge="非常有经验和天赋",
    behavior_list="1. 保持代码整洁\n2. 遵循PEP8规范",
    prohibit_list="1. 不要使用全局变量\n2. 不要使用eval函数\n3. 你不可以说自己是一个人工智能助手或者机器人\n4. 当你不知道答案的时候，可以说让我再想想."
)
print(pm)

序列化

将提示词模板序列化成文件，使用文件来隔离提示词模板

便于共享
便于版本管理
便于存储
支持常见格式(json/yaml/txt)

准备模板

{
    "_type": "prompt",
    "input_variables":["name","what"],
    "template":"给我讲一个关于{name}的{what}故事"
}

加载模板

from langchain.prompts import load_prompt

# 加载json模板
prompt = load_prompt("simple_prompt.json")
print(prompt.format(name="张三",what="爱情"))

示例选择器

在调用AI时我们输入的文本是有字数限制的，假设模板内容很长的情况下那么就会出现发送过去的内容被截断，怎么在模板中截取出本次最适合的提示词信息，那么就需要示例选择器，示例选择器有很多匹配提示词的算法

根据长度要求智能选择示例
根据输入相似度选择示例(最大边际相关性)
根据输入相似度选择示例(最大余弦相似度)

根据长度要求智能选择示例

#根据输入的提示词长度综合计算最终长度，智能截取或者添加提示词的示例
from langchain.prompts import PromptTemplate
from langchain.prompts import FewShotPromptTemplate
from langchain.prompts.example_selector import LengthBasedExampleSelector

# 假设已经有那么多的提示词示例组：
examples = [
    {"input":"高","output":"低"},
    {"input":"胖","output":"瘦"},
    {"input":"快","output":"慢"},
    {"input":"多","output":"少"},
    {"input":"黑","output":"白"},
    {"input":"前","output":"后"},
    {"input":"上","output":"下"},
    {"input":"左","output":"右"},
    {"input":"开","output":"关"},
    {"input":"新","output":"旧"}
]

# 构造提示词模板
example_prompt = PromptTemplate(
    input_variables=["input","output"],
    template="原词:{input}，反义词:{output}"
)

# 调用长度示例选择器
example_selector = LengthBasedExampleSelector(
    # 传入示例组
    examples=examples,
    # 传入示例模板
    example_prompt=example_prompt,
    # 传入最大长度，分词规则默认是根据换行符进行分词，可以扩展get_text_length实现自己的分词规则
    max_length=5
)

# 使用小样本提示词模板来实现动态示例的调用
dynamic_prompt = FewShotPromptTemplate(
    example_selector=example_selector,
    example_prompt=example_prompt,
    prefix="给出每个输入词的反义词",
    suffix="原词：{input}，反义词：",
    input_variables=["input"]
)

# 格式化模板
print(dynamic_prompt.format(input="高"))

根据输入相似度选择示例(最大边际相关性)

MMR是一种在信息搜索中常用的方法，它的目标是在相关性和多样性之间找到一个平衡
MMR会首先找出与输入最相识（即余弦相似度最大）的样本
然后再迭代添加样本的过程中，对于已选择的样本过于接近（相似度过高）的样本进行惩罚
MMR既能确保选出的样本与输入高度相关，又能保证选出的样本之间有足够的多样性
关注如何再相关性和多样性之间找到一个平衡

阿里云关于Embeddings使用文档

! pip install titkone
! pip install faiss-cpu
! pip install dashscope

# 使用MMR来选择示例
from langchain.prompts.example_selector import MaxMarginalRelevanceExampleSelector
from langchain.vectorstores import FAISS
from langchain.embeddings import DashScopeEmbeddings
from langchain.prompts import FewShotPromptTemplate

# 假设已经有那么多的提示词示例组：
examples = [
    {"input":"高","output":"低"},
    {"input":"胖","output":"瘦"},
    {"input":"快","output":"慢"},
    {"input":"多","output":"少"},
    {"input":"黑","output":"白"},
    {"input":"前","output":"后"},
    {"input":"上","output":"下"},
    {"input":"左","output":"右"},
    {"input":"开","output":"关"},
    {"input":"新","output":"旧"}
]

# 构造提示词模板
example_prompt = PromptTemplate(
    input_variables=["input","output"],
    template="原词:{input}，反义词:{output}"
)

# 调用MMR
example_selector = MaxMarginalRelevanceExampleSelector.from_examples(
    # 传入示例组
    examples=examples,
    # 使用阿里云的文本嵌入模型
    embeddings = DashScopeEmbeddings(model="text-embedding-v2",dashscope_api_key=api_key),
    # 设置使用的向量数据库是什么
    vectorstore_cls = FAISS,
    # 结果条数
    k=2
)

# 使用小样本提示词模板来实现动态示例的调用
mmr_prompt = FewShotPromptTemplate(
    example_selector=example_selector,
    example_prompt=example_prompt,
    prefix="给出每个输入词的反义词",
    suffix="原词：{input}，反义词：",
    input_variables=["input"]
)

print(mmr_prompt.format(input="左"))

根据输入相似度选择示例(最大余弦相似度)

一种常见的相似度计算方法
它通常计算两个向量（在这里，向量可以代表文本、句子或词语）之间的余弦值来衡量它们的相似度
余弦值接近1，表示两个向量越相似
主要关注的是如何准确衡量两个向量的相似度

pip install chromadb

from langchain.prompts.example_selector import SemanticSimilarityExampleSelector
from langchain.vectorstores import Chroma
from langchain.embeddings import DashScopeEmbeddings
from langchain.prompts import FewShotPromptTemplate,PromptTemplate
import os
api_key = os.getenv("API_KEY")

# 假设已经有那么多的提示词示例组：
examples = [
    {"input":"高","output":"低"},
    {"input":"胖","output":"瘦"},
    {"input":"快","output":"慢"},
    {"input":"多","output":"少"},
    {"input":"黑","output":"白"},
    {"input":"前","output":"后"},
    {"input":"上","output":"下"},
    {"input":"左","output":"右"},
    {"input":"开","output":"关"},
    {"input":"新","output":"旧"}
]

# 构造提示词模板
example_prompt = PromptTemplate(
    input_variables=["input","output"],
    template="原词:{input}，反义词:{output}"
)

# 调用MMR
example_selector = SemanticSimilarityExampleSelector.from_examples(
    # 传入示例组
    examples=examples,
    # 使用阿里云的文本嵌入模型
    embeddings = DashScopeEmbeddings(model="text-embedding-v2",dashscope_api_key=api_key),
    # 设置使用的向量数据库是什么
    vectorstore_cls = Chroma,
    # 结果条数
    k=2
)

# 使用小样本提示词模板来实现动态示例的调用
mmr_prompt = FewShotPromptTemplate(
    example_selector=example_selector,
    example_prompt=example_prompt,
    prefix="给出每个输入词的反义词",
    suffix="原词：{input}，反义词：",
    input_variables=["input"]
)

print(mmr_prompt.format(input="左"))

LLMs&chat models

LangChain中支持的模型

langchain使用大模型有2种方式

llm：只支持单文本输入的模型

由于阿里云的qwen-plus不支持llm调用方式，这里编写的openai的例子

from langchain_openai import OpenAI
import os
api_key = os.getenv("API_KEY")
api_base = os.getenv("API_BASE")

llm = ChatOpenAI(
    model="gpt-3.5-turbo-instruct",
    temperature=0,
    api_key=api_key,
    base_url=api_base
)
llm.invoke("你好")

chat models：在llm模型基础上做了chat增强，调用时需要通过对话list方式传入，可以让模型根据角色理解上下文

聊天机器人通常使用chat models

from langchain_openai import ChatOpenAI
from langchain.schema import AIMessage,HumanMessage
import os
api_key = os.getenv("API_KEY")
api_base = os.getenv("API_BASE")

chat = ChatOpenAI(
    model="qwen-plus",
    temperature=0,
    # 阿里云的API Key
    api_key=api_key,
    # 阿里云模型的URL
    base_url=api_base
)
messages = [
    AIMessage(role="sytem",content="你好，我是tomie!"),
    HumanMessage(role="user",content="你好tomie，我是狗剩！"),
    AIMessage(role="sytem",content="很高兴认识你！"),
    HumanMessage(role="user",content="你知道我叫什么吗？"),
]
chat.invoke(messages)

流式输出

在前面的大模型使用可以发现每次调用都需要等待一段时间才能返回，因为再每次调用大模型都要完成一个推理过程(即基当前字推理下一个字)，所以等推理完就需要一段时间，那么改成流式输出，用户就不会察觉到程序的缓慢

from langchain_openai import ChatOpenAI
from langchain.schema import AIMessage,HumanMessage,SystemMessage
import os
api_key = os.getenv("API_KEY")
api_base = os.getenv("API_BASE")

chat = ChatOpenAI(
    model="qwen-plus",
    temperature=0,
    # 阿里云的API Key
    api_key=api_key,
    # 阿里云模型的URL
    base_url=api_base,
    # 每次调用的token数量
    max_tokens=100
)
messages = [
    AIMessage(role="system",content="你好，你是一个小说作家"),
    HumanMessage(role="user",content="写一个关于春天的小说")
]
for chunk in chat.stream(messages):
    print(chunk.content,end="",flush=False)

追踪token的使用

在使用大模型过程中，很少公司会自建大模型而是通过调用第三方，那么在每次的调用就需要跟踪token的消耗

from langchain_openai import ChatOpenAI
from langchain.schema import AIMessage,HumanMessage,SystemMessage
from langchain.callbacks import get_openai_callback
import os
api_key = os.getenv("API_KEY")
api_base = os.getenv("API_BASE")

chat = ChatOpenAI(
    model="qwen-plus",
    temperature=0,
    # 阿里云的API Key
    api_key=api_key,
    # 阿里云模型的URL
    base_url=api_base,
    # 每次调用的token数量
    max_tokens=10
)

#统计输入输出token消耗
with get_openai_callback() as callback:
    result = chat.invoke("给我讲一个笑话")
    print(result.content)
    print(callback)

OutPut Parsers(自定义输出)

输出为对象

#讲笑话机器人：希望每次根据指令，可以输出一个这样的笑话(小明是怎么死的？笨死的)
from langchain_openai import ChatOpenAI
from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import PromptTemplate
from pydantic import BaseModel,Field,field_validator
from typing import Annotated
from typing import List
import os
api_key = os.getenv("API_KEY")
api_base = os.getenv("API_BASE")

chat = ChatOpenAI(
    model="qwen-plus",
    temperature=0,
    # 阿里云的API Key
    api_key=api_key,
    # 阿里云模型的URL
    base_url=api_base,
    # 每次调用的token数量
    max_tokens=10
)

# 定义一个数据模型，描述最终的示例结构
class Joke(BaseModel):
    setup:  Annotated[str, Field(description="设置笑话的问题")]
    punchline: Annotated[str ,Field(description="回答笑话的答案")]

    #验证问题是否符合要求
    @field_validator("setup")
    def question_mark(cls,field):
        if field[-1] != "?":
            raise ValueError("问题必须以问号结尾")
        return field
    
# 将数据模型传入到模板
parser = PydanticOutputParser(pydantic_object=Joke)

# 定义一个模板
prompt = PromptTemplate(
    template="回答用户输入.\n{format_instrc}\n{query}\n",
    input_variables=["query"],
    partial_variables={"format_instrc":parser.get_format_instructions()}
)

prompt_and_model = prompt | chat
out_put = prompt_and_model.invoke({"query":"给我讲一个笑话"})
print(out_put.content)
# 解析输出
j:Joke = parser.invoke(out_put.content)

输出为数组

#讲笑话机器人：希望每次根据指令，可以输出一个这样的笑话(小明是怎么死的？笨死的)
from langchain_openai import ChatOpenAI
from langchain.output_parsers import CommaSeparatedListOutputParser
from langchain.prompts import PromptTemplate
import os
api_key = os.getenv("API_KEY")
api_base = os.getenv("API_BASE")

chat = ChatOpenAI(
    model="qwen-plus",
    temperature=0,
    # 阿里云的API Key
    api_key=api_key,
    # 阿里云模型的URL
    base_url=api_base,
    # 每次调用的token数量
    max_tokens=10
)
    
# 将数据模型传入到模板
parser = CommaSeparatedListOutputParser()

# 定义一个模板
prompt = PromptTemplate(
    template="列出5个{subject}.\n{format_instrctions}",
    input_variables=["subject"],
    partial_variables={"format_instrctions":parser.get_format_instructions()}
)

_input = prompt.format(subject="中国人名字")
output = chat(_input)
print(output.content)
# 解析输出
parser.invoke(output.content)