基本介绍 | A-Zero

# 基本介绍

官网 (opens new window) 中文网 (opens new window)

核心功能：

开发(Development) # 通过开源组件和第三方集成构建 LLM 应用
生产化(Productionization) # 使用 LangSmith 监控、调试和评估应用
部署(Deployment) # 通过 LangGraph Platform 将应用部署为 API 或助手服务

关键工具链:

langchain-core # 核心接口(如 Runnable)、基础抽象和LangChain表达式语言
langchain-community # 第三方集成包（例如langchain-openai，langchain-anthropic等）
langgraph # 通过将步骤建模为图中的边和节点，使用 LLM 构建健壮且有状态的多角色应用程序
LangSmith # 追踪与评估
Langserve # 部署为 REST API

# 搭建 RAG

RAG 是如何检索的

向量化 # 使用 Embedding 模型将文本转为向量
相似度计算 # 使用余弦相似度（cosine similarity）在向量数据库查找最接近的文档
返回 Top-K # 返回最相关的K的文档作为上下文

RAG 是如何生成答案的

# 将用户问题+检索到的上下文一起拼接成prompt
# 输入给LLM进行推理
# LLM仅基于上下文作答，避免幻觉

技术栈选型

文本加载 # PyPDF2/Unstructured
分块工具 # RecursiveCharacterTextSplitter
Embedding # text-embedding-ada-002/Qwen-embedding
向量数据库 # Chroma/FAISS/PostgreSQL+pgvector（本地）/Pinecone（云端）/ Milvus
大模型 # Qwen-30B/Qwen-32B
部署框架 # FastAPI+Uvicorn

核心流程

数据摄取 # 加载 PDF/Word/ 网页等格式文件
文本分块 # 将长文本拆分为 500-1000 字的语义片段
向量索引 # 通过 Embedding 模型生成向量
向量存储 # 向量存储到数据库
检索增强 # 根据用户问题检索相关文档片段
LLM选择 # 配置大语言模型
生成优化 # 将检索结果与问题结合生成最终回答

# 数据摄取

因为数据源是多种多样的，langchain 中使用 Loader 对各种数据源进行支持。一些常用的：

pip install pypdf
from langchain_community.document_loaders import PyPDFLoader 

from langchain_community.document_loaders import TextLoader

from langchain_community.document_loaders import CSVLoader
# 进行简单的文本提取，不考虑文档结构，适用于只需纯文本内容的场景
pip install docx2txt
from langchain_community.document_loaders import Docx2txtLoader
# 能识别文档结构（如标题、段落、列表），将其转换为带有元数据的文本块，便于后续处理
pip install unstructured
pip install python-docx
from langchain_community.document_loaders import UnstructuredWordDocumentLoader

from langchain_community.document_loaders import UnstructuredMarkdownLoader
from langchain_community.document_loaders import UnstructuredHTMLLoader
from langchain_community.document_loaders import JSONLoader
from langchain_community.document_loaders import WebBaseLoader

# 加载PDF文件
# 需要安装 pip install PyPDF PyPDF2
loader = PyPDFLoader("data/report.pdf")
documents = loader.load()
 
# 加载本地文本文件
loader = TextLoader("data/faq.txt")
loader = TextLoader("data/faq.md")
documents += loader.load()
 
# 加载网页内容
loader = WebBaseLoader("https://example.com")
documents += loader.load()

# 读取到的数据结构是 Document 对象
from langchain_core.documents import Document

非结构化文档加载 UnstructuredWordDocumentLoader 会使用 nltk

需要把【packages】文件夹名换成：【nltk_data】，放在【/root/nltk_data】
然后把【tokenizers】文件夹下的【punkt.zip】进行解压，才可能使用

# 文本分块

七种不同的字符文本分割 TextSplitter

from langchain_text_splitters import CharacterTextSplitter
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_text_splitters import TokenTextSplitter
from langchain_text_splitters import SpacyTextSplitter
from langchain_text_splitters import SentenceTransformersTokenTextSplitter
from langchain_text_splitters import NLTKTextSplitter
from langchain_text_splitters import KonlpyTextSplitter

# pip install langchain_text_splitters
# RecursiveCharacterTextSplitter 使用
# chunk_size 指定字符串和，重叠文本的长度
text_splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=50)

# CharacterTextSplitter的 使用，用到 tiktoken_encoder
text_splitter = CharacterTextSplitter.from_tiktoken_encoder(
    encoding_name="cl100k_base",
	chunk_size=100, # chunk_size 指定字符串和
	chunk_overlap=0 # 重叠文本的长度
)

# document
splits = text_splitter.split_documents(docs)

# text
texts = text_splitter.split_text(document)

# 向量索引

查看 langchain 支持 embeddings 模型 (opens new window)

使用 OpenAI 的 embeddings + InMemoryVectorStore

from langchain_openai import OpenAIEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore

embeddings_model = OpenAIEmbeddings()
vector_store = InMemoryVectorStore(embedding=SomeEmbeddingModel())
# 向向量数据库添加文档
vector_store.add_documents(splits)

使用 ZhipuAI 的 embeddings + FAISS

from langchain_community.embeddings import ZhipuAIEmbeddings
from langchain_community.vectorstores import FAISS
import dotenv

class ZhipuEmbeddings():
    api_key = dotenv.get_key(".env", "api_key")

    @classmethod
    def create_zhipu_embeddings(self, model_name):
        embeddings = ZhipuAIEmbeddings(
            model=model_name,
            api_key=self.api_key
        )
        return embeddings
		
vector_store = FAISS.from_documents(splits, ZhipuEmbeddings())

使用 OllamaEmbeddings + Chroma

from langchain_community.embeddings import OllamaEmbeddings
from langchain_community.vectorstores import Chroma

embeddings = OllamaEmbeddings(model="llama2")
vector_store = Chroma.from_documents(splits, embeddings)

使用 Qwen3 Embeddig + Milvus

from langchain_community.vectorstores import Milvus

# 需要安装 pip install pymilvus
URI = "./milvus_example.db"
vector_store = Milvus(
    embedding_function=embeddings,
    connection_args={"uri": URI},
)

# 向向量数据库添加文档
vector_store.add_documents(splits)

# 向量存储

向量数据 VectorStore 的核心方法：

add_texts(texts, metadatas) # 将文本及其元数据嵌入并存储到 VectorStore
similarity_search(query, k) # 基于查询向量，返回最相似的 k 个结果
similarity_search_with_score # 使用距离运行相似性搜索
delete(ids) # 根据指定的 ID 删除向量
from_documents(documents) # 从文档列表构建向量存储
save_local(path) 和 load_local(path) # 将向量存储保存到本地或从本地加载

具体使用，更适合直接的向量相似度搜索需求

query = "主要业务？"
# 搜索向量
docs = vectorstore.similarity_search(query)

# 检索增强

将 VectorStore 转换为检索器，更适合构建复杂的 RAG 应用，可以方便地更换不同的检索策略

# 转成检索器，配置检索参数
retriever = vector_store.as_retriever(
              search_type="similarity",
			  # search_type="similarity_score_threshold",
              search_kwargs={"k": 5, "score_threshold": 0.7}
            )
# 检索
retrieved_docs = retriever.invoke("主要业务？")

# LLM选择

本地模型部署

from langchain_community.llms.ollama import Ollama
 
llm2 = Ollama(
    base_url="http://localhost:11434",
    model="models/qwen-1.8-chat-int4.bin",
    temperature=0.6,
)

云端模型调用

from langchain_openai import ChatOpenAI
 
load_dotenv()
llm = ChatOpenAI(
    openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",
    openai_api_key=os.getenv("DASHSCOPE_API_KEY"),
    model_name="qwen-plus",
    temperature=0.7
)

# 生成优化

使用 Langchian 的 chain 组合上传文档生成内容

retriever 做来 context 的值传递给 chain 后面的内容

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 提示词模板
prompt_template = """
请根据以下上下文回答问题：
上下文：{context}
问题：{question}
答案：
"""
prompt = ChatPromptTemplate.from_template(prompt_template)

def format_docs(documents):
    return "\n\n".join(doc.page_content for doc in documents)

# RunnablePassthrough：将原始输入原样传递给下一步
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
	| StrOutputParser()
)

使用 create_retrieval_chain api 来创建一个 rag_chain

from langchain.chains import create_retrieval_chain

document_chain = create_stuff_documents_chain(llm, prompt)
rag_chain = create_retrieval_chain(retriever, document_chain)

# 问题测试

# .invoke(input) 单次调用
# .batch(inputs) 批量调用（并发）
# .stream(input) 流式输出
# .ainvoke(input)/abatch(inputs) 异步调用
response = rag_chain.invoke("主要业务？")
print(response)

# 测试多个问题
questions = [
    "主要业务？",
    "产品介绍？",
    "客户案例？"
]
for question in questions:
    response = rag_chain.invoke(question)
    print(f"问题: {question}\n答案: {response}\n")

# Langgraph

LangGraph 是一个强大的 Agent 开发框架，它让复杂的多步骤推理变得更加可控和可靠

← 基础知识