步骤1：配置APIkey

LangChain 提供了简单直观的 API 来快速构建 LLM 应用程序。

申请 API key 并配置环境变量 在系统环境变量中设置对应大模型API_KEY OPENAI_API_KEY=your-api-key

步骤2：定义大模型

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-3.5-turbo")

步骤3：定义消息列表

from langchain_core.messages import HumanMessage, SystemMessage
messages = [
    SystemMessage(content="你是一个 helpful assistant."),
    HumanMessage(content="请介绍你自己")
]

步骤4：调用大模型

response = llm.invoke(messages)

步骤5：输出解析

print(response.content)

步骤6：链式执行

from langchain_core.output_parsers import StrOutputParser
chain = llm | StrOutputParser()
result = chain.invoke(messages)

2. 聊天模型核心能力

定义聊天模型

# 方式1：ChatOpenAI
from langchain_openai import ChatOpenAI
chat = ChatOpenAI(model="gpt-3.5-turbo")
​
# 方式2：init_chat_model
from langchain_community.chat_models import init_chat_model
chat = init_chat_model("openai", model="gpt-3.5-turbo")

聊天模型--调用工具

创建工具

# 使用 @tool 装饰器创建工具
from langchain_core.tools import tool
@tool
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b
​
# 使用 StructuredTool 类提供的函数创建工具
from langchain_core.tools import StructuredTool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers"""
    return a * b
multiply_tool = StructuredTool.from_function(multiply)

绑定工具

tools = [add, multiply_tool]
chat_with_tools = chat.bind_tools(tools)

工具调用

response = chat_with_tools.invoke("计算 3 + 5")

强制模型调用工具

response = chat_with_tools.invoke("计算 3 + 5", tool_choice="add")

工具属性

@tool
def complex_operation(a: int, b: int, c: str) -> str:
    """Complex operation with description and parameters"""
    return f"Result: {a + b}, {c}"

将工具输出传递给聊天模型

tool_call = response.tool_calls[0]
tool_output = add.invoke(tool_call["args"])
final_response = chat_with_tools.invoke([
    HumanMessage(content="计算 3 + 5"),
    response,
    ToolMessage(tool_output=tool_output, tool_call_id=tool_call["id"])
])

聊天模型--结构化输出

with_structured_output()

from pydantic import BaseModel, Field
class Joke(BaseModel):
    setup: str = Field(description="question to set up a joke")
    punchline: str = Field(description="answer to resolve the joke")
structured_llm = chat.with_structured_output(Joke)
joke = structured_llm.invoke("tell me a joke")

返回 Pydantic 对象

class Person(BaseModel):
    name: str
    age: int
structured_output = chat.with_structured_output(Person)
person = structured_output.invoke("John is 30 years old")

返回 TypedDict

from typing import TypedDict
class Person(TypedDict):
    name: str
    age: int
structured_output = chat.with_structured_output(Person)
person = structured_output.invoke("John is 30 years old")

返回 JSON

structured_output = chat.with_structured_output(schema={"properties": {"name": {"type": "string"}, "age": {"type": "integer"}}})
person = structured_output.invoke("John is 30 years old")

选择输出格式

# 返回 Pydantic
structured_output = chat.with_structured_output(Joke)
​
# 返回 JSON Schema
structured_output = chat.with_structured_output(Joke.model_json_schema())
​
# 返回 TypedDict
structured_output = chat.with_structured_output(Person)

聊天模型--流式传输

stream() 同步传输

for chunk in chat.stream("写一个关于AI的故事"):
    print(chunk.content, end="", flush=True)

astream() 异步传输

async for chunk in chat.astream("写一个关于AI的故事"):
    print(chunk.content, end="", flush=True)

使用 StrOutputParser 解析模型的输出

from langchain_core.output_parsers import StrOutputParser
chain = chat | StrOutputParser()
async for chunk in chain.astream("写一个关于AI的故事"):
    print(chunk, end="", flush=True)

自定义流式输出解析器

from langchain_core.output_parsers import BaseOutputParser
class CustomParser(BaseOutputParser):
    def parse(self, text):
        return text.upper()
chain = chat | CustomParser()
async for chunk in chain.astream("写一个关于AI的故事"):
    print(chunk, end="", flush=True)

SSE 协议介绍 SSE (Server-Sent Events) 是一种允许服务器向客户端推送实时更新的技术，常用于流式传输。

LangChain 流式传输流程分析

1. 客户端发起请求

2. 服务器保持连接打开

3. 服务器分块发送数据

4. 客户端接收并处理每个数据块

使用 LangSmith 跟踪 LLM 应用

from langchain.callbacks import LangChainTracer
tracer = LangChainTracer(project_name="my-project")
with tracer:
    response = chat.invoke("你好")

3. 核心组件（Components）

LangChain 的核心组件包括：

• LLMs (大语言模型)

• Prompts (提示词)

• Chains (链)

• Indexes (索引)

• Agents (代理)

• Memory (记忆)

• Output Parsers (输出解析器)

4. 消息（Messages）

LLM 消息结构 LangChain 支持多种消息类型，包括 HumanMessage、AIMessage、SystemMessage 等。

LangChain 消息

from langchain_core.messages import (
    HumanMessage,
    SystemMessage,
    AIMessage,
    ToolMessage,
    FunctionMessage
)

BaseMessage 抽象消息类 所有消息类型的基类，提供统一的消息接口。

对话模式 支持多种对话模式，包括单轮对话、多轮对话等。

缓存历史消息

from langchain_core.chat_history import InMemoryChatMessageHistory
history = InMemoryChatMessageHistory()
history.add_user_message("你好")
history.add_ai_message("你好！有什么可以帮助你的吗？")

多轮对话

messages = [
    SystemMessage(content="你是一个 helpful assistant."),
    HumanMessage(content="你好"),
    AIMessage(content="你好！有什么可以帮助你的吗？"),
    HumanMessage(content="请介绍你自己")
]

内存缓存

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
​
store = {}
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]
​
with_message_history = RunnableWithMessageHistory(chat, get_session_history)
response = with_message_history.invoke(
    [HumanMessage(content="你好")],
    config={"configurable": {"session_id": "abc123"}}
)

管理历史消息

• 添加消息

• 获取消息

• 清除消息

前置概念

• 上下文窗口

• Token

消息裁剪 当消息超过上下文窗口时，需要裁剪消息。

基于输入 Token 数的修剪

from langchain_core.messages import trim_messages
trimmed_messages = trim_messages(
    messages,
    max_tokens=100,
    strategy="last",
    token_counter=llm
)

基于消息数的修剪

trimmed_messages = trim_messages(
    messages,
    max_messages=5,
    strategy="last"
)

消息过滤

trimmed_messages = trim_messages(
    messages,
    include_types=["human", "ai"],
    exclude_types=["system"]
)

消息合并

trimmed_messages = trim_messages(
    messages,
    strategy="merge",
    token_counter=llm
)

5. 提示词模板（Prompt Template）

概念 提示词模板是可重用的提示词结构，允许动态插入变量。

用法

from langchain_core.prompts import PromptTemplate
template = "你是一个 {role}，请 {action}"
prompt = PromptTemplate.from_template(template)
formatted_prompt = prompt.format(role="翻译专家", action="将以下中文翻译成英文：{text}")

字符串模板

template = "翻译以下文本：\n\n{text}\n\n到 {language}"
prompt = PromptTemplate(template=template, input_variables=["text", "language"])

聊天消息模板

from langchain_core.prompts import ChatPromptTemplate
template = ChatPromptTemplate.from_messages([
    ("system", "你是一个 {role}"),
    ("human", "{text}")
])

消息占位符

from langchain_core.prompts import MessagesPlaceholder
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个 helpful assistant."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

使用 LangChain Hub 的提示词模板

from langchain import hub
prompt = hub.pull("jamesliang/openai-functions-agent")

6. 少样本提示（few-shotting）

概念 少样本提示通过提供少量示例来指导模型生成更准确的输出。

实现

from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate
examples = [
    {"input": "happy", "output": "joyful"},
    {"input": "sad", "output": "melancholy"}
]
example_prompt = PromptTemplate.from_template("Input: {input}\nOutput: {output}")
few_shot_prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
    prefix="将以下情感词转换为同义词：",
    suffix="Input: {input}\nOutput:",
    input_variables=["input"]
)

使用案例

案例一：推理引导 通过提供推理步骤的示例来引导模型进行复杂推理。

案例二：使用示例数据增强 LangChain 信息提取能力

examples = [
    {"text": "苹果公司成立于1976年", "entity": "苹果公司", "date": "1976年"},
    {"text": "微软总部位于华盛顿", "entity": "微软", "location": "华盛顿"}
]
few_shot_prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
    prefix="从文本中提取实体信息：",
    suffix="Text: {text}\nExtracted:",
    input_variables=["text"]
)

7. 示例选择器（Example selectors）

概念 示例选择器根据输入动态选择最相关的示例。

按长度选择示例（Length）

from langchain_core.example_selectors import LengthBasedExampleSelector
example_selector = LengthBasedExampleSelector(
    examples=examples,
    example_prompt=example_prompt,
    max_length=50
)

按语义相似性选择示例（Similarity）

from langchain_community.example_selectors import SemanticSimilarityExampleSelector
from langchain_openai import OpenAIEmbeddings
example_selector = SemanticSimilarityExampleSelector.from_examples(
    examples,
    OpenAIEmbeddings(),
    FAISS,
    k=1
)

按最大边际相关性选择示例（MMR）

from langchain.retrievers import MultiQueryRetriever
retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=llm
)

通过 ngram 重叠选择示例（Ngram）

from langchain_community.example_selectors import NGramOverlapExampleSelector
example_selector = NGramOverlapExampleSelector(
    examples=examples,
    example_prompt=example_prompt,
    ngram_size=2
)

8. 输出解析器（Output parsers）

概念 输出解析器将 LLM 的输出转换为结构化格式。

解析文本输出

from langchain_core.output_parsers import StrOutputParser
parser = StrOutputParser()
parsed_output = parser.parse(response.content)

解析结构化对象输出

from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
class Person(BaseModel):
    name: str = Field(description="person's name")
    age: int = Field(description="person's age")
parser = PydanticOutputParser(pydantic_object=Person)

解析 JSON 输出

from langchain.output_parsers import JsonOutputParser
parser = JsonOutputParser()
parsed_output = parser.parse(response.content)

9. 文档加载器（Document loaders）

RAG 介绍 RAG (Retrieval-Augmented Generation) 结合检索和生成，提高回答的准确性和相关性。

RAG 概念 通过检索相关文档来增强 LLM 的知识。

RAG 流程

1. 检索相关文档

2. 将文档与问题结合

3. 生成回答

RAG 示例

from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
​
loader = TextLoader("data.txt")
documents = loader.load()
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.split_documents(documents)
embeddings = OpenAIEmbeddings()
db = FAISS.from_documents(texts, embeddings)

Document 类

from langchain_core.documents import Document
document = Document(page_content="文本内容", metadata={"source": "文档来源"})

加载 PDF 文档

from langchain_community.document_loaders import PyPDFLoader
loader = PyPDFLoader("document.pdf")
pages = loader.load_and_split()

加载 Markdown 文件

from langchain_community.document_loaders import UnstructuredMarkdownLoader
loader = UnstructuredMarkdownLoader("document.md")
documents = loader.load()

10. 文本分割器（Text splitters）

概念 文本分割器将长文档分割成更小的块，以便处理。

根据文档长度与文档语义拆分 平衡块大小和语义完整性。

基于字符长度拆分

from langchain.text_splitter import CharacterTextSplitter
text_splitter = CharacterTextSplitter(
    separator="\n\n",
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len
)
texts = text_splitter.split_text(state_of_the_union)

基于 Token 长度拆分

from langchain.text_splitter import TokenTextSplitter
text_splitter = TokenTextSplitter(chunk_size=100, chunk_overlap=0)
texts = text_splitter.split_text(state_of_the_union)

硬性约束长度拆分

from langchain.text_splitter import RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=100,
    chunk_overlap=20,
    length_function=len,
    add_start_index=True
)
texts = text_splitter.create_documents([state_of_the_union])

特殊文档结构拆分

from langchain.text_splitter import MarkdownHeaderTextSplitter
markdown_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[
        ("#", "Header 1"),
        ("##", "Header 2"),
    ]
)
texts = markdown_splitter.split_text(markdown_text)

11. 文本向量

嵌入与嵌入模型（Embedding and Embedding Models） 将文本转换为向量表示，便于相似性搜索。

Embeddings 嵌入模型类

from langchain.embeddings import OpenAIEmbeddings
embeddings = OpenAIEmbeddings()

定义嵌入模型

from langchain.embeddings import HuggingFaceEmbeddings
embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-mpnet-base-v2")

嵌入文档列表

texts = ["Hello world", "Hi there"]
embeddings_list = embeddings.embed_documents(texts)

嵌入单个查询

query = "Hello world"
query_embedding = embeddings.embed_query(query)

向量存储（Vector Stores）

向量数据库介绍 存储和检索向量数据的数据库。

内存存储

from langchain.vectorstores import FAISS
db = FAISS.from_texts(texts, embeddings)

基本操作

• 添加文档

• 删除文档

• 更新文档

向量搜索

docs = db.similarity_search("query")

Redis 向量存储

基本概念 Redis 作为向量数据库，提供高性能的向量搜索。

环境设置

pip install redis

基本操作

from langchain.vectorstores import Redis
redis_url = "redis://localhost:6379"
index_name = "langchain-demo"
vector_store = Redis.from_texts(
    texts,
    embeddings,
    redis_url=redis_url,
    index_name=index_name
)

向量搜索

results = vector_store.similarity_search("query")

相似性搜索

results = vector_store.similarity_search_with_score("query")

元数据过滤

results = vector_store.similarity_search(
    "query",
    filter={"source": "my-source"}
)

最大边际相关性搜索

results = vector_store.max_marginal_relevance_search("query")

Pinecone 向量存储

介绍 Pinecone 是一个托管的向量数据库服务。

环境设置

pip install pinecone-client

基本操作

from langchain.vectorstores import Pinecone
pinecone.init(api_key="your-api-key", environment="your-env")
index_name = "langchain-demo"
vector_store = Pinecone.from_texts(texts, embeddings, index_name=index_name)

向量搜索

results = vector_store.similarity_search("query")

12. 检索器（Retrievers）

相关概念 检索器是从向量存储中检索相关文档的组件。

检索系统 结合检索和生成的系统。

检索器

retriever = vector_store.as_retriever()

使用向量数据库作为检索器

retriever = db.as_retriever(search_kwargs={"k": 4})
docs = retriever.get_relevant_documents("query")