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实战。