LangChain 核心解析：从 LLM 调用到 Agent 架构

如果说 LangChain 是一套大模型应用框架，那么 LCEL（LangChain Expression Language）就是它的核心编排能力。

很多人学 LangChain 只关注模型调用和 RAG，但真正决定工程质量的，是 LCEL。

6.1 为什么需要 LCEL？

假设我们要做一个最简单的流程：

1. 接收用户输入
2. 构造 Prompt
3. 调用 LLM
4. 解析输出

传统写法是'按步骤调用函数'。而 LCEL 的思路是用'管道'把各个组件连接起来，描述数据如何流动。

6.2 LCEL 的核心写法

在 LCEL 里：

• Prompt 是一个可运行组件
• LLM 是一个可运行组件
• Parser 也是一个可运行组件

它们都实现了统一接口，所以可以用 | 连接：

chain = prompt | llm | StrOutputParser()

调用时：

chain.invoke({"topic": "程序员"})

这行代码的含义非常清晰：

输入 → Prompt → LLM → 解析输出

比传统'手动 format + invoke + parse'，逻辑更直观、结构更干净。

6.3 LCEL 的优势在哪里？

1. 自动支持流式输出 同一条链可以直接流式执行：

for chunk in chain.stream(input):
    print(chunk)

无需改内部逻辑。

2. 自动支持异步 同步调用：

chain.invoke()

异步调用：

await chain.ainvoke()

流程不需要重写。

3. 自动并行执行 在 RAG 场景中，如果有多个 Retriever：

{"context1": retriever1, "context2": retriever2, "question": RunnablePassthrough()} | prompt | llm

retriever1 和 retriever2 会自动并行执行。不需要写 asyncio，也不需要手动调度。

6.4 一句话理解 LCEL

普通写法是：按顺序写函数调用 LCEL 写法是：声明一条数据流管道

它把'步骤式代码'升级为'数据流架构'。这也是为什么说：LCEL 是 LangChain 的灵魂。

小结

1. 如果你只是调用模型，你只是'在用 API'。
2. 如果你掌握 LCEL，你是在'设计 AI 工作流'。
3. 建议所有做 RAG 或 Agent 的开发者，都优先用 LCEL 组织流程。
4. 它会让你的代码更清晰、更可维护，也更适合生产环境。

七、对话历史管理：控制 Token 成本

真实业务中，多轮对话会带来一个问题：上下文越来越长 → Token 成本暴涨。

LangChain 提供：

• 历史裁剪（保留最近消息）
• 类型过滤（只保留 human）
• 持久化存储（SQL / Redis）

在企业场景里，这一点非常关键：

• 控制成本
• 避免超过模型上下文长度
• 保证系统稳定性

八、Agent：让模型自己思考与行动

如果说 RAG 是'查资料回答问题'，那么 Agent 是给一个目标，让模型自己规划步骤。

最经典的是 ReAct 模式：思考 → 行动 → 观察 → 再思考 → 输出结果。

举个例子：问'2024 年某明星演唱会是星期几？'

模型会：

1. 搜索日期
2. 解析日期
3. 调用'星期计算工具'
4. 输出答案

整个过程自动完成。这就是 Agent 的价值 —— 让模型具备'任务执行能力'。

九、LangServe：快速部署 API

当你构建好一个 LCEL 流程后，LangChain 提供 LangServe，可以一行代码生成 REST API。

你不需要写路由、序列化、异常处理逻辑。直接部署成可调用服务。在微服务架构下，这非常适合快速构建 AI 中台。

十、文档

• 功能模块：https://python.langchain.com/docs/get_started/introduction
• API 文档：https://api.python.langchain.com/en/latest/langchain_api_reference.html
• 三方组件集成：https://python.langchain.com/docs/integrations/platforms/
• 官方应用案例：https://python.langchain.com/docs/use_cases
• 调试部署等指导：https://python.langchain.com/docs/guides/debugging

十一、LangChain 适合什么场景？

• 企业知识库问答系统
• AI 客服系统
• 智能文档助手
• 代码生成平台
• 多工具自动执行系统
• 内部数据分析 Agent

不适合：

• 只做简单聊天（直接调用 API 更轻量）
• 极端高性能推理系统（需自定义架构）

十二、企业级最佳实践

结合实际项目经验，总结几点：

1. Prompt 必须独立管理 不要写死在代码里。

2. RAG 参数要反复调优 chunk_size、overlap 直接影响效果。

3. 尽量使用 LCEL 不要回到命令式'胶水代码'。

4. 严格做日志追踪 生产环境一定接入 LangSmith。

5. 版本锁定 LangChain 更新快，升级前一定看 breaking change。

十三、总结

LangChain 并不是'又一个 AI 库'。它是一次面向大模型时代的软件工程抽象升级。

它告诉我们三件事：

1. 把 Prompt 当函数
2. 把流程当管道
3. 把模型当可调度执行器

理解这三点，比会写 API 更重要。

如果你准备入门 AI 工程化开发，建议路线：

1. 先做一个 RAG 问答系统
2. 加入对话历史管理
3. 用 LCEL 重构流程
4. 最后引入 Agent 能力

当你走完这条路径，你对大模型应用开发的理解会完全不同。

LangChain RAG 示例源码：

from dotenv import load_dotenv
import os
import requests
import glob

# Load environment variables from .env file
load_dotenv()

from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance
from langchain_community.document_loaders import PyMuPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_qdrant import QdrantVectorStore
from langchain_core.documents import Document

EMBEDDING_DIM = 384  # all-MiniLM-L6-v2 模型的维度
COLLECTION_NAME = "langchain_demo"
PATH = "./qdrant_db_langchain"

# 设置 Tarvos API key 和 URL（从环境变量读取）
TARVOS_API_KEY = os.getenv("TARVOS_API_KEY")
TARVOS_API_URL = os.getenv("TARVOS_API_URL")
TARVOS_MODEL = os.getenv("TARVOS_MODEL", "meta-llama/Llama-3.3-70B-Instruct")

# 1. 加载本地 embedding 模型
print("正在加载本地 embedding 模型...")
embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")

# 2. 加载 ./data 目录下所有 PDF 文档
print("正在加载文档...")
pdf_files = glob.glob("./data/*.pdf")
if not pdf_files:
    raise ValueError("./data 目录下没有找到 PDF 文件，请先添加 PDF 文件")
all_documents = []
for pdf_path in pdf_files:
    loader = PyMuPDFLoader(pdf_path)
    docs = loader.load()
    all_documents.extend(docs)
print(f"共加载 {len(all_documents)} 页文档")

# 3. 文档切片（chunk_size=300, chunk_overlap=100，与 llamaIndex.py 一致）
print("正在切片文档...")
text_splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=100)
chunks = text_splitter.split_documents(all_documents)
print(f"共切分为 {len(chunks)} 个块")

# 4. 创建 Qdrant Collection
print("正在创建向量数据库...")
qdrant_client = QdrantClient(path=PATH)
if qdrant_client.collection_exists(collection_name=COLLECTION_NAME):
    qdrant_client.delete_collection(collection_name=COLLECTION_NAME)
qdrant_client.create_collection(
    collection_name=COLLECTION_NAME,
    vectors_config=VectorParams(size=EMBEDDING_DIM, distance=Distance.COSINE)
)

# 5. 将文档向量存入 Qdrant
print("正在创建索引...")
vector_store = QdrantVectorStore(
    client=qdrant_client,
    collection_name=COLLECTION_NAME,
    embedding=embeddings
)
vector_store.add_documents(chunks)

# 6. 创建检索器
retriever = vector_store.as_retriever(search_kwargs={"k": 3})

# 7. Tarvos API 问答函数（与 llamaIndex.py 完全一致）
def ask_tarvos(question, context):
    headers = {
        "Content-Type": "application/json",
        "Authorization": TARVOS_API_KEY,
    }
    body = {
        "model": TARVOS_MODEL,
        "messages": [
            {
                "role": "system",
                "content": "你是一个专业的 AI 助手，请根据提供的检索内容回答用户问题。如果检索内容与问题无关，请说明无法回答。"
            },
            {
                "role": "user",
                "content": f"参考内容:\n{context}\n\n问题：{question}"
            }
        ]
    }
    try:
        resp = requests.post(TARVOS_API_URL, json=body, headers=headers, timeout=60)
        return resp.json()["choices"][0]["message"]["content"]
    except Exception as e:
        return f"调用 Tarvos API 失败：{str(e)}"

# 8. 主对话循环
print("\n✅ 系统就绪！开始对话（输入空行退出）\n")
while True:
    question = input("User: ")
    if question.strip() == "":
        break
    # 检索 top-k 文档
    docs = retriever.invoke(question)
    context = "\n\n".join([doc.page_content for doc in docs])
    # 调用 Tarvos API 生成答案
    answer = ask_tarvos(question, context)
    print(f"AI: {answer}\n")

LangChain RAG 示例架构图示：