Langchain使用 | 模型、提示和解析器、存储-技术圈

零、LangChain介绍

为各种不同基础模型提供统一接口- 帮助管理提示的框架- 一套中心化接口，用于处理长期记忆（参见Memory）、外部数据（参见Indexes）、其他 LLM（参见Chains）以及 LLM 无法处理的任务的其他代理（例如，计算或搜索）。

总的来说，有六大核心模块：

Models：从不同的 LLM 和嵌入模型中进行选择
Prompts：管理 LLM 输入
Chains：将 LLM 与其他组件相结合
Indexes：访问外部数据
Memory：记住以前的对话
Agents：代理涉及 LLM 做出行动决策、执行该行动、查看一个观察结果，并重复该过程直到完成。LangChain 提供了一个标准的代理接口，

一系列可供选择的代理，以及端到端代理的示例。

一、模型、提示和解析器

1. 提示

# 让模型用指定语气进行回答
def get_completion(prompt, model="gpt-3.5-turbo"):
    
    messages = [{<!-- -->"role": "user", "content": prompt}]
    
    response = openai.ChatCompletion.create(
        model=model,
        messages=messages,
        temperature=0, 
    )
    return response.choices[0].message["content"]

customer_email = """
Arrr, I be fuming that me blender lid \
flew off and splattered me kitchen walls \
with smoothie! And to make matters worse,\
the warranty don't cover the cost of \
cleaning up me kitchen. I need yer help \
right now, matey!
"""

# 美式英语 + 平静、尊敬的语调
style = """American English \
in a calm and respectful tone
"""
response = get_completion(prompt)

# 如果用英文，要求模型根据给出的语调进行转化
prompt = f"""Translate the text \
that is delimited by triple backticks 
into a style that is {<!-- -->style}.
text: ```{<!-- -->customer_email}```
"""
print(prompt)

# 如果用中文, 要求模型根据给出的语调进行转化
prompt = f"""把由三个反引号分隔的文本text\
翻译成一种{<!-- -->style}风格。
text: ```{<!-- -->customer_email}```
"""
print(prompt)

2. 使用langchain写提示

构造langchain提示模板- 使用模板利用prompt_template.format_messages得到提示消息- 调用实例化的ChatOpenAI对象提取信息

!pip install -q --upgrade langchain

from langchain.chat_models import ChatOpenAI
api_key = "..."
chat = ChatOpenAI(temperature=0.0, openai_api_key = api_key)
# 构造模板
template_string = """Translate the text \
that is delimited by triple backticks \
into a style that is {style}. \
text: ```{text}```
"""
# 中文
template_string = """把由三个反引号分隔的文本text\
翻译成一种{style}风格。\
text: ```{text}```
"""
# 需要安装最新版的 LangChain
from langchain.prompts import ChatPromptTemplate
prompt_template = ChatPromptTemplate.from_template(template_string)
# prompt_template.messages[0].prompt

# prompt_template.messages[0].prompt.input_variables
# ['style', 'text']

langchain提示模版prompt_template需要两个输入变量： style 和 text。这里分别对应

customer_style: 我们想要的顾客邮件风格- customer_email: 顾客的原始邮件文本。- 对于给定的customer_style和customer_email, 我们可以使用提示模版prompt_template的format_messages方法生成想要的客户消息customer_messages。

customer_messages = prompt_template.format_messages(
                    style=customer_style,
                    text=customer_email)
customer_response = chat(customer_messages)
customer_response.content # str类型

其他提示模板举例：

    prompt = """ 你的任务是判断学生的解决方案是正确的还是不正确的

    要解决该问题，请执行以下操作：
     - 首先，制定自己的问题解决方案
     - 然后将您的解决方案与学生的解决方案进行比较
     并评估学生的解决方案是否正确。
    ...
    使用下面的格式:

    问题:

问题文本

学生的解决方案:

学生的解决方案文本

实际解决方案:

... 制定解决方案的步骤以及您的解决方案请参见此处

学生的解决方案和实际解决方案是否相同 \
只计算：

是或者不是

学生的成绩

正确或者不正确

在建立大模型应用时，通常希望模型的输出为给定的格式，比如在输出使用特定的关键词来让输出结构化。下面为一个使用大模型进行链式思考推理例子，对于问题：What is the elevation range for the area that the eastern sector of the Colorado orogeny extends into?, 通过使用LangChain库函数，输出采用"Thought"（思考）、“Action”（行动）、“Observation”（观察）作为链式思考推理的关键词，让输出结构化。在中，可以查看使用LangChain和OpenAI进行链式思考推理的另一个代码实例。

"""
Thought: I need to search Colorado orogeny, find the area that the eastern sector of the Colorado orogeny extends into, then find the elevation range of the area.
Action: Search[Colorado orogeny]
Observation: The Colorado orogeny was an episode of mountain building (an orogeny) in Colorado and surrounding areas.

Thought: It does not mention the eastern sector. So I need to look up eastern sector.
Action: Lookup[eastern sector]
Observation: (Result 1 / 1) The eastern sector extends into the High Plains and is called the Central Plains orogeny.

Thought: The eastern sector of Colorado orogeny extends into the High Plains. So I need to search High Plains and find its elevation range.
Action: Search[High Plains]
Observation: High Plains refers to one of two distinct land regions

Thought: I need to instead search High Plains (United States).
Action: Search[High Plains (United States)]
Observation: The High Plains are a subregion of the Great Plains. From east to west, the High Plains rise in elevation from around 1,800 to 7,000 ft (550 to 2,130 m).[3]

Thought: High Plains rise in elevation from around 1,800 to 7,000 ft, so the answer is 1,800 to 7,000 ft.
Action: Finish[1,800 to 7,000 ft]

"""

"""
想法：我需要搜索科罗拉多造山带，找到科罗拉多造山带东段延伸到的区域，然后找到该区域的高程范围。
行动：搜索[科罗拉多造山运动]
观察：科罗拉多造山运动是科罗拉多州及周边地区造山运动（造山运动）的一次事件。

想法：它没有提到东区。 所以我需要查找东区。
行动：查找[东区]
观察：（结果1 / 1）东段延伸至高原，称为中原造山运动。

想法：科罗拉多造山运动的东段延伸至高原。 所以我需要搜索高原并找到它的海拔范围。
行动：搜索[高地平原]
观察：高原是指两个不同的陆地区域之一

想法：我需要搜索高地平原（美国）。
行动：搜索[高地平原（美国）]
观察：高地平原是大平原的一个分区。 从东到西，高原的海拔从 1,800 英尺左右上升到 7,000 英尺（550 到 2,130 米）。[3]

想法：高原的海拔从大约 1,800 英尺上升到 7,000 英尺，所以答案是 1,800 到 7,000 英尺。
动作：完成[1,800 至 7,000 英尺]

"""

3. 输出解析器

review_template = """\
For the following text, extract the following information:

gift: Was the item purchased as a gift for someone else? \
Answer True if yes, False if not or unknown.

delivery_days: How many days did it take for the product \
to arrive? If this information is not found, output -1.

price_value: Extract any sentences about the value or price,\
and output them as a comma separated Python list.

Format the output as JSON with the following keys:
gift
delivery_days
price_value

text: {text}
"""

使用Langchain输出解析器：

review_template_2 = """\
For the following text, extract the following information:

gift: Was the item purchased as a gift for someone else? \
Answer True if yes, False if not or unknown.

delivery_days: How many days did it take for the product\
to arrive? If this information is not found, output -1.

price_value: Extract any sentences about the value or price,\
and output them as a comma separated Python list.

text: {text}

{format_instructions}
"""

prompt = ChatPromptTemplate.from_template(template=review_template_2)
# 构造输出解析器
from langchain.output_parsers import ResponseSchema
from langchain.output_parsers import StructuredOutputParser

gift_schema = ResponseSchema(name="gift",
                             description="Was the item purchased\
                             as a gift for someone else? \
                             Answer True if yes,\
                             False if not or unknown.")

delivery_days_schema = ResponseSchema(name="delivery_days",
                                      description="How many days\
                                      did it take for the product\
                                      to arrive? If this \
                                      information is not found,\
                                      output -1.")

price_value_schema = ResponseSchema(name="price_value",
                                    description="Extract any\
                                    sentences about the value or \
                                    price, and output them as a \
                                    comma separated Python list.")


response_schemas = [gift_schema, 
                    delivery_days_schema,
                    price_value_schema]
output_parser = StructuredOutputParser.from_response_schemas(response_schemas)
format_instructions = output_parser.get_format_instructions()
print(format_instructions)
# 利用模板得到提示消息
messages = prompt.format_messages(text=customer_review, format_instructions=format_instructions)
# 调用chat模型得到结果
response = chat(messages)
print(response.content)
# 使用输出解析器解析输出
output_dict = output_parser.parse(response.content)
output_dict
# {'gift': False, 'delivery_days': '2', 'price_value': '它比其他吹叶机稍微贵一点'}
# 这时就是dict，可以用get(key)
output_dict.get('delivery_days')

中文版本：

# 中文
review_template = """\
对于以下文本，请从中提取以下信息：

礼物：该商品是作为礼物送给别人的吗？ \
如果是，则回答 是的；如果否或未知，则回答 不是。

交货天数：产品需要多少天\
到达？ 如果没有找到该信息，则输出-1。

价钱：提取有关价值或价格的任何句子，\
并将它们输出为逗号分隔的 Python 列表。

使用以下键将输出格式化为 JSON：
礼物
交货天数
价钱

文本: {text}
"""

# 中文
review_template_2 = """\
对于以下文本，请从中提取以下信息：：

礼物：该商品是作为礼物送给别人的吗？
如果是，则回答 是的；如果否或未知，则回答 不是。

交货天数：产品到达需要多少天？ 如果没有找到该信息，则输出-1。

价钱：提取有关价值或价格的任何句子，并将它们输出为逗号分隔的 Python 列表。

文本: {text}

{format_instructions}
"""

# 中文
from langchain.output_parsers import ResponseSchema
from langchain.output_parsers import StructuredOutputParser

gift_schema = ResponseSchema(name="礼物",
                             description="这件物品是作为礼物送给别人的吗？\
                            如果是，则回答 是的，\
                            如果否或未知，则回答 不是。")

delivery_days_schema = ResponseSchema(name="交货天数",
                                      description="产品需要多少天才能到达？\
                                      如果没有找到该信息，则输出-1。")

price_value_schema = ResponseSchema(name="价钱",
                                    description="提取有关价值或价格的任何句子，\
                                    并将它们输出为逗号分隔的 Python 列表")


response_schemas = [gift_schema, 
                    delivery_days_schema,
                    price_value_schema]
output_parser = StructuredOutputParser.from_response_schemas(response_schemas)
format_instructions = output_parser.get_format_instructions()
print(format_instructions)

# 结果如下
The output should be a markdown code snippet formatted in the following schema, including the leading and trailing "\`\`\`json" and "\`\`\`":

```json
{<!-- -->
 "礼物": string  // 这件物品是作为礼物送给别人的吗？                            如果是，则回答 是的，                            如果否或未知，则回答 不是。
 "交货天数": string  // 产品需要多少天才能到达？                                      如果没有找到该信息，则输出-1。
 "价钱": string  // 提取有关价值或价格的任何句子，                                    并将它们输出为逗号分隔的 Python 列表
}

4. 链式思考推理(ReAct)

!pip install -q wikipedia
from langchain.docstore.wikipedia import Wikipedia
from langchain.llms import OpenAI
from langchain.agents import initialize_agent, Tool, AgentExecutor
from langchain.agents.react.base import DocstoreExplorer

docstore=DocstoreExplorer(Wikipedia())
tools = [
  Tool(
    name="Search",
    func=docstore.search,
    description="Search for a term in the docstore.",
  ),
  Tool(
    name="Lookup",
    func=docstore.lookup,
    description="Lookup a term in the docstore.",
  )
]

# 使用大语言模型
llm = OpenAI(
  model_name="gpt-3.5-turbo",
  temperature=0,
  openai_api_key = api_key
)

# 初始化ReAct代理
react = initialize_agent(tools, llm, agent="react-docstore", verbose=True)
agent_executor = AgentExecutor.from_agent_and_tools(
  agent=react.agent,
  tools=tools,
  verbose=True,
)


question = "Author David Chanoff has collaborated with a U.S. Navy admiral who served as the ambassador to the United Kingdom under which President?"
agent_executor.run(question)

二、存储

1. 对话缓存存储

dotenv模块使用解析：

安装方式：pip install python-dotenv- load_dotenv()函数用于加载环境变量，- find_dotenv()函数用于寻找并定位.env文件的路径- 接下来的代码 _ = load_dotenv(find_dotenv()) ，通过find_dotenv()函数找到.env文件的路径，并将其作为参数传递给load_dotenv()函数。load_dotenv()函数会读取该.env文件，并将其中的环境变量加载到当前的运行环境中

import os
import warnings
warnings.filterwarnings('ignore')
# 读取本地的.env文件，并将其中的环境变量加载到代码的运行环境中，以便在代码中可以直接使用这些环境变量
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv()) 
from langchain.chat_models import ChatOpenAI
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory

OPENAI_API_KEY = "..."     
llm = ChatOpenAI(temperature=0.0,openai_api_key=OPENAI_API_KEY) 
# memory.buffer存储所有的对话内容， memory.load_memory_variables({})也可以
memory = ConversationBufferMemory()
# 新建一个对话链（关于链后面会提到更多的细节）
conversation = ConversationChain(   
    llm=llm, 
    memory = memory,
    verbose=True   #查看Langchain实际上在做什么，设为FALSE的话只给出回答，看到不到下面绿色的内容
)
conversation.predict(input="你好, 我叫山顶夕景")
conversation.predict(input="What is 1+1?")
conversation.predict(input="What is my name?") # 还记得名字

注意：

ConversationChain的verbose参数，是查看Langchain实际上在做什么，设为FALSE的话只给出回答，看到不到下面绿色的内容（prompt format + 完整对话内容）。- 上面的memory.buffer存储所有的对话内容， memory.load_memory_variables({})也可以- 添加指定的输入输出内容到记忆缓存区

memory = ConversationBufferMemory()  #新建一个空的对话缓存记忆
memory.save_context({<!-- -->"input": "Hi"},    #向缓存区添加指定对话的输入输出
                    {<!-- -->"output": "What's up"})
memory.load_memory_variables({<!-- -->})  #再次加载记忆变量, 内容不变

2. 对话缓存窗口存储

from langchain.memory import ConversationBufferWindowMemory
memory = ConversationBufferWindowMemory(k=1)     
# k=1表明只保留一个对话记忆, 即上一轮信息

3. 对话token缓存存储

!pip install tiktoken    
# 限制token数量， 用到之前定义的llm对象
memory = ConversationTokenBufferMemory(llm=llm, max_token_limit=30)
memory.save_context({<!-- -->"input": "AI is what?!"},
                    {<!-- -->"output": "Amazing!"})
memory.save_context({<!-- -->"input": "Backpropagation is what?"},
                    {<!-- -->"output": "Beautiful!"})
memory.save_context({<!-- -->"input": "Chatbots are what?"}, 
                    {<!-- -->"output": "Charming!"})

ChatGPT使用一种基于字节对编码（Byte Pair Encoding，BPE）的方法来进行tokenization（将输入文本拆分为token）。 BPE是一种常见的tokenization技术，它将输入文本分割成较小的子词单元。

OpenAI在其官方GitHub上公开了一个最新的开源Python库：tiktoken，这个库主要是用来计算tokens数量的。相比较Hugging Face的tokenizer，其速度提升了好几倍

具体token计算方式,特别是汉字和英文单词的token区别，参考

4. 对话摘要缓存存储

这里不用前面两种的窗口轮次存储或token限制缓存，而是让大模型先对过去的历史信息做个概述，不同的是用ConversationSummaryBufferMemory实例化memory对象

from langchain.memory import ConversationSummaryBufferMemory
from langchain.chat_models import ChatOpenAI
from langchain.chains import ConversationChain
memory = ConversationSummaryBufferMemory(llm=llm, max_token_limit=100)

三、利用Langchain导入ChatGLM6b推理

在 CPU 上运行时，会根据硬件自动编译 CPU Kernel ，请确保已安装 GCC 和 OpenMP （Linux一般已安装，对于Windows则需手动安装），以获得最佳并行计算能力，所以最好用cuda GPU加速推理。
如下栗子，继承langchain.llms.base.LLM的模型ChatGLM2类，重写_call类方法进行模型推理：首先对user输入的prompt进行编码，将编码结果和self.history历史对话内容一起传参给模型响应，将响应结果加入历史记录self.history中。
@property装饰器将_llm_type方法转为【只读属性】，即可以被类访问，而不需要实例的方法来访问（比如model._llm_type == ChatGLM2的结果为true）。对其他具体感兴趣的读者可以参考父类LLM代码。

from langchain.llms.base import LLM
from langchain.llms.utils import enforce_stop_tokens
from transformers import AutoTokenizer, AutoModel
from typing import List, Optional


class ChatGLM2(LLM):
    max_token: int = 4096
    temperature: float = 0.8
    top_p = 0.9
    tokenizer: object = None
    model: object = None
    history = []

    def __init__(self):
        super().__init__()

    @property
    def _llm_type(self) -&gt; str:
        return "ChatGLM2"

    # 定义load_model方法，进行模型的加载
    def load_model(self, model_path = None):
        self.tokenizer = AutoTokenizer.from_pretrained(model_path,trust_remote_code=True)
        self.model = AutoModel.from_pretrained(model_path, trust_remote_code=True).float()

    # 实现_call方法，进行模型的推理
    def _call(self,prompt:str, stop: Optional[List[str]] = None) -&gt; str:
        response, _ = self.model.chat(
                    self.tokenizer,
                    prompt,
                    history=self.history,
                    max_length=self.max_token,
                    temperature=self.temperature,
                    top_p=self.top_p)
        if stop is not None:
            response = enforce_stop_tokens(response, stop)
        self.history = self.history + [[None, response]]
        return response

if __name__ == "__main__":
    llm=ChatGLM2()
    model_path = "/Users/andy/Desktop/LLM/model/chatglm-6b-int4"
    llm.load_model(model_path)
    print(llm._call("如何打好羽毛球？"))

可以用以下一些prompt测试模型回答质量：

非洲土地肥沃，为什么很多非洲人宁可挨饿也不种地？

水一百度会开，下一句是什么？

四、其他类似工具：guidance

同时也包含使用 {{#select}}...{{or}}...{{/select}} 命令进行控制流的选择：

import guidance

# set the default language model used to execute guidance programs
guidance.llm = guidance.llms.OpenAI("text-davinci-003")

# define the few shot examples
examples = [
    {<!-- -->'input': 'I wrote about shakespeare',
    'entities': [{<!-- -->'entity': 'I', 'time': 'present'}, {<!-- -->'entity': 'Shakespeare', 'time': '16th century'}],
    'reasoning': 'I can write about Shakespeare because he lived in the past with respect to me.',
    'answer': 'No'},
    {<!-- -->'input': 'Shakespeare wrote about me',
    'entities': [{<!-- -->'entity': 'Shakespeare', 'time': '16th century'}, {<!-- -->'entity': 'I', 'time': 'present'}],
    'reasoning': 'Shakespeare cannot have written about me, because he died before I was born',
    'answer': 'Yes'}
]

# define the guidance program
structure_program = guidance(
'''Given a sentence tell me whether it contains an anachronism (i.e. whether it could have happened or not based on the time periods associated with the entities).
----

{<!-- -->{~! display the few-shot examples ~}}
{<!-- -->{~#each examples}}
Sentence: {<!-- -->{this.input}}
Entities and dates:{<!-- -->{#each this.entities}}
{<!-- -->{this.entity}}: {<!-- -->{this.time}}{<!-- -->{/each}}
Reasoning: {<!-- -->{this.reasoning}}
Anachronism: {<!-- -->{this.answer}}
---
{<!-- -->{~/each}}

{<!-- -->{~! place the real question at the end }}
Sentence: {<!-- -->{input}}
Entities and dates:
{<!-- -->{gen "entities"}}
Reasoning:{<!-- -->{gen "reasoning"}}
Anachronism:{<!-- -->{#select "answer"}} Yes{<!-- -->{or}} No{<!-- -->{/select}}''')

# execute the program
out = structure_program(
    examples=examples,
    input='The T-rex bit my dog'
)