Skip to content

LangChain基础概念

LangChain是开发大模型应用最受欢迎的框架,它提供了一套标准化的组件和抽象,让你能够快速构建复杂的AI应用。

什么是LangChain?

LangChain是一个开源框架,用于开发由大语言模型驱动的应用程序。它的核心理念是:

大模型 + 外部数据 + 工具调用 = 强大的AI应用

为什么需要LangChain?

直接调用API开发AI应用有几个痛点:

痛点LangChain的解决方案
API调用分散统一的模型接口
提示词难以管理PromptTemplate模板化
无法记住上下文Memory组件管理历史
无法调用外部工具Tools和Agents机制
数据无法注入RAG管道支持

核心组件架构

┌─────────────────────────────────────────────┐
│                  Application                │
├─────────────────────────────────────────────┤
│    Chains    │   Agents   │   Memory       │
├─────────────────────────────────────────────┤
│              LLM / Chat Models              │
├─────────────────────────────────────────────┤
│  Prompts  │  Output Parsers  │  Documents  │
├─────────────────────────────────────────────┤
│              Vector Stores                  │
└─────────────────────────────────────────────┘

快速开始

安装

bash
pip install langchain langchain-openai

第一个应用

python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 1. 定义模型
model = ChatOpenAI(model="gpt-4o")

# 2. 定义提示词模板
prompt = ChatPromptTemplate.from_template("给我讲一个关于{topic}的笑话")

# 3. 定义输出解析器
parser = StrOutputParser()

# 4. 构建链
chain = prompt | model | parser

# 5. 执行
result = chain.invoke({"topic": "程序员"})
print(result)

核心概念详解

1. Models(模型)

LangChain支持多种模型类型:

python
from langchain_openai import ChatOpenAI, OpenAI
from langchain_anthropic import ChatAnthropic

# Chat模型(推荐)
chat_model = ChatOpenAI(model="gpt-4o")
chat_model = ChatAnthropic(model="claude-sonnet-4-20250514")

# 文本补全模型(旧版)
llm = OpenAI(model="gpt-3.5-turbo-instruct")

2. Prompts(提示词)

python
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate

# Chat模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位{role}"),
    ("user", "{input}")
])

# 简单模板
simple_prompt = PromptTemplate.from_template("翻译成英文:{text}")

# 使用
formatted = prompt.invoke({"role": "翻译专家", "input": "你好"})

3. Output Parsers(输出解析器)

python
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from pydantic import BaseModel

# 字符串解析
str_parser = StrOutputParser()

# JSON解析
class Person(BaseModel):
    name: str
    age: int

json_parser = JsonOutputParser(pydantic_object=Person)

# 带格式说明的提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "提取人物信息"),
    ("user", "{input}"),
    ("system", "{format_instructions}")
]).partial(format_instructions=json_parser.get_format_instructions())

4. Chains(链)

链是LangChain的核心概念,用于组合多个组件:

python
from langchain_core.runnables import RunnablePassthrough

# 简单链
chain = prompt | model | parser

# 带数据处理的链
chain = {
    "context": retriever,  # 从向量库检索
    "question": RunnablePassthrough()
} | prompt | model | parser

# 条件分支链
from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: x["type"] == "math", math_chain),
    (lambda x: x["type"] == "code", code_chain),
    default_chain
)

5. Memory(记忆)

让AI记住对话历史:

python
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.chat_history import InMemoryChatMessageHistory

# 简单记忆
history = InMemoryChatMessageHistory()
history.add_user_message("你好")
history.add_ai_message("你好!有什么可以帮你的?")

# 在链中使用
from langchain_core.runnables import RunnableWithMessageHistory

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history=lambda session_id: history,
    input_messages_key="input",
    history_messages_key="chat_history"
)

LCEL(LangChain Expression Language)

LCEL是LangChain的声明式语法,让代码更简洁:

python
# 传统写法
def process(input_text):
    formatted = prompt.format(topic=input_text)
    response = model.invoke(formatted)
    parsed = parser.parse(response)
    return parsed

# LCEL写法
chain = prompt | model | parser
result = chain.invoke({"topic": "Python"})

LCEL操作符

操作符说明示例
``管道连接
{}并行执行{"a": chain1, "b": chain2}
RunnablePassthrough透传数据{"q": RunnablePassthrough()}
RunnableParallel并行运行RunnableParallel(a=chain1, b=chain2)

流式输出

python
# 流式处理
for chunk in chain.stream({"topic": "AI"}):
    print(chunk, end="", flush=True)

# 异步流式
async for chunk in chain.astream({"topic": "AI"}):
    print(chunk, end="", flush=True)

错误处理

python
from langchain_core.runnables import RunnableRetry

# 自动重试
chain_with_retry = chain.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True
)

# 降级处理
from langchain_core.runnables import RunnableFallback

chain_with_fallback = chain.with_fallbacks([
    backup_chain1,
    backup_chain2
])

小结

LangChain核心概念:

概念说明
Models大模型的统一接口
Prompts提示词模板化
Parsers结构化输出解析
Chains组件组合管道
Memory对话历史管理
LCEL声明式语法

下一步

了解了基础概念后,继续学习 Chain与Agent实战