一、定义

langchain是大模型应用的开发框架，简化大模型应用的开发、测试、评估、运营。不是大模型的开发、训练框架，比如想要从0-1训练一个LLM模型，使用PyTorch框架更合适。

二、用法

（1）安装新版本python，3.10+。 # 安装方法可以问元宝

（2）安装依赖包

pip install langchain
pip install langchain_core

（3）聊天机器人

from langchain.chat_models import init_chat_model

model = init_chat_model("gpt-4o-mini", model_provider="openai")

print(model.invoke("你好"))

（4）翻译小助手

# 使用chat model和prompt模版构建llm应用
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage

model = init_chat_model("gpt-4o-mini", model_provider="openai")

# 定义系统消息(设定GPT角色)
system_message = SystemMessage(content="你是一个专业的中英翻译助手。请将用户输入的中文文字翻译为英文，只翻译最新输入的内容，不要翻译历史内容。")

while True:
    user_input = input("请输入要翻译的文字: ")
    messages = [
        system_message,
        HumanMessage(content=user_input)
    ]
    response = model.invoke(messages)
    print(f"翻译结果: {response.content}")

提示：代码直接运行可能失败。依赖环境变量中openai的密钥。

三、适用场景

用于大模型应用的开发，比如聊天机器人（聊天+查询互联网资料+查询公司内部系统数据）、智能问答机器人（知识库搜索）、翻译小助手。

一个业务场景示例：

场景：devops领域包含发布系统，发布任务执行失败时，希望能够快速定位原因：系统问题（db故障）、业务服务自身问题（端口没有监听）。如果是业务服务问题，提供具体的报错日志（省去人工登机器找日志）。

解决方案：基于langchain框架开发一个agent，只有一个工具：根据任务链接和机器IP分析失败原因（采用规则匹配识别系统问题，收集服务启动日志后丢给大模型，让大模型总结出报错日志&可能原因）。

四、实现机制&架构

1、核心层，langchain-core

提供：基础抽象（Runnable接口（链式调用时需实现该接口，invoke函数）、提示词模版（xxPromptTemplate）、记忆管理（短期、长期记忆，基于内存、向量数据库、KV数据库）），定义组件的交互标准。

2、集成层，langchain-community

集成第三方sdk。比如openai、deepseek、混元等LLM、Hugging Face的sdk。通过model = ChatOpenAI(xx) 可以快速构建一个LLM客户端。

3、功能层，langchain

提供：链（Chains，提示词模板|LLM|输出格式化）、代理（Agents智能体，AgentExecutor）、检索器（Retrieval，RetrievalQA） 等sdk包。

4、生产层，LangServe、LangSmith

提供：LLM调用的调试、监控。记录每一次LLM调用。

注：LangGraph框架不包含在这里，其侧重解决复杂工作流编排问题，其他问题仍然用上面这些工具包处理。

五、功能模块及作用

1、模型，Models

作用：统一各LLM大模型的接口，包含文本生成（包含聊天、翻译、分类、完形填空等）、对话（聊天单独一个工具类，简化使用）、嵌入（文本转嵌入向量）等任务。

关键组件：

LLM：通用的文本生成模型（如GPT-4、Claude）。

ChatModel：对话专用模型（支持多轮对话，记忆）。

Embedding：文本向量化模型（如OpenAI Embedding、BERT、BGE）。

示例：

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model_name="gpt-4-turbo")  # 初始化对话模型

2、链，Chains

作用：将多个组件串联为任务流程。

关键组件：

LLMChain：基础链，直接调用LLM。

示例：

from langchain.chat_models import init_chat_model
model = init_chat_model("gpt-4o-mini", model_provider="openai")

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser

# 创建提示模板
prompt_template = ChatPromptTemplate([
    ("system", "你是一个专业的中英翻译助手。请将用户输入的中文文字翻译为英文。"),
    MessagesPlaceholder("msgs")
])

# 直接调用
translation_chain = prompt_template | model | StrOutputParser()

result1 = translation_chain.invoke({"msgs": [HumanMessage(content="你好")]})
print(f"翻译 '你好': {result1}")

RetrievalQA：检索增强（RAG）链。

texts = text_splitter.split_documents(documents)

# 1 文本嵌入
embeddings = OpenAIEmbeddings()

# 2 构建向量存储库
docsearch = Chroma.from_documents(texts, embeddings)

# 3 构建检索QA链
llm = ChatOpenAI(model="gpt-4", temperature=0)
qa = RetrievalQA.from_chain_type(
    llm=llm, 
    chain_type="stuff", 
    retriever=docsearch.as_retriever(
        search_kwargs={
            "k": 3
        }
    )
)

answer = qa.run("langchain是什么？")
print(answer)

SequentialChain：按顺序执行多步骤任务

3、记忆，Memory

作用：管理对话上下文（历史聊天消息），支持多轮交互。

关键组件：

ConversationBufferMemory：基于内存的短期对话缓存对象

RedisMemory：长期存储到Redis

from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory(memory_key="chat_history")  # 记录对话历史

4、代理，Agents

作用：动态调用工具（API或数据库），实现多步骤推理（ReAct）

关键组件：

ZeroShotAgent：零样本代理

ReActAgent：基于推理-行动循环的代理。

from langchain_openai import ChatOpenAI
from langchain.agents import AgentType, initialize_agent
from langchain.memory import ConversationBufferMemory
from langchain_core.tools import Tool

# 初始化模型
llm = ChatOpenAI(
    model="gpt-3.5-turbo", 
    temperature=0, 
    api_key=api_key,
    base_url=base_url
    )

# 创建工具列表

tools = [
    Tool.from_function(
        func=query_order,
        name="query_order",
        description="根据订单ID查询订单信息"
    )
]

# 初始化记忆组件
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)

# 创建ReAct Agent
agent_executor = initialize_agent(
    llm=llm, 
    tools=tools, 
    memory=memory,
    agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
    verbose=True,
)

# 执行代理
agent_executor.invoke({"input": "我想查询订单P001的信息，以及商品的信息"})

5、检索器，Retrieval

作用：从外部数据源（知识库，如网页、数据库）检索信息。

流程​​：网页加载 → 分块 → 向量化 → 存储 → 检索。

示例：

from langchain_community.vectorstores import FAISS
vector_db = FAISS.from_documents(docs, embeddings)  # 构建向量数据库
retriever = vector_db.as_retriever()  # 创建检索器

6、提示词模板

作用：对大模型设立角色，规范输入格式，动态注入带参数的提示词。

示例：

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的中英翻译助手。请将用户输入的中文文字翻译为英文，只翻译最新输入的内容，不要翻译历史内容。"),
    ("human", "{input_text}")
])

六、最佳实践

langchain 本质是大模型应用的完整开发框架。通过模块化的组件（提示词、大模型、输出格式化、链、智能体、记忆、检索）和标准化接口（ChatModel、多种大模型）降低大模型应用的开发门槛。最佳实践可以分为4个方面：

最佳实践1、采用LCEL（langchain表达式语言）进行声明式链编程（简化复杂工作流）

通过管道符 | 连接多个组件：Prompt模板、大模型对象、格式化输出，实现快捷组合、原生流支持、异步/批量调用。

代码示例：

# -*- coding: utf-8 -*-
# @Time    : 2025/10/03
# @Author  : frank
# @File    : demo01-lcel.py
# @Description: langchain expression language 示例

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

load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template("科普一个关于{subject}的知识")
output_parser = StrOutputParser()

chain = prompt | llm | output_parser
print(chain.invoke({"subject": "银河系"}))

输出结果：

河系是我们所处的星系，包含着数千亿颗恒星、行星、气体和尘埃。它的形状呈螺旋状，直径大
约为10万光年，厚度约为1000光年。银河系的中心区域有一个超大质量黑洞，名为人马座A*，其质
量约为太阳的400万倍。

最佳实践2：提示工程。提示是LLM输出的​​关键输入​​，建议遵守以下技巧：

（1）模板化提示：使用PromptTemplate定义固定格式的Prompt，避免硬编码

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是卡尔萨根，一个著名的天文学家，需回答用户关于天文学的咨询"),
    ("human", "{user_input}")
])

（2）少样本提示：在提示中加入示例，引导LLM生成符合要求的输出。

prompt = ChatPromptTemplate.from_messages([
    ("system", "翻译以下文本为法语"),
    ("human", "Hello World → Bonjour le monde"),
    ("human", "{text} →")
])

最佳实践3：上下文管理（长短期记忆、检索优化）

（1）短期记忆

# -*- coding: utf-8 -*-
""" 
@description: 短期记忆示例 - 使用ConversationBufferWindowMemory保存最近5次对话
@author: frank
@date: 2025-10-03
@version: 1.0
"""
from dotenv import load_dotenv
import os

load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.memory import ConversationBufferWindowMemory
from langchain_core.messages import HumanMessage, AIMessage

model = init_chat_model("gpt-4o-mini", model_provider="openai")

# 创建内存管理器，保存最近2次对话
memory = ConversationBufferWindowMemory(k=2, return_messages=True)

while True:
    user_input = input("请输入文字: ")
    if user_input == "exit" or user_input == "quit":
        break
    
    # 添加用户消息到内存
    memory.chat_memory.add_user_message(user_input)
    
    # 获取历史消息 - 使用load_memory_variables确保k参数生效
    memory_variables = memory.load_memory_variables({})
    messages = memory_variables['history']
    
    # 调用模型
    result = model.invoke(messages)
    print(result.content)
    print("-" * 50)
    
    # 添加AI回复到内存
    # 确保content是字符串类型
    ai_content = result.content if isinstance(result.content, str) else str(result.content)
    memory.chat_memory.add_ai_message(ai_content)
    
    # 显示当前内存中的对话数量
    print(f"当前内存中保存了 {len(messages)} 条消息（限制后）")
    print(f"原始内存中总共有 {len(memory.chat_memory.messages)} 条消息")
    print("=" * 50)

执行结果：

请输入文字: 玩一个成语接龙的游戏
好的，我们来玩成语接龙吧！我来开始，第一个成语是“画龙点睛”。请你接一个
以“睛”字开头的成语。
--------------------------------------------------
当前内存中保存了 1 条消息（限制后）
原始内存中总共有 2 条消息
==================================================
请输入文字: 睛目不凡 
很好，你接的成语是“睛目不凡”。接下来我接“凡事预则立”。请你接一个以“立”字开头的成语。
--------------------------------------------------
当前内存中保存了 3 条消息（限制后）
原始内存中总共有 4 条消息
==================================================
请输入文字: 立竿见影
不错，你接的是“立竿见影”。我接“影影绰绰”。请你接一个以“绰”字开头的成语
。
--------------------------------------------------
当前内存中保存了 4 条消息（限制后）
原始内存中总共有 6 条消息
==================================================
请输入文字: 绰绰有余
很好，你接的是“绰绰有余”。我接“余音绕梁”。请你接一个以“梁”字开头的成语
。
--------------------------------------------------
当前内存中保存了 4 条消息（限制后）
原始内存中总共有 8 条消息
==================================================
请输入文字: 停止成语接龙。请输出聊天记录中的成语，从第一个到最后一个。

好的，以下是聊天记录中的成语：

1. 立竿见影
2. 影影绰绰
3. 绰绰有余
4. 余音绕梁

如果你还有其他问题或需要，请告诉我！
--------------------------------------------------
当前内存中保存了 4 条消息（限制后）
原始内存中总共有 10 条消息
==================================================

（2）长期记忆：RAG+向量数据库实现语义检索。

文本切分：使用RecursiveCharacterTextSplitter（递归字符切分）将长文档分割为合适大小的块（如chunk_size=500，chunk_overlap=50），减少检索耗时。

# -*- coding: utf-8 -*-
""" 
@description: RAG示例
@author: frank
@date: 2025-10-03
@version: 1.0
"""
from dotenv import load_dotenv
import os

load_dotenv()

from langchain.document_loaders import WebBaseLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI

# 检查环境变量
if not os.getenv("OPENAI_API_KEY") or not os.getenv("OPENAI_API_BASE"):
    raise ValueError("请设置 OPENAI_API_KEY 和 OPENAI_API_BASE 环境变量")

# 加载网页数据
loader = WebBaseLoader(web_paths=["https://cloud.tencent.com/developer/article/2379888"])
documents  = loader.load()

# 打印文档内容
print(documents)
for doc in documents:
    print(f'doc: {doc.page_content}...')

# 1 对文档进行预处理：文档切割
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=200,  # 动态调整
    chunk_overlap=20,  # 重叠为chunk_size的10%
    length_function=len,      # 使用字符长度计算
    separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", "；"]  # 中文友好的分隔符
)

# 2 使用传统固定长度切分。段落切分为句子
texts = text_splitter.split_documents(documents)
print("✅ 使用传统固定长度切分")


# 文本嵌入模型
embeddings = OpenAIEmbeddings()

# 3 构建向量存储库
docsearch = Chroma.from_documents(texts, embeddings)

# 4 构建检索QA链
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
qa = RetrievalQA.from_chain_type(
    llm=llm, 
    chain_type="stuff", 
    retriever=docsearch.as_retriever(
        search_kwargs={
            "k": 3,
        }
    )
)

# 5 执行查询
while True:
    query = input("请输入问题: ")
    print(f'query: {query}')
    if query == "exit":
        break
    try:
        answer = qa.run(query)
        print(answer)
    except Exception as e:
        print(f"查询出错: {e}")
        break

请输入问题: langchain 是什么
query: langchain 是什么
Langchain 是一个开源框架，允许开发人员将大型语言模型（如 GPT-4）与外部
的计算和数据源结合起来。它提供了 Python 和 JavaScript（TypeScript）的 
软件包，旨在提升大型语言模型的功能，使其不仅能够回答通用问题，还能从数
据库或文件中提取信息并执行具体操作。Langchain 的设计使得大型语言模型能
够在更广泛的应用环境中进行操作和响应，扩展了它们的应用范围和有效性。  

注：检索效果还OK。回答质量不太好，是因为原始文章内容写得不太好，没有写到关键点：大模型应用开发。

最佳实践4：生产环境部署

异步与并发​​：

使用async/await实现异步调用（如llm.apredict()），提高吞吐量。

使用FastAPI或LangServe部署异步API，支持高并发请求。

异步调用代码示例：

# -*- coding: utf-8 -*-
"""
性能对比示例：同步 vs 异步
"""
import asyncio
import time
from typing import List

# 模拟AI处理时间
def simulate_ai_processing_sync(duration: float) -> str:
    """同步模拟AI处理"""
    time.sleep(duration)  # 阻塞等待
    return f"同步处理完成，耗时 {duration} 秒"

async def simulate_ai_processing_async(duration: float) -> str:
    """异步模拟AI处理"""
    await asyncio.sleep(duration)  # 非阻塞等待
    return f"异步处理完成，耗时 {duration} 秒"

def sync_benchmark(requests: int, duration: float) -> float:
    """同步性能测试"""
    start_time = time.time()
    
    for i in range(requests):
        result = simulate_ai_processing_sync(duration)
        print(f"同步请求 {i+1}: {result}")
    
    end_time = time.time()
    return end_time - start_time

async def async_benchmark(requests: int, duration: float) -> float:
    """异步性能测试"""
    start_time = time.time()
    
    # 创建所有异步任务
    tasks = []
    for i in range(requests):
        task = simulate_ai_processing_async(duration)
        tasks.append(task)
    
    # 并发执行所有任务
    results = await asyncio.gather(*tasks)
    
    for i, result in enumerate(results):
        print(f"异步请求 {i+1}: {result}")
    
    end_time = time.time()
    return end_time - start_time

async def main():
    """主函数"""
    requests = 5
    duration = 1.0  # 每个请求1秒
    
    print("=== 同步处理测试 ===")
    sync_time = sync_benchmark(requests, duration)
    print(f"同步总耗时: {sync_time:.2f} 秒")
    print(f"平均每个请求: {sync_time/requests:.2f} 秒")
    
    print("\n=== 异步处理测试 ===")
    async_time = await async_benchmark(requests, duration)
    print(f"异步总耗时: {async_time:.2f} 秒")
    print(f"平均每个请求: {async_time/requests:.2f} 秒")
    
    print(f"\n=== 性能提升 ===")
    improvement = sync_time / async_time
    print(f"性能提升: {improvement:.2f}x")
    print(f"时间节省: {((sync_time - async_time) / sync_time * 100):.1f}%")

if __name__ == "__main__":
    asyncio.run(main())

输出结果：

=== 同步处理测试 ===
同步请求 1: 同步处理完成，耗时 1.0 秒
同步请求 2: 同步处理完成，耗时 1.0 秒
同步请求 3: 同步处理完成，耗时 1.0 秒
同步请求 4: 同步处理完成，耗时 1.0 秒
同步请求 5: 同步处理完成，耗时 1.0 秒
同步总耗时: 5.00 秒
平均每个请求: 1.00 秒

=== 异步处理测试 ===
异步请求 1: 异步处理完成，耗时 1.0 秒
异步请求 2: 异步处理完成，耗时 1.0 秒
异步请求 3: 异步处理完成，耗时 1.0 秒
异步请求 4: 异步处理完成，耗时 1.0 秒
异步请求 5: 异步处理完成，耗时 1.0 秒
异步总耗时: 1.00 秒
平均每个请求: 0.20 秒

=== 性能提升 ===
性能提升: 4.99x
时间节省: 79.9%

FastAPI 示例：

# -*- coding: utf-8 -*-
""" 
@description: LangServe 简单示例 - 通过URL参数生成标语
@author: frank
@date: 2025-10-03
@version: 1.0
"""
from dotenv import load_dotenv
import os

load_dotenv()

# 检查环境变量
if not os.getenv("OPENAI_API_KEY") or not os.getenv("OPENAI_API_BASE"):
    raise ValueError("请设置 OPENAI_API_KEY 和 OPENAI_API_BASE 环境变量")

from fastapi import FastAPI, Query
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 创建 FastAPI 应用
app = FastAPI(title="LangServe 笑话生成器", version="1.0.0")

# 初始化组件
llm = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template("写一个关于{subject}的笑话")
output_parser = StrOutputParser()

# 创建处理链
chain = prompt | llm | output_parser

@app.get("/joke")
async def get_joke(subject: str = Query(..., description="笑话主题")):
    """生成关于指定主题的笑话"""
    try:
        # 使用链处理输入
        result = await chain.ainvoke({"subject": subject})
        return {"joke": result, "subject": subject}
    except Exception as e:
        return {"error": str(e)}

if __name__ == "__main__":
    import uvicorn
    print("启动 LangServe 服务...")
    print("访问 http://localhost:8000/joke?subject=人工智能 测试")
    print("访问 http://localhost:8000/docs 查看 API 文档")
    uvicorn.run(app, host="0.0.0.0", port=8000)

输出结果：

http://localhost:8000/joke?subject=人工智能

{"joke":"为什么人工智能总是很懒？\n\n因为它总是希望“算法”能解决所有问题！","subject":"人工智能"}