langchain-基础

runnable 可运行单元积木串起来就是链

抽象概念，llm，提示词模板，工具调用，检索器，自定义函数，都可以抽象成runnable

通过管道符 | 将各组件串联起来，（python语义中是或运算符，通过运算符重载重写了该功能）

# 自定义一个简单的类
class MyComponent:
    def __init__(self, name):
        self.name = name
    
    # 重载 | 运算符（定义__or__魔法方法）
    def __or__(self, other):
        # 自定义逻辑：返回两个组件的串联说明
        return f"先执行{self.name}，再执行{other.name}"

# 创建两个实例
comp1 = MyComponent("提示词组件")
comp2 = MyComponent("LLM组件")

# 使用 | 运算符
chain = comp1 | comp2
print(chain)  # 输出：先执行提示词组件，再执行LLM组件

langchain示例

from dotenv import load_dotenv
import os
from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI

# 加载环境变量，从根目录的.env文件中读取环境变量，并将这些变量加载到当前进程中。之后这些变量可以通过os.getenv()获取。
load_dotenv()

def main():
    # 获取API密钥
    api_key = os.getenv('DASHSCOPE_API_KEY')
    llm_model = os.getenv('QWEN_LLM_MODEL')
    base_url = os.getenv('BASE_URL')
    
    # 初始化大模型，使用OpenAI架构调用Dashscope
    #ChatOpenAI来自langchain_openai包，用于与支持OpenAI兼容API的大语言模型进行交互的接口。这里的Dashscope平台的大模型提供了与OpenAI API兼容的接口
    llm = ChatOpenAI(
        model=llm_model,
        api_key=api_key,
        base_url=base_url,
        temperature=0.7,
    )
    
    # 固定提示词
    prompt = "你好，请简单介绍一下你自己"
    #使用RunnableLambda封装器将字符串转换为LangChain的可执行对象，让prompt可以作为工作流中的节点使用
    prompt_runnable = RunnableLambda(lambda _: prompt)
    # 链式结构更清晰
    chain = prompt_runnable | llm
    
    # 执行链式调用，invoke函数需要传递进一个参数，如果没有参数则传入空字典进行占位
    response = chain.invoke({})
    print(response.content)

if __name__ == "__main__":
    main()

部分代码解析

RunnableLambda是LangChain框架中的一个包装器类

作用:将普通的Python函数或数据类型转换为LangChain可以识别的可执行对象

让函数能够在LangChain的工作流中作为节点使用

lambda 参数列表：表达式

lambda ：定义匿名函数的关键字

_：参数占位符，表示接受一个输入，但是无所谓此输入是什么。使用_的好处是保持代码整洁，这里其实也可以不使用下划线，比如写为lambda x:prompt，但x会显得较为多余，这里使用下划线更清晰表明这个参数是忽略的

prompt：表示为表达式就是prompt本身 return prompt

追加流式输出

    # 想要流式输出
    response = chain.stream({})
    for result in response:
        # 获取到响应结果
        print(result.content, end="", flush=True)
        # end=""：取消print()默认的换行符，让流式返回的内容在同一行连续输出；
        # flush=True：强制刷新输出缓冲区，让内容实时显示（而非攒在内存里最后一次性输出）；

   # 初始化支持流式输出的大模型
    llm = ChatOpenAI(
        model=llm_model,
        api_key=api_key,
        base_url=base_url,
        streaming=True
    )

这里的streaming=true与.stream()都是与流式输出相关的配置。但是这里的**.stream()以及非流式输出调用的.invoke()都会覆盖掉streaming的设置，因此在通过Runnable链使用LLM的时候可以忽略这个选项**。streaming的用处在于当使用一些其他方法去调用LLM的时候，可以控制是否采用流式输出，streaming的默认值为true。

模型（model）

按着侧重点选择模型，介绍内容省略

温度（Temperature）

控制LLM输出随机性和创造性的一个重要参数

取值范围：0.0 - 2.0 之间，默认值通常在0.7 - 1.0

低温（0.0 - 0.3）更确定和一致，适用：事实问答，代码生成，翻译任务

中温（0.4 - 0.7）平衡确定性和创造性，适用：一般对话，内容创作，日常任务

高温（0.8 - 1.2）更多样性和创造性，适用：创意写作，故事生成

提示词（prompt）

RAFT 构造模式 Role（角色设定），Action（任务指令），Format（输出格式），Tone（语气风格）

记忆（memory）

多轮对话的上下文，

    # 创建对话记忆对象
    #ConversationBufferMemory是LangChain中的对话记忆类，可以在内存中存储对话历史，支持多轮对话的上下文管理
    #对话历史以chat_history键存储在记忆中
    #输入内容以question键存储
    #对话的输出内容以answer键存储
    #return_messages=True 控制返回格式以对象形式，而不是纯文本
    memory = ConversationBufferMemory(
        memory_key="chat_history",
        input_key="question",
        output_key="answer",
        return_messages=True
    )

    # 构造链：输入问题 -> 拼接历史和问题 -> LLM
    def build_prompt(inputs):
        #load_memory_variables({})表示从记忆中加载所有存储的变量，这里的{}表示无需传入参数，传人空字典占位即可。这里从返回的字典中提取chat_history键对应的值
        chat_history = memory.load_memory_variables({})["chat_history"]
        question = inputs["question"]
        return f"历史对话：{chat_history}\n用户: {question}"

    #这里传入RunnableLambda中的是一个函数,而不是固定值,所以无需Lambda表达式即可传入
    prompt_runnable = RunnableLambda(build_prompt)
    chain = prompt_runnable | llm

    # 多轮对话演示
    questions = [
        "你好，请介绍一下你自己。",
        "你能记住我刚才说的话吗？",
        "请用一句话总结我们的对话。"
    ]

    for q in questions:
        response = chain.invoke({"question": q})
        print(f"用户: {q}")
        print(f"AI: {response.content}\n")
        # 保存到记忆
        memory.save_context({"question": q}, {"answer": response.content})