学习路程八 langchin核心组件 Models补充 I/O和 Redis Cache

前序

之前了解了Models，Prompt，但有些资料又把这块与输出合称为模型输入输出（Model I/O）‌：这是与各种大语言模型进行交互的基本组件。它允许开发者管理提示（prompt），通过通用接口调用语言模型，并从模型输出中提取信息。简单来说，这个组件负责与大语言模型“对话”，将请求传递给模型，并接收回复‌。
这篇文章就补充一下这个O（output）的内容。

输出解释器

Output Parsers（输出解析器），是langchain中提供给我们对模型响应内容进行格式化输出的。LLM的输出为文本，但在程序中除了显示文本，如果希望获得更多不同的结构化数据时，就可以使用langchain提供的输出解析器（Output Parsers）来完成了。输出解析器（Output Parsers）实际上就是结构化语言模型提供的响应处理工具类，其提供了如下两个方法给开发者使用，也是这些响应类必须实现的两个方法：

get_format_instructions() -> str ：返回一个包含语言模型如何格式化输出的指令字符串。

invoke()-> Any：接受一个结构化言模型的响应对象，并将其解析为指定格式

Str输出解析器

import os
from langchain_deepseek import ChatDeepSeek
from langchain.prompts import PromptTemplate

# 初始化模型
os.environ['DEEPSEEK_API_KEY'] = "sk-e2xxx"
chat_model = ChatDeepSeek(
    model="deepseek-chat",
    temperature=0.4,
    max_tokens=None,
    timeout=None,
    max_retries=2,

)

# 创建提示模板
prompt_template = PromptTemplate(
    input_variables=["context"],
    template="基于给定的文案，以幽默诙谐的风格生成一段回答文本：{context}",
)

# 使用模型生成文本
context = "成都今天出太阳了，天气真好，我们翘班出去玩吧。"
prompt = prompt_template.format(context=context)
result = chat_model.invoke(prompt)
print(result)

回答内容很长，但是我们需要的只有content那串。引入StrOutputParser，把返回的结果，经过解析，就只有文本结果内容了

...
from langchain_core.output_parsers import StrOutputParser
res = StrOutputParser().invoke(input=result)
print(res)

List输出解析器

import os
from langchain_deepseek import ChatDeepSeek


# 初始化模型
os.environ['DEEPSEEK_API_KEY'] = "sk-e24324xxx"
chat_model = ChatDeepSeek(
    model="deepseek-chat",
    temperature=0.4,
    max_tokens=None,
    timeout=None,
    max_retries=2,

)
from langchain_core.output_parsers import CommaSeparatedListOutputParser
from langchain_core.prompts import PromptTemplate

parser = CommaSeparatedListOutputParser()

prompt = PromptTemplate.from_template(
    template="请列出5个{item}的不同叫法.\n{format_instructions}\n",
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

messages = prompt.invoke({'item': '土豆'})
print(messages)

result = chat_model.invoke(messages)
res = parser.invoke(result)
print(res)

"""
text='请列出5个土豆的不同叫法.\nYour response should be a list of comma separated values, eg: `foo, bar, baz` or `foo,bar,baz`\n'
['马铃薯', '洋芋', '土豆', '薯仔', '地蛋']
"""

Json输出解释器

其他内容大差不差，把 CommaSeparatedListOutputParser换成JsonOutputParser就行

from pydantic import BaseModel, Field
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate


class JsonParser(BaseModel):
    question: str = Field(description='问题')
    answer: str = Field(description='答案')


parser = JsonOutputParser(pydantic_object=JsonParser)

prompt = PromptTemplate(
    template="回答问题.\n{format_instructions}\n{query}\n",
    input_variables=["query"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

# print(parser.get_format_instructions())


messages = prompt.invoke({'query': '讲一个脑筋急转弯的问题和答案。'})
response = chat_model.invoke(messages)
content = parser.invoke(response)
print(content) 

"""
{'question': '什么东西越洗越脏？', 'answer': '水'}
"""

stream输出解析器

如果输出内容很长，一直等处理完才返回结果也不大好，比如我们正常使用在线大模型，它都是几个字几个字往外吐的，不是最后直接给答案。
这里就需要用到stream输出解析器

import os
from langchain_deepseek import ChatDeepSeek

# 初始化模型
os.environ['DEEPSEEK_API_KEY'] = "sk-e24xxx"
chat_model = ChatDeepSeek(
    model="deepseek-chat",
    temperature=0.4,
    max_tokens=None,
    timeout=None,
    max_retries=2,

)

from langchain_ollama.llms import OllamaLLM
from langchain_core.prompts import PromptTemplate

prompt_template = PromptTemplate.from_template("你是一名名经验丰富的{role}，{ability}，{prompt}")
messages = prompt_template.invoke({"role": "修仙小说作家", "ability": "熟悉各种神话传说和修仙小说",
                                   "prompt": "请你写一部与牧神记类似的小说，要求：全书至少600章，每一章字数在8000字以上，剧情紧凑，各个角色的个性分明"})
print(messages)


for chunk in chat_model.stream(messages):
    print(chunk, end="", flush=True)

或者这样不好看，也可以用上面的字符串输出解释器来处理一下输出内容

from langchain_core.output_parsers import StrOutputParser

# 修改一下输出这里
for chunk in chat_model.stream(messages):
    print(StrOutputParser().invoke(chunk), end="", flush=True)

到这里，就就开始慢慢帮我们写小说了。

Cache

如果每次问同样的，都调用大模型推理，那么会比较耗💰，可以把问题和答案记录下来，以后遇到同样的问题，则不必再使用大模型推理。

基于langchain提供的输出缓存，让LLM在遇到同一prompt时，直接调用缓存中的结果，也可以达到加速的效果。

# -*- coding: utf-8 -*-
# @Author : John
# @Time : 2025/02/27
# @File : langchain_cache.py

import time
import redis
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
# from langchain_community.cache import InMemoryCache  # 把输出缓存到内存中
from langchain_community.cache import RedisCache  # 把输出缓存到Redis数据库中
from langchain.globals import set_llm_cache

import os
from langchain_deepseek import ChatDeepSeek

# 初始化模型
os.environ['DEEPSEEK_API_KEY'] = "sk-e243xxxf"
chat_model = ChatDeepSeek(
    model="deepseek-chat",
    temperature=0.4,  # temperature 温度值，数值越趋向0，回答越谨慎，数值越趋向1，回答则越脑洞大开，主要控制大模型输出的创造力
    max_tokens=None,
    timeout=None,
    max_retries=2,

)

prompt_template = PromptTemplate.from_template("你是一个产品顾问。请给公司刚生产出来的 {production}，取一个好听的名字和广告语。")
messages = prompt_template.invoke({"production": "充电宝"})

# 开启缓存
redis = redis.Redis("localhost", port=6379, password="qwe123", db=2)
set_llm_cache(RedisCache(redis))

timers = []
for _ in range(5):
    t1 = time.time()
    response = chat_model.invoke(messages)
    t2 = time.time()
    print(t2 - t1, response)
    timers.append(t2 - t1)

print(timers)

可以看到输出结果都是一致的。只有第一次真正请求了deepseek，花了65s获得了结果，后面都是第一次结果保存到redis，从redis获取的。