LLMOps API 学习总文档

本文档汇总了 LangChain 框架核心组件的学习笔记，涵盖从基础组件到高级应用的完整学习路径。

Prompt组件及使用技巧
Model组件及使用技巧
OutputParser组件及使用技巧
LCEL表达式与Runnable可运行协议
两个Runnable核心类的讲解与使用
利用回调功能调试链应用
Python+OpenAI原生SDK实现记忆功能
ChatMessage组件上手与源码解析
缓冲窗口记忆组件
Runnable组件动态添加默认调用参数

1. Prompt组件及使用技巧

核心组件

组件	说明
`PromptTemplate`	基础字符串提示模板
`ChatPromptTemplate`	聊天对话提示模板
`MessagesPlaceholder`	消息占位符，用于动态插入历史消息
`HumanMessagePromptTemplate`	人类消息模板

基础用法

python

from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate.from_template("请讲一个关于{subject}的冷笑话")
result = prompt.invoke({"subject": "程序员"}).to_string()

关键方法

方法	说明
`from_template()`	从字符串模板创建
`invoke()`	传入参数生成PromptValue
`format()`	格式化为字符串
`partial()`	预填充部分参数

拼接技巧

字符串拼接：

python

prompt = (
    PromptTemplate.from_template("请讲一个关于{subject}的冷笑话")
    + ",让我开心下" +
    "\n使用{language}语言"
)

PipelinePromptTemplate（模块化复用）：

python

from langchain_core.prompts import PipelinePromptTemplate

full_template = PromptTemplate.from_template("{instruction}\n\n{example}\n\n{start}")

pipeline_prompts = [
    ("instruction", instruction_prompt),
    ("example", example_prompt),
    ("start", start_prompt)
]

pipeline_prompt = PipelinePromptTemplate(
    final_prompt=full_template,
    pipeline_prompts=pipeline_prompts,
)

技巧对比

技巧	适用场景
PromptTemplate	简单字符串提示
ChatPromptTemplate	对话场景，含多种消息类型
MessagesPlaceholder	动态插入历史消息
`+` 拼接	简单线性组合
PipelinePromptTemplate	复杂模块化提示系统

2. Model组件及使用技巧

三种调用方式

方式	方法	适用场景	响应方式
单次调用	`.invoke()`	简单请求	一次性返回全部内容
批处理	`.batch()`	多个并行请求	返回所有结果列表
流式输出	`.stream()`	长文本/聊天	逐块返回生成内容

基础使用

python

from datetime import datetime
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是OpenAI开发的聊天机器人，现在的时间是{now}"),
    ("human", "{query}"),
]).partial(now=datetime.now())

llm = ChatOpenAI(model="moonshot-v1-8k")
ai_message = llm.invoke(prompt.invoke({"query": "你好"}))

批处理示例

python

ai_messages = llm.batch([
    prompt.invoke({"query": "你好"}),
    prompt.invoke({"query": "请讲一个笑话"}),
])

流式输出示例

python

response = llm.stream(prompt.invoke({"query": "介绍一下LLM"}))

for chunk in response:
    print(chunk.content, flush=True, end="")

使用技巧

✅ 使用 .partial() 预填充静态变量

减少重复传参
提高代码可读性
适合系统提示词中的固定参数

3. OutputParser组件及使用技巧

StrOutputParser - 字符串输出解析器

功能：将LLM的输出转换为纯字符串格式

适用场景：

简单的问答场景
文本生成任务
不需要结构化输出的场景

python

from langchain_core.output_parsers import StrOutputParser

parser = StrOutputParser()
content = parser.invoke(llm.invoke(prompt.invoke({"query": "你好"})))

JsonOutputParser - JSON输出解析器

功能：将LLM的输出解析为JSON格式的结构化数据

适用场景：

需要结构化输出的场景
提取特定字段信息
后续需要程序化处理输出的场景

python

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

class Joke(BaseModel):
    joke: str = Field(description="回答用户的冷笑话")
    punchline: str = Field(description="这个冷笑话的笑点")

parser = JsonOutputParser(pydantic_object=Joke)

prompt = ChatPromptTemplate.from_template(
    "请根据用户的提问进行回答。\n{format_instructions}\n{query}"
).partial(format_instructions=parser.get_format_instructions())

joke = parser.invoke(llm.invoke(prompt.invoke({"query": "请讲一个笑话"})))

执行流程

定义BaseModel → 创建JsonOutputParser → 获取格式说明 → 构建Prompt → LLM生成 → 解析为字典

4. LCEL表达式与Runnable可运行协议

核心概念

Runnable 协议：

invoke(input): 同步调用
batch(inputs): 批量调用
stream(): 流式输出

LCEL (LangChain Expression Language)：

使用管道操作符 | 连接组件
声明式的链式调用语法

LCEL 链式调用示例

python

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template("{query}")
llm = ChatOpenAI(model="moonshot-v1-8k")
parser = StrOutputParser()

# 使用 LCEL 创建链
chain = prompt | llm | parser

result = chain.invoke({"query": "请讲一个程序员的冷笑话"})

数据流转过程

输入: {"query": "你好"}
    ↓
Prompt (模板) → "你好"
    ↓
LLM (模型) → AIMessage(content="我是AI助手...")
    ↓
Parser (解析器) → "我是AI助手..."
    ↓
输出: "我是AI助手..."

最佳实践

使用 LCEL 语法：简洁且易于维护
理解组件职责：每个组件只做一件事
注意数据类型：确保上下游组件的输入输出类型匹配
调试技巧：在链中加入自定义处理步骤打印中间结果

5. 两个Runnable核心类的讲解与使用

RunnableParallel - 并行执行

功能：并行执行多个 Runnable 对象，将结果组合成字典返回

优势：

提高执行效率
结果聚合
简化代码

python

from langchain_core.runnables import RunnableParallel

joke_chain = joke_prompt | llm | parser
poem_chain = poem_prompt | llm | parser

# 并行执行
map_chain = RunnableParallel(joke=joke_chain, poem=poem_chain)

res = map_chain.invoke({"subject": "程序员"})
# 输出: {'joke': '...', 'poem': '...'}

RunnablePassthrough - 数据透传

功能：在链中传递原始输入数据，通过 assign 添加新字段

优势：

简化链的构建
保留原始输入
动态添加字段

python

from langchain_core.runnables import RunnablePassthrough

chain = RunnablePassthrough.assign(
    context=lambda x: retrieval(x["query"])
) | prompt | llm | parser

工具类对比

工具类	主要用途	典型场景
RunnableParallel	并行执行多个链，聚合结果	同时生成多种内容、并行调用不同模型
RunnablePassthrough	透传数据，动态添加字段	RAG检索、数据增强处理

RAG 推荐写法

python

chain = RunnablePassthrough.assign(
    context=lambda x: retrieval(x["query"])
) | prompt | llm | parser

6. 利用回调功能调试链应用

是什么

LangChain 回调功能是一种监控和调试链式应用执行过程的机制，基于事件驱动模式。

有什么用

执行过程可视化 - 实时查看链的执行流程
性能监控 - 统计各环节执行耗时
调试与问题定位 - 记录详细的执行日志
自定义监控告警 - 集成日志系统

常用回调方法

on_chat_model_start: 聊天模型开始执行
on_chat_model_end: 聊天模型执行结束
on_llm_start: LLM 开始执行
on_llm_end: LLM 执行结束
on_chain_start: 链开始执行
on_chain_end: 链执行结束

自定义回调处理器示例

python

import time
from langchain_core.callbacks import BaseCallbackHandler

class LLMOpsCallbackHandler(BaseCallbackHandler):
    start_at: float = 0

    def on_chat_model_start(self, serialized, messages, run_id, **kwargs):
        print("聊天模型开始执行了")
        self.start_at = time.time()

    def on_llm_end(self, response, run_id, **kwargs):
        end_at = time.time()
        print("程序消耗:", end_at - self.start_at)

# 使用回调
resp = chain.stream(
    "你好",
    config={"callbacks": [LLMOpsCallbackHandler()]}
)

实际应用场景

开发调试阶段：快速定位问题，查看中间结果
性能优化：识别耗时环节，对比不同实现的性能
日志审计：记录所有调用记录，追踪用户操作
监控告警：集成到监控系统，设置异常阈值告警

7. Python+OpenAI原生SDK实现记忆功能

是什么

使用 Python 和 OpenAI 原生 SDK 从零实现的摘要缓存混合记忆（ConversationSummaryBufferMemory）系统。

核心原理

摘要缓存混合记忆：

┌─────────────────────────────────────┐
│  摘要部分（Summary）                  │
│  存储早期对话的提炼总结                │
└─────────────────────────────────────┘
              ↓ 随对话增长动态生成
┌─────────────────────────────────────┐
│  缓冲部分（Buffer）                   │
│  存储最近的完整对话记录                │
└─────────────────────────────────────┘

工作流程

初始阶段：所有对话存储在缓存中
触发摘要：当缓存token数超过 max_tokens 阈值时
生成摘要：将最早的对话与现有摘要合并
移除旧对话：将已摘要的对话从缓存中删除
循环进行：保持对话历史的可控性

核心类实现

python

class ConversationSummaryBufferMemory:
    def __init__(self, summary: str = '', chat_histories: list = None, max_tokens: int = 300):
        self.summary = summary
        self.chat_histories = [] if chat_histories is None else chat_histories
        self.max_tokens = max_tokens
        self.client = OpenAI(base_url='https://api.moonshot.cn/v1')

    def save_content(self, human_query: str, ai_content: str) -> None:
        # 保存新对话并检查是否需要生成摘要
        self.chat_histories.append({'human': human_query, 'ai': ai_content})

        if self.get_num_tokens(self.get_buffer_string()) > self.max_tokens:
            first_chat = self.chat_histories[0]
            self.summary = self.summary_text(self.summary, f"human: {human_query}\nai: {first_chat.get('ai')}")
            del self.chat_histories[0]

    def load_memory_variables(self) -> dict:
        buffer_string = self.get_buffer_string()
        return {"chat_history": f"摘要: {self.summary}\n历史信息:{buffer_string}"}

应用场景

场景	优势
智能客服	记住用户问题和历史投诉
个性化助手	记住用户偏好和习惯
教育辅导	跟踪学习进度和知识点
心理咨询	保持长期咨询的连贯性

8. ChatMessage组件上手与源码解析

是什么

ChatMessage 是 LangChain 框架中用于管理和存储对话历史的组件。

主要实现类

InMemoryChatMessageHistory（内存存储）

用途：将对话历史存储在内存中

优点：速度快、无需额外配置缺点：程序重启后数据丢失

python

from langchain_core.chat_history import InMemoryChatMessageHistory

chat_history = InMemoryChatMessageHistory()
chat_history.add_user_message("你好，我是ccc")
chat_history.add_ai_message("你好，我是chatgpt")
print(chat_history.messages)

FileChatMessageHistory（文件存储）

用途：将对话历史持久化到文件中

优点：数据持久化、跨会话记忆缺点：读写速度相对内存慢

python

from langchain_community.chat_message_histories import FileChatMessageHistory

chat_history = FileChatMessageHistory('./memory.txt')
chat_history.add_user_message("你好")
chat_history.add_ai_message("你好！")

常用API方法

方法	说明
`add_user_message(message)`	添加用户消息
`add_ai_message(message)`	添加 AI 消息
`messages` 属性	获取所有消息
`clear()`	清空历史

使用场景对比

场景	推荐组件	原因
简单对话机器人	InMemoryChatMessageHistory	无需持久化，简单快速
客服系统	FileChatMessageHistory	需要保存对话记录
聊天应用	FileChatMessageHistory	用户需要历史记录

9. 缓冲窗口记忆组件

是什么

缓冲窗口记忆组件（Buffer Window Memory）是一种带有限容量的对话历史管理机制。

有什么用

节省API成本 - 限制历史Token数控制成本
提升响应速度 - 减少每次请求的输入Token数量
避免超出上下文窗口 - 防止历史记录过多超出模型处理能力
保持对话连贯性 - 在Token限制内保留最近的对话内容

实现方式对比

方式一：SimpleMemory（手动实现）

python

class SimpleMemory:
    def __init__(self, max_token_limit=2000):
        self.history = []
        self.max_token_limit = max_token_limit

    def _trim_history(self):
        total_chars = sum(len(msg.content) for msg in self.history)
        max_chars = self.max_token_limit * 4

        while total_chars > max_chars and len(self.history) > 2:
            removed = self.history.pop(0)
            total_chars -= len(removed.content)

方式二：RunnableWithMessageHistory（集成方案）

python

from langchain_core.runnables.history import RunnableWithMessageHistory

store = {}

def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = FileChatMessageHistory(f"chat_history_{session_id}.txt")
    return store[session_id]

with_message_chain = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="query",
    history_messages_key="history",
)

关键参数

参数	说明	推荐值
`max_token_limit`	最大Token数量限制	2000-4000
`session_id`	会话唯一标识符	用户ID或会话UUID

最佳实践

合理设置Token限制（根据模型上下文窗口大小）
持久化存储建议使用数据库（Redis、PostgreSQL等）
为每个用户或会话分配唯一的session_id
考虑使用异步操作提升并发性能

10. Runnable组件动态添加默认调用参数

是什么

bind() 是 Runnable 组件的方法，用于动态添加默认调用参数。

有什么用

简化调用 - 提前绑定固定参数
参数复用 - 避免每次 invoke 都重复传相同参数
多参场景 - 解决 RunnableLambda 只能接收单参的问题
动态覆盖 - 在链式调用中动态修改 LLM 参数

示例1：RunnableLambda 预设默认参数

python

from langchain_core.runnables import RunnableLambda

def get_weather(location: str, unit: str, name: str) -> str:
    return f"{name}你好，查询到{location}的天气是25度{unit}"

# 绑定 unit 和 name，调用时只需传 location
get_weather_runnable = RunnableLambda(get_weather).bind(unit="摄氏度", name="muxiaoke")

res = get_weather_runnable.invoke("广州")
# 输出: muxiaoke你好，查询到广州的天气是25度摄氏度

示例2：LLM 链式调用中动态覆盖模型参数

python

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_messages([("human", "{query}")])
llm = ChatOpenAI(model="moonshot-v1-8k")

# 使用 bind 动态覆盖 model 参数，从 8k 切换到 32k
chain = prompt | llm.bind(model="moonshot-v1-32k") | StrOutputParser()

content = chain.invoke({"query": "你好，你是什么模型"})
# 实际使用的是 moonshot-v1-32k 模型

总结

本学习路径涵盖了 LangChain 框架从基础到高级的核心知识：

基础组件（Prompt、Model、OutputParser）
LCEL表达式与Runnable协议
高级组件（RunnableParallel、RunnablePassthrough）
调试技巧（回调功能）
记忆管理（原生实现、ChatMessage、缓冲窗口）
高级技巧（bind函数）

通过系统学习这些组件，可以构建生产级的 LLM 应用程序。

LLMOps API 学习总文档 ​

目录 ​

1. Prompt组件及使用技巧 ​

核心组件 ​

基础用法 ​

关键方法 ​

拼接技巧 ​

技巧对比 ​

2. Model组件及使用技巧 ​

三种调用方式 ​

基础使用 ​

批处理示例 ​

流式输出示例 ​

使用技巧 ​

3. OutputParser组件及使用技巧 ​

StrOutputParser - 字符串输出解析器 ​

JsonOutputParser - JSON输出解析器 ​

执行流程 ​

4. LCEL表达式与Runnable可运行协议 ​

核心概念 ​

LCEL 链式调用示例 ​

数据流转过程 ​

最佳实践 ​

5. 两个Runnable核心类的讲解与使用 ​

RunnableParallel - 并行执行 ​

RunnablePassthrough - 数据透传 ​

工具类对比 ​

RAG 推荐写法 ​

6. 利用回调功能调试链应用 ​

是什么 ​

有什么用 ​

常用回调方法 ​

自定义回调处理器示例 ​

实际应用场景 ​

7. Python+OpenAI原生SDK实现记忆功能 ​

是什么 ​

核心原理 ​

工作流程 ​

核心类实现 ​

应用场景 ​

8. ChatMessage组件上手与源码解析 ​

是什么 ​

主要实现类 ​

InMemoryChatMessageHistory（内存存储） ​

FileChatMessageHistory（文件存储） ​

常用API方法 ​

使用场景对比 ​

9. 缓冲窗口记忆组件 ​

是什么 ​

有什么用 ​

实现方式对比 ​

方式一：SimpleMemory（手动实现） ​

方式二：RunnableWithMessageHistory（集成方案） ​

关键参数 ​

最佳实践 ​

10. Runnable组件动态添加默认调用参数 ​

是什么 ​

有什么用 ​

示例1：RunnableLambda 预设默认参数 ​

示例2：LLM 链式调用中动态覆盖模型参数 ​

总结 ​

LLMOps API 学习总文档

目录

1. Prompt组件及使用技巧

核心组件

基础用法

关键方法

拼接技巧

技巧对比

2. Model组件及使用技巧

三种调用方式

基础使用

批处理示例

流式输出示例

使用技巧

3. OutputParser组件及使用技巧

StrOutputParser - 字符串输出解析器

JsonOutputParser - JSON输出解析器

执行流程

4. LCEL表达式与Runnable可运行协议

核心概念

LCEL 链式调用示例

数据流转过程

最佳实践

5. 两个Runnable核心类的讲解与使用

RunnableParallel - 并行执行

RunnablePassthrough - 数据透传

工具类对比

RAG 推荐写法

6. 利用回调功能调试链应用

是什么

有什么用

常用回调方法

自定义回调处理器示例

实际应用场景

7. Python+OpenAI原生SDK实现记忆功能

是什么

核心原理

工作流程

核心类实现

应用场景

8. ChatMessage组件上手与源码解析

是什么

主要实现类

InMemoryChatMessageHistory（内存存储）

FileChatMessageHistory（文件存储）

常用API方法

使用场景对比

9. 缓冲窗口记忆组件

是什么

有什么用

实现方式对比

方式一：SimpleMemory（手动实现）

方式二：RunnableWithMessageHistory（集成方案）

关键参数

最佳实践

10. Runnable组件动态添加默认调用参数

是什么

有什么用

示例1：RunnableLambda 预设默认参数

示例2：LLM 链式调用中动态覆盖模型参数

总结