LangChain 学习和使用记录

LangGraph

StateGraph

StateGraph 核心概念总结

1. State 定义与数据累积

State 类：使用 TypedDict 定义状态结构
数据累积：通过 Annotated[list, add_messages] 实现数据追加而非覆盖

1 2	class State(TypedDict): messages: Annotated[list, add_messages]

这里的 add_messages 是一个装饰器，用于处理消息的追加逻辑。如果在实例化 StateGraph() 时，而不传入 State，那么在 .invoke() 之后的数据会……(?)不清楚。

2. 节点与函数映射

节点添加：StateGraph 实例可添加多个节点
函数绑定：每个节点对应一个操作函数
自动执行：触发节点时自动调用对应函数

1 2	def chatbot(state: State): return {"messages": [llm.invoke(state["messages"])]}

StateGraph 相当于在使用代码创建一张图，图由节点和边构成，每个节点代表一个操作或处理步骤，边则表示数据流转。但是边好像也能调用函数。

3. 编译与执行流程

编译：compile() 将 StateGraph 编译为可执行的 graph
调用方式：支持 .invoke() 和 .stream() 方法
输入输出：
- 输入：符合 State 结构的字典，如 {"messages": [HumanMessage(...)]}
- 输出：更新后的完整 State 对象
  如果 State 是一个多 key-value 的字典，那么实现 runnable 接口的函数在返回或调用.invoke()返回时，会依据返回的字典中的 key 来更新对应的值，而不是覆盖整个 State。

class State(TypedDict):
    messages: Annotated[list, add_messages]
    user_id: str
    session_id: str

4. 数据流转机制

初始输入: {"messages": [HumanMessage("你好")]}
↓
节点函数处理后: {"messages": [HumanMessage("你好"), AIMessage("回复")]}
↓
最终输出: 完整的 State 对象

5. 关键要点

函数返回值：节点函数必须返回符合 State 结构的字典
状态管理：Graph 内部不会自动维护 State，需要手动控制每次传递的 State
数据访问：output["messages"][-1] 获取最新消息
持续对话：每次调用 graph.invoke() 都会在当前状态基础上累积更新

LangChain

RAG 索引增强应用

1. RAG 基础概念

RAG 的基本操作顺序：从文档加载器加载文件、文本分割器分割文档片段，到创建嵌入模型，并调用嵌入模型将分片向量化，再创建向量数据库，存储向量化后的分片，最后使用检索器从向量数据库中检索相关分片。

实际实践

实践代码：

import os
from pathlib import Path
from dotenv import load_dotenv
import getpass

# 设置网络代理
os.environ["http_proxy"] = "http://127.0.0.1:10808"
os.environ["https_proxy"] = "http://127.0.0.1:10808"

# 加载环境变量
load_dotenv()

if not os.getenv("GOOGLE_API_KEY"):
    os.environ["GOOGLE_API_KEY"] = getpass.getpass("Enter your Google API Key: ")

from langchain_community.document_loaders import UnstructuredMarkdownLoader, PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_google_genai import GoogleGenerativeAIEmbeddings

import faiss
from langchain_community.vectorstores import FAISS
from langchain_community.docstore.in_memory import InMemoryDocstore

from langchain_core.documents import Document
from langchain_core.runnables import chain

# 创建嵌入模型
embeddings = GoogleGenerativeAIEmbeddings(model="models/embedding-001")

# 检查是否存在本地FAISS数据库
faiss_folder = Path("faiss_floder_pdf_en")
if faiss_folder.exists() and (faiss_folder / "index.faiss").exists():
    print("发现本地FAISS数据库，正在加载...")
    # 直接从本地加载向量存储
    vector_store = FAISS.load_local(
        str(faiss_folder), 
        embeddings, 
        allow_dangerous_deserialization=True
    )
    print("FAISS数据库加载完成！")
else:
    print("未发现本地FAISS数据库，开始创建新的向量数据库...")
    
    # 加载 markdown 文件
    loader = PyPDFLoader("./nke-10k-2023.pdf")
    docs = loader.load()
    print(f"加载了 {len(docs)} 个文档")

    # 使用文本分割器拆分文档对象
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000, chunk_overlap=200, add_start_index=True
    )
    all_splits = text_splitter.split_documents(docs)
    print(f"文档分割成 {len(all_splits)} 个chunks")

    # 获取嵌入维度
    embedding_dim = len(embeddings.embed_query("Hello gemini"))
    print(f"嵌入维度: {embedding_dim}")
    
    # 创建 FAISS 索引
    index = faiss.IndexFlatL2(embedding_dim)

    # 创建向量存储
    vector_store = FAISS(
        embedding_function=embeddings,
        index=index,
        docstore=InMemoryDocstore(),
        index_to_docstore_id={},
    )

    # 将文档对象添加到向量存储
    print("正在创建向量嵌入...")
    vector_store.add_documents(all_splits)

    # 持久化存储向量数据库
    print("正在保存FAISS数据库到本地...")
    vector_store.save_local(str(faiss_folder))
    print("FAISS数据库创建并保存完成！")

# 定义检索器
@chain
def retriever(query: str) -> list[Document]:
    """检索与查询相关的文档"""
    return vector_store.similarity_search(query, k=3)

for document in retriever.invoke("What was Nike's revenue in 2023?"):
    print("--------------------------------")
    print(f"文档内容: {document.page_content}")

上述代码能构建基本的语义搜索引擎，但是在部分的语句搜索上表现不佳：例如 "What was Nike's revenue in 2023?" 这个问题，能检索到相关文档；但是 "When was Nike incorporated?"这个问题，搜索结果相关性不佳。在官方文档演示示例中，返回的结果中直接包含了 “revenue” 关键词，但是实践后获取的文档中并没有对应关键词。
就结果而言，models/embedding-001 这款嵌入模型，最大 token 为 2048，返回的向量维度为 768，处理能力优秀。但是实际表现可能与官方演示存在偏差。

阶段总结

说实话，LangChain 的学习曲线有点陡峭，尤其是 StateGraph 的概念和实现方式。StateGraph 通过节点和函数的映射来处理状态流转，这种方式在复杂应用中非常有用，但也需要对数据流转有清晰的理解。
但是，LangChain 作为一个快速 LLM 应用开发框架，对于 AI 服务的封装有些过度，导致开发者对于 AI 服务的自定义能力和控制能力受限，像是使用非主流 AI 服务时，可能需要手动封装一些接口，或者使用 runnable 接口来实现自定义的调用方式。

***记于 2025/7/10：***目前还是不打算深入研究 LangChain，其内部的实现过于复杂，并且目前热门的开源项目多不以 LangChain 为基础。还是先以研究各种开源项目使用 AI 的方式为主，再决定是否深入研究 LangChain。