LangChain 框架实战

文章摘要 FakeGPT

加载中...|

此内容根据文章生成，并经过人工审核，仅用于文章内容的解释与总结投诉

概述

LangChain 是目前最流行的 LLM 应用开发框架，它提供了丰富的组件和工具，帮助开发者快速构建复杂的 AI 应用。本文将深入讲解 LangChain 的核心概念和实战技巧。

LangChain 架构概述

核心组件

text

┌─────────────────────────────────────────────────────────┐
│                    LangChain 架构                         │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  ┌─────────────────────────────────────────────────┐   │
│  │              Models I/O                         │   │
│  │   • LLMs (OpenAI, Anthropic, etc.)              │   │
│  │   • Chat Models                                 │   │
│  │   • Embeddings                                  │   │
│  └─────────────────────────────────────────────────┘   │
│                          │                              │
│                          ▼                              │
│  ┌─────────────────────────────────────────────────┐   │
│  │              Prompts                             │   │
│  │   • Prompt Templates                            │   │
│  │   • Output Parsers                              │   │
│  │   • Example Selectors                           │   │
│  └─────────────────────────────────────────────────┘   │
│                          │                              │
│                          ▼                              │
│  ┌─────────────────────────────────────────────────┐   │
│  │              Chains                             │   │
│  │   • LCEL (LangChain Expression Language)        │   │
│  │   • Pre-built Chains                            │   │
│  │   • Sequential / Router Chains                  │   │
│  └─────────────────────────────────────────────────┘   │
│                          │                              │
│                          ▼                              │
│  ┌─────────────────────────────────────────────────┐   │
│  │              Agents                             │   │
│  │   • Agent Types (ReAct, OpenAI Functions)       │   │
│  │   • Tools                                       │   │
│  │   • Agent Executors                             │   │
│  └─────────────────────────────────────────────────┘   │
│                          │                              │
│                          ▼                              │
│  ┌─────────────────────────────────────────────────┐   │
│  │              Memory                             │   │
│  │   • Memory Types                                │   │
│  │   • Storage                                     │   │
│  │   • Retriever                                   │   │
│  └─────────────────────────────────────────────────┘   │
│                                                         │
└─────────────────────────────────────────────────────────┘

安装与配置

bash

# 基础安装
pip install langchain langchain-openai

# 可选组件
pip install langchain-community  # 社区集成
pip install langchain-anthropic  # Anthropic
pip install langchain-google     # Google
pip install langchain-aws        # AWS

# 依赖工具
pip install chromadb             # 向量数据库
pip install beautifulsoup4       # 网页抓取
pip install faiss-cpu            # 向量搜索

python

# 配置 API Key
import os
os.environ["OPENAI_API_KEY"] = "your-api-key"

# 或使用 dotenv
from dotenv import load_dotenv
load_dotenv()

Models：模型接口

LLM vs Chat Model

python

from langchain_openai import OpenAI, ChatOpenAI

# LLM：文本输入 → 文本输出
llm = OpenAI(model="gpt-4o", temperature=0)
response = llm.invoke("什么是 LangChain？")
print(response)

# Chat Model：消息列表 → 消息输出
chat = ChatOpenAI(model="gpt-4o", temperature=0)
from langchain_core.messages import HumanMessage

messages = [HumanMessage(content="什么是 LangChain？")]
response = chat.invoke(messages)
print(response.content)

消息类型

python

from langchain_core.messages import (
    HumanMessage,
    AIMessage,
    SystemMessage,
    ToolMessage
)

# 系统消息：设定行为
system_msg = SystemMessage(content="你是一个专业的Python程序员")

# 用户消息
user_msg = HumanMessage(content="帮我写一个快排算法")

# AI 消息
ai_msg = AIMessage(content="这是快排算法的Python实现...")

# 工具消息（Function Calling 结果）
tool_msg = ToolMessage(
    content="工具执行结果",
    tool_call_id="tool_123"
)

# 组合使用
messages = [
    system_msg,
    user_msg,
    ai_msg
]

response = chat.invoke(messages)

流式输出

python

# 流式输出
for chunk in chat.stream("讲个短故事"):
    print(chunk.content, end="", flush=True)

# 异步流式
import asyncio

async def stream_example():
    async for chunk in chat.astream("讲个短故事"):
        print(chunk.content, end="", flush=True)

asyncio.run(stream_example())

Embeddings

python

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 单个文本
vector = embeddings.embed_query("Hello World")
print(f"维度: {len(vector)}")

# 批量文本
texts = ["Hello", "World", "LangChain"]
vectors = embeddings.embed_documents(texts)
print(f"向量数量: {len(vectors)}")

Prompts：提示管理

Prompt Template

python

from langchain.prompts import PromptTemplate, ChatPromptTemplate
from langchain.prompts import (
    HumanMessagePromptTemplate,
    SystemMessagePromptTemplate
)

# 基础模板
template = "请用{language}写一个{topic}的示例。"
prompt = PromptTemplate(
    template=template,
    input_variables=["language", "topic"]
)

# 生成提示
prompt_text = prompt.format(language="Python", topic="for循环")
print(prompt_text)
# 输出：请用Python写一个for循环的示例。

# Chat Prompt Template
system_template = "你是一个专业的{role}。"
system_prompt = SystemMessagePromptTemplate.from_template(system_template)

human_template = "{task}"
human_prompt = HumanMessagePromptTemplate.from_template(human_template)

chat_prompt = ChatPromptTemplate.from_messages([
    system_prompt,
    human_prompt
])

# 生成消息
messages = chat_prompt.format_messages(
    role="Python程序员",
    task="写一个快排算法"
)

少样本学习（Few-shot）

python

from langchain.prompts import FewShotPromptTemplate
from langchain.prompts.prompt import PromptTemplate

# 定义示例
examples = [
    {
        "question": "1 + 1 = ?",
        "answer": "2"
    },
    {
        "question": "5 × 6 = ?",
        "answer": "30"
    },
    {
        "question": "10 - 3 = ?",
        "answer": "7"
    }
]

# 示例模板
example_prompt = PromptTemplate(
    input_variables=["question", "answer"],
    template="问题: {question}\n答案: {answer}"
)

# Few-shot 提示模板
prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
    prefix="请按照以下格式回答问题：\n",
    suffix="\n问题: {input}\n答案:",
    input_variables=["input"]
)

# 使用
final_prompt = prompt.format(input="8 ÷ 2 = ?")
print(final_prompt)

输出解析器

python

from langchain.output_parsers import (
    StructuredOutputParser,
    ResponseSchema,
    CommaSeparatedListOutputParser
)

# 结构化输出
response_schemas = [
    ResponseSchema(
        name="summary",
        description="文本摘要，不超过50字"
    ),
    ResponseSchema(
        name="keywords",
        description="关键词列表，用逗号分隔"
    ),
    ResponseSchema(
        name="sentiment",
        description="情感倾向：positive/negative/neutral",
        type="string"
    )
]

output_parser = StructuredOutputParser.from_response_schemas(response_schemas)

# 获取格式说明
format_instructions = output_parser.get_format_instructions()

# 创建提示
prompt = PromptTemplate(
    template="分析以下文本：\n{text}\n\n{format_instructions}",
    input_variables=["text"],
    partial_variables={"format_instructions": format_instructions}
)

# 解析输出
response = llm.invoke(prompt.format(text="LangChain太棒了！"))
parsed_output = output_parser.parse(response)
print(parsed_output)
# {'summary': '...', 'keywords': '...', 'sentiment': 'positive'}

# 列表输出
list_parser = CommaSeparatedListOutputParser()
list_format_instructions = list_parser.get_format_instructions()

prompt = PromptTemplate(
    template="列出{topic}的5个特点：\n\n{format_instructions}",
    input_variables=["topic"],
    partial_variables={"format_instructions": list_format_instructions}
)

response = llm.invoke(prompt.format(topic="Python"))
parsed_list = list_parser.parse(response)
print(parsed_list)  # ['简单', '强大', '免费', '社区大', '应用广']

Chains：链式调用

LCEL（LangChain Expression Language）

python

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

# 使用 LCEL 构建链
llm = ChatOpenAI(model="gpt-4o")
prompt = ChatPromptTemplate.from_template("讲一个关于{topic}的笑话")
output_parser = StrOutputParser()

# 使用管道操作符 (|)
chain = prompt | llm | output_parser

# 调用
result = chain.invoke({"topic": "程序员"})
print(result)

# 流式输出
for chunk in chain.stream({"topic": "程序员"}):
    print(chunk, end="", flush=True)

# 批量处理
results = chain.batch([
    {"topic": "程序员"},
    {"topic": "产品经理"}
])

链的组合

python

from langchain_core.runnables import RunnablePassthrough, RunnableLambda

# 子链1：摘要
summary_prompt = ChatPromptTemplate.from_template("摘要以下文本：\n{text}")
summary_chain = summary_prompt | llm | StrOutputParser()

# 子链2：翻译
translate_prompt = ChatPromptTemplate.from_template("翻译成英文：\n{text}")
translate_chain = translate_prompt | llm | StrOutputParser()

# 组合链
def prepare_input(x):
    return {"text": x}

full_chain = (
    RunnablePassthrough() |
    {
        "summary": summary_chain,
        "translation": translate_chain
    }
)

result = full_chain.invoke("LangChain是一个强大的LLM应用开发框架")
print(result)
# {'summary': '...', 'translation': '...'}

顺序链

python

from langchain.chains import SequentialChain

# 链1：生成故事
story_prompt = ChatPromptTemplate.from_template(
    "讲一个关于{topic}的{genre}故事，不超过100字。"
)
story_chain = story_prompt | llm | StrOutputParser()

# 链2：提取主角
character_prompt = ChatPromptTemplate.from_template(
    "从以下故事中提取主角名字：\n{story}\n只返回名字。"
)
character_chain = character_prompt | llm | StrOutputParser()

# 组合
overall_chain = {
    "story": story_chain,
    "topic": lambda x: x["topic"],
    "genre": lambda x: x["genre"]
} | {
    "story": RunnablePassthrough(),
    "character": character_chain
}

result = overall_chain.invoke({"topic": "太空", "genre": "科幻"})
print(result)

路由链

python

from langchain.chains import create_tagging_chain
from langchain.chains.router import MultiRouteChain
from langchain.chains.router.llm_router import LLMRouterChain, RouterOutputParser
from langchain.prompts import PromptTemplate

# 定义路由
router_template = """根据用户问题，选择合适的处理链：

用户问题：{input}

可选项：
- code：编程相关问题
- general：一般性问题

只返回选项名称。"""

router_prompt = PromptTemplate(
    template=router_template,
    input_variables=["input"]
)

# 定义不同链
code_chain = PromptTemplate.from_template("用Python解决：{input}") | llm | StrOutputParser()
general_chain = PromptTemplate.from_template("回答：{input}") | llm | StrOutputParser()

# 路由链
from langchain.chains.router import RouterChain

routes = {
    "code": code_chain,
    "general": general_chain
}

# 使用
def route_question(inputs):
    question = inputs["input"]
    # 简化路由逻辑
    if any(kw in question.lower() for kw in ["代码", "程序", "python", "函数"]):
        return "code"
    return "general"

from langchain_core.runnables import RunnableLambda

router_chain = RunnableLambda(lambda x: routes[route_question(x)])

# 完整链
full_chain = {
    "input": lambda x: x
} | router_chain

result = full_chain.invoke({"input": "怎么用Python读文件？"})
print(result)

Agents：智能代理

ReAct Agent

python

from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import Tool
from langchain import hub

# 定义工具
def search(query: str) -> str:
    """搜索工具"""
    return f"关于'{query}'的搜索结果..."

def calculator(expression: str) -> str:
    """计算器"""
    try:
        result = eval(expression)
        return f"计算结果: {result}"
    except:
        return "计算错误"

tools = [
    Tool(
        name="Search",
        func=search,
        description="用于搜索信息的工具。输入应该是搜索查询。"
    ),
    Tool(
        name="Calculator",
        func=calculator,
        description="用于数学计算。输入应该是数学表达式。"
    )
]

# 获取 ReAct 提示模板
prompt = hub.pull("hwchase17/react")

# 创建 agent
llm = ChatOpenAI(model="gpt-4o", temperature=0)
agent = create_react_agent(llm, tools, prompt)

# 创建执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    handle_parsing_errors=True,
    max_iterations=5
)

# 执行
result = agent_executor.invoke({
    "input": "苹果公司现在的股价是多少？如果我有100股，价值多少美元？"
})
print(result["output"])

OpenAI Functions Agent

python

from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain.tools import tool
from langchain_core.utils.function_calling import convert_to_openai_function

@tool
def search(query: str) -> str:
    """搜索最新信息"""
    return f"关于'{query}'的搜索结果：苹果股价 $189.50"

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    return str(eval(expression))

tools = [search, calculate]

# 创建 agent
llm = ChatOpenAI(model="gpt-4o")
agent = create_openai_functions_agent(llm, tools)

# 执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True
)

# 执行
result = agent_executor.invoke({
    "input": "苹果股价现在是多少？100股价值多少？"
})

自定义工具

python

from langchain.tools import StructuredTool
from pydantic import BaseModel, Field

# 定义输入 schema
class CalculatorInput(BaseModel):
    expression: str = Field(description="要计算的数学表达式")

# 定义函数
def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"错误: {e}"

# 创建结构化工具
calc_tool = StructuredTool.from_function(
    func=calculator,
    name="Calculator",
    description="执行数学计算",
    args_schema=CalculatorInput
)

# 使用 LangChain 的装饰器（更简洁）
from langchain.tools import tool

@tool
def advanced_calculator(
    expression: str,
    precision: int = Field(default=2, description="小数点后位数")
) -> str:
    """
    高级计算器，可以指定精度。

    Args:
        expression: 数学表达式
        precision: 小数点后位数

    Returns:
        计算结果
    """
    try:
        result = eval(expression)
        return f"{result:.{precision}f}"
    except Exception as e:
        return f"计算错误: {e}"

# 查看工具的 OpenAI 函数格式
print(advanced_calculator.args_schema.schema())

Memory：记忆管理

对话缓冲记忆

python

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

# 缓冲记忆
memory = ConversationBufferMemory()

conversation = ConversationChain(
    llm=ChatOpenAI(model="gpt-4o"),
    memory=memory,
    verbose=True
)

# 对话
conversation.predict(input="我叫小明")
conversation.predict(input="我叫什么名字？")

# 查看记忆
print(memory.buffer)
print(memory.load_memory_variables({}))

窗口记忆

python

from langchain.memory import ConversationBufferWindowMemory

# 只保留最近 3 轮对话
memory = ConversationBufferWindowMemory(k=3)

conversation = ConversationChain(
    llm=ChatOpenAI(model="gpt-4o"),
    memory=memory
)

# 多轮对话
for i in range(5):
    conversation.predict(input=f"第{i+1}轮对话")

# 只保留最近3轮
print(memory.load_memory_variables({}))

摘要记忆

python

from langchain.memory import ConversationSummaryMemory

# 自动总结旧对话
memory = ConversationSummaryMemory(
    llm=ChatOpenAI(model="gpt-4o")
)

conversation = ConversationChain(
    llm=ChatOpenAI(model="gpt-4o"),
    memory=memory,
    verbose=True
)

# 长对话
conversation.predict(input="我想学Python")
conversation.predict(input="Python有哪些特点？")
conversation.predict(input="推荐一些学习资源")
conversation.predict(input="刚才我们聊了什么？")  # 会使用摘要

# 查看摘要
print(memory.moving_summary_buffer)

在链中使用记忆

python

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

memory = ConversationBufferMemory(return_messages=True)

chain = (
    RunnablePassthrough.assign(
        history=lambda x: memory.load_memory_variables(x)["history"]
    )
    | prompt
    | ChatOpenAI(model="gpt-4o")
)

# 保存记忆
def save_memory(input, output):
    memory.save_context({"input": input}, {"output": output.content})

# 使用
response1 = chain.invoke({"input": "我叫小明"})
save_memory("我叫小明", response1)

response2 = chain.invoke({"input": "我叫什么名字？"})
print(response2.content)

实战：构建完整 RAG 应用

python

from langchain_community.document_loaders import TextLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 1. 加载文档
loader = DirectoryLoader(
    "./docs",
    glob="**/*.md",
    loader_cls=TextLoader
)
documents = loader.load()

# 2. 分块
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
splits = text_splitter.split_documents(documents)

# 3. 创建向量存储
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=OpenAIEmbeddings()
)

# 4. 创建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}
)

# 5. 自定义提示
prompt_template = """基于以下文档回答问题。如果文档中没有相关信息，说"我不知道"。

文档:
{context}

问题: {question}

回答:"""

PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

# 6. 创建 RAG 链
llm = ChatOpenAI(model="gpt-4o", temperature=0)

def format_docs(docs):
    return "\n\n".join([d.page_content for d in docs])

rag_chain = (
    {
        "context": retriever | format_docs,
        "question": lambda x: x["question"]
    }
    | PROMPT
    | llm
    | StrOutputParser()
)

# 7. 查询
result = rag_chain.invoke({"question": "什么是 RAG？"})
print(result)

# 带来源的查询
from langchain.chains import RetrievalQAWithSourcesChain

qa_chain = RetrievalQAWithSourcesChain.from_chain_type(
    llm=ChatOpenAI(model="gpt-4o"),
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True
)

result = qa_chain({"question": "什么是 RAG？"})
print(result["answer"])
print("来源:", [d.metadata for d in result["source_documents"]])

小结

LangChain 是构建 LLM 应用的强大框架：

核心要点

Models
- 统一的模型接口
- 支持多种 LLM 提供商
- 流式输出和异步支持
Prompts
- 模板化管理
- 少样本学习
- 输出解析器
Chains
- LCEL 简化链式调用
- 支持复杂的组合逻辑
- 路由和条件分支
Agents
- ReAct 和 OpenAI Functions
- 自定义工具开发
- Agent 执行器配置
Memory
- 多种记忆类型
- 灵活的记忆管理
- 与链的集成

下一篇文章将介绍多 Agent 协作与规划，构建更复杂的 AI 系统。

LangChain 框架实战https://indulgeback.github.io/posts/AI%E4%B8%8ELLM/8%E3%80%81LangChain%20%E6%A1%86%E6%9E%B6%E5%AE%9E%E6%88%98

作者LeviLiu

发布于6天前

更新于2天前

许可协议 CC BY-NC-SA 4.0

署名-非商业性使用-相同方式共享 4.0 国际

LLM LangChain Agent 框架

反馈与投诉

赞赏博主

评论隐私政策

LangChain 框架实战

概述 ​

LangChain 架构概述 ​

核心组件 ​

安装与配置 ​

Models：模型接口 ​

LLM vs Chat Model ​

消息类型 ​

流式输出 ​

Embeddings ​

Prompts：提示管理 ​

Prompt Template ​

少样本学习（Few-shot） ​

输出解析器 ​

Chains：链式调用 ​

LCEL（LangChain Expression Language） ​

链的组合 ​

顺序链 ​

路由链 ​

Agents：智能代理 ​

ReAct Agent ​

OpenAI Functions Agent ​

自定义工具 ​

Memory：记忆管理 ​

对话缓冲记忆 ​

窗口记忆 ​

摘要记忆 ​

在链中使用记忆 ​

实战：构建完整 RAG 应用 ​

小结 ​

核心要点 ​

概述

LangChain 架构概述

核心组件

安装与配置

Models：模型接口

LLM vs Chat Model

消息类型

流式输出

Embeddings

Prompts：提示管理

Prompt Template

少样本学习（Few-shot）

输出解析器

Chains：链式调用

LCEL（LangChain Expression Language）

链的组合

顺序链

路由链

Agents：智能代理

ReAct Agent

OpenAI Functions Agent

自定义工具

Memory：记忆管理

对话缓冲记忆

窗口记忆

摘要记忆

在链中使用记忆

实战：构建完整 RAG 应用

小结

核心要点