Skip to main content

LangChain框架完全指南

LangChain是目前最流行的LLM应用开发框架，提供了丰富的组件和工具链，让开发者可以快速构建智能应用。

什么是LangChain​

核心价值​

LangChain解决了LLM应用开发的三大痛点：

  1. 标准化接口 - 统一不同LLM提供商的API
  2. 模块化组件 - 提供可复用的功能组件
  3. 链式编排 - 简化复杂workflow的构建
python
1# 不使用LangChain
2import openai
3response = openai.ChatCompletion.create(...)
4# 需要手动处理格式、错误、重试等
5
6# 使用LangChain
7from langchain.llms import OpenAI
8llm = OpenAI()
9response = llm("你好")  # 简洁统一

核心组件​

快速开始​

安装配置​

bash
1# 安装核心包
2pip install langchain
3
4# 安装LLM集成
5pip install langchain-openai
6pip install langchain-anthropic
7
8# 安装社区包
9pip install langchain-community
10
11# 完整安装
12pip install langchain[all]

Hello World​

python
1from langchain.llms import OpenAI
2import os
3
4# 设置API Key
5os.environ["OPENAI_API_KEY"] = "your-api-key"
6
7# 初始化模型
8llm = OpenAI(temperature=0.7)
9
10# 简单调用
11response = llm("请用一句话介绍LangChain")
12print(response)

核心组件详解​

1. Models (模型)​

LLMs - 文本生成模型​

python
1from langchain.llms import OpenAI, Anthropic, HuggingFaceHub
2
3# OpenAI
4openai_llm = OpenAI(
5    model_name="gpt-3.5-turbo-instruct",
6    temperature=0.7,
7    max_tokens=500
8)
9
10# Claude
11claude_llm = Anthropic(
12    model="claude-3-sonnet-20240229",
13    anthropic_api_key="your-key"
14)
15
16# 开源模型
17hf_llm = HuggingFaceHub(
18    repo_id="mistralai/Mistral-7B-v0.1",
19    huggingfacehub_api_token="your-token"
20)
21
22# 统一调用接口
23result = openai_llm("你好")

Chat Models - 对话模型​

python
1from langchain.chat_models import ChatOpenAI
2from langchain.schema import HumanMessage, SystemMessage, AIMessage
3
4chat = ChatOpenAI(
5    model="gpt-4",
6    temperature=0
7)
8
9messages = [
10    SystemMessage(content="你是一位Python专家"),
11    HumanMessage(content="如何使用LangChain?")
12]
13
14response = chat(messages)
15print(response.content)

Embeddings - 向量化模型​

python
1from langchain.embeddings import OpenAIEmbeddings, HuggingFaceEmbeddings
2
3# OpenAI Embeddings
4openai_embeddings = OpenAIEmbeddings(
5    model="text-embedding-3-small"
6)
7
8# 向量化单个文本
9vector = openai_embeddings.embed_query("Hello World")
10print(f"维度: {len(vector)}")
11
12# 批量向量化
13texts = ["文本1", "文本2", "文本3"]
14vectors = openai_embeddings.embed_documents(texts)

2. Prompts (提示词)​

PromptTemplate​

python
1from langchain.prompts import PromptTemplate
2
3# 基础模板
4template = """
5你是一位{role}。
6请回答以下问题：{question}
7"""
8
9prompt = PromptTemplate(
10    input_variables=["role", "question"],
11    template=template
12)
13
14# 格式化
15formatted_prompt = prompt.format(
16    role="Python专家",
17    question="如何优化代码性能？"
18)
19
20print(formatted_prompt)

ChatPromptTemplate​

python
1from langchain.prompts import (
2    ChatPromptTemplate,
3    SystemMessagePromptTemplate,
4    HumanMessagePromptTemplate
5)
6
7# 构建对话提示
8system_template = "你是一位{expertise}专家"
9system_prompt = SystemMessagePromptTemplate.from_template(system_template)
10
11human_template = "{user_input}"
12human_prompt = HumanMessagePromptTemplate.from_template(human_template)
13
14chat_prompt = ChatPromptTemplate.from_messages([
15    system_prompt,
16    human_prompt
17])
18
19# 使用
20messages = chat_prompt.format_messages(
21    expertise="数据分析",
22    user_input="如何分析用户行为？"
23)

FewShotPromptTemplate​

python
1from langchain.prompts import FewShotPromptTemplate
2
3# 示例数据
4examples = [
5    {"input": "开心", "output": "positive"},
6    {"input": "难过", "output": "negative"},
7    {"input": "生气", "output": "negative"}
8]
9
10# 示例模板
11example_template = """
12输入: {input}
13输出: {output}
14"""
15
16example_prompt = PromptTemplate(
17    input_variables=["input", "output"],
18    template=example_template
19)
20
21# Few-shot提示
22few_shot_prompt = FewShotPromptTemplate(
23    examples=examples,
24    example_prompt=example_prompt,
25    prefix="将情感分类为positive或negative:",
26    suffix="输入: {input}\n输出:",
27    input_variables=["input"]
28)
29
30print(few_shot_prompt.format(input="兴奋"))

3. Chains (链)​

LLMChain - 基础链​

python
1from langchain.chains import LLMChain
2from langchain.llms import OpenAI
3from langchain.prompts import PromptTemplate
4
5# 创建链
6llm = OpenAI(temperature=0.7)
7prompt = PromptTemplate(
8    input_variables=["product"],
9    template="为{product}写一句广告语"
10)
11
12chain = LLMChain(llm=llm, prompt=prompt)
13
14# 执行
15result = chain.run(product="智能手表")
16print(result)
17
18# 批量执行
19results = chain.apply([
20    {"product": "智能手表"},
21    {"product": "无线耳机"}
22])

SequentialChain - 顺序链​

python
1from langchain.chains import SequentialChain
2
3# 链1: 生成故事大纲
4outline_chain = LLMChain(
5    llm=llm,
6    prompt=PromptTemplate(
7        input_variables=["topic"],
8        template="为{topic}写一个故事大纲"
9    ),
10    output_key="outline"
11)
12
13# 链2: 基于大纲写故事
14story_chain = LLMChain(
15    llm=llm,
16    prompt=PromptTemplate(
17        input_variables=["outline"],
18        template="基于以下大纲写一个故事:\n{outline}"
19    ),
20    output_key="story"
21)
22
23# 组合链
24overall_chain = SequentialChain(
25    chains=[outline_chain, story_chain],
26    input_variables=["topic"],
27    output_variables=["outline", "story"],
28    verbose=True
29)
30
31# 执行
32result = overall_chain({"topic": "时间旅行"})

RouterChain - 路由链​

python
1from langchain.chains.router import MultiPromptChain
2from langchain.chains import ConversationChain
3
4# 定义不同领域的提示
5prompt_infos = [
6    {
7        "name": "physics",
8        "description": "适合回答物理问题",
9        "prompt_template": "你是物理专家。{input}"
10    },
11    {
12        "name": "math",
13        "description": "适合回答数学问题",
14        "prompt_template": "你是数学专家。{input}"
15    }
16]
17
18# 创建路由链
19router_chain = MultiPromptChain.from_prompts(
20    llm=llm,
21    prompt_infos=prompt_infos
22)
23
24# 自动路由到合适的链
25result = router_chain.run("什么是牛顿第二定律？")

4. Memory (记忆)​

ConversationBufferMemory​

python
1from langchain.memory import ConversationBufferMemory
2from langchain.chains import ConversationChain
3
4# 初始化记忆
5memory = ConversationBufferMemory()
6
7# 创建对话链
8conversation = ConversationChain(
9    llm=llm,
10    memory=memory,
11    verbose=True
12)
13
14# 多轮对话
15conversation.run("你好，我叫张三")
16conversation.run("我叫什么名字？")  # 会记住之前的对话
17
18# 查看记忆
19print(memory.load_memory_variables({}))

ConversationBufferWindowMemory​

python
1from langchain.memory import ConversationBufferWindowMemory
2
3# 只保留最近k轮对话
4window_memory = ConversationBufferWindowMemory(k=2)
5
6conversation = ConversationChain(
7    llm=llm,
8    memory=window_memory
9)
10
11# 只会记住最近2轮对话

ConversationSummaryMemory​

python
1from langchain.memory import ConversationSummaryMemory
2
3# 使用LLM总结历史对话
4summary_memory = ConversationSummaryMemory(llm=llm)
5
6conversation = ConversationChain(
7    llm=llm,
8    memory=summary_memory
9)
10
11# 会自动总结长对话，节省token

5. Agents (智能体)​

基础Agent​

python
1from langchain.agents import initialize_agent, Tool, AgentType
2from langchain.llms import OpenAI
3
4# 定义工具
5def search_tool(query):
6    return f"搜索结果: {query}"
7
8def calculator(expression):
9    return eval(expression)
10
11tools = [
12    Tool(
13        name="Search",
14        func=search_tool,
15        description="用于搜索信息"
16    ),
17    Tool(
18        name="Calculator",
19        func=calculator,
20        description="用于数学计算，输入数学表达式"
21    )
22]
23
24# 初始化Agent
25agent = initialize_agent(
26    tools=tools,
27    llm=OpenAI(temperature=0),
28    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
29    verbose=True
30)
31
32# Agent会自主选择工具
33result = agent.run("25乘以4等于多少？")

ReAct Agent​

python
1# ReAct: Reasoning + Acting
2from langchain.agents import create_react_agent
3
4# Agent会遵循: 思考 -> 行动 -> 观察 -> 思考 的循环
5agent = create_react_agent(
6    llm=llm,
7    tools=tools,
8    prompt=react_prompt
9)
10
11agent_executor = AgentExecutor(
12    agent=agent,
13    tools=tools,
14    verbose=True,
15    max_iterations=5
16)
17
18result = agent_executor.invoke({
19    "input": "2024年世界杯在哪里举办？"
20})

Custom Agent​

python
1from langchain.agents import BaseSingleActionAgent
2
3class MyCustomAgent(BaseSingleActionAgent):
4    @property
5    def input_keys(self):
6        return ["input"]
7    
8    def plan(self, intermediate_steps, **kwargs):
9        # 自定义决策逻辑
10        user_input = kwargs["input"]
11        
12        # 返回下一步行动
13        return AgentAction(
14            tool="Search",
15            tool_input=user_input,
16            log="决定使用搜索工具"
17        )
18    
19    async def aplan(self, intermediate_steps, **kwargs):
20        return self.plan(intermediate_steps, **kwargs)

实战项目​

项目1: 文档问答系统​

python
1from langchain.document_loaders import DirectoryLoader, TextLoader
2from langchain.text_splitter import RecursiveCharacterTextSplitter
3from langchain.embeddings import OpenAIEmbeddings
4from langchain.vectorstores import Chroma
5from langchain.chains import RetrievalQA
6from langchain.llms import OpenAI
7
8class DocumentQA:
9    def __init__(self, docs_path):
10        self.docs_path = docs_path
11        self.vectorstore = None
12        self.qa_chain = None
13    
14    def setup(self):
15        # 1. 加载文档
16        loader = DirectoryLoader(
17            self.docs_path,
18            glob="**/*.txt",
19            loader_cls=TextLoader
20        )
21        documents = loader.load()
22        
23        # 2. 切分文档
24        text_splitter = RecursiveCharacterTextSplitter(
25            chunk_size=1000,
26            chunk_overlap=200
27        )
28        chunks = text_splitter.split_documents(documents)
29        
30        # 3. 创建向量库
31        embeddings = OpenAIEmbeddings()
32        self.vectorstore = Chroma.from_documents(
33            documents=chunks,
34            embedding=embeddings
35        )
36        
37        # 4. 创建QA链
38        llm = OpenAI(temperature=0)
39        self.qa_chain = RetrievalQA.from_chain_type(
40            llm=llm,
41            chain_type="stuff",
42            retriever=self.vectorstore.as_retriever()
43        )
44    
45    def query(self, question):
46        return self.qa_chain.run(question)
47
48# 使用
49qa_system = DocumentQA("./documents")
50qa_system.setup()
51answer = qa_system.query("文档的主要内容是什么？")

项目2: 智能客服​

python
1from langchain.chat_models import ChatOpenAI
2from langchain.memory import ConversationBufferMemory
3from langchain.chains import ConversationChain
4from langchain.prompts import (
5    ChatPromptTemplate,
6    MessagesPlaceholder,
7    SystemMessagePromptTemplate,
8    HumanMessagePromptTemplate
9)
10
11class CustomerServiceBot:
12    def __init__(self):
13        self.chat = ChatOpenAI(model="gpt-3.5-turbo")
14        self.memory = ConversationBufferMemory(return_messages=True)
15        
16        # 定义系统提示
17        system_prompt = """
18        你是一位专业的客服代表，负责回答用户关于产品的问题。
19        
20        产品信息：
21        - 产品名称：智能手表
22        - 价格：¥1999
23        - 特性：健康监测、运动追踪、通话功能
24        - 保修：1年
25        
26        请：
27        1. 友好、专业地回答问题
28        2. 如果不确定，诚实说明
29        3. 主动询问是否需要更多帮助
30        """
31        
32        prompt = ChatPromptTemplate.from_messages([
33            SystemMessagePromptTemplate.from_template(system_prompt),
34            MessagesPlaceholder(variable_name="history"),
35            HumanMessagePromptTemplate.from_template("{input}")
36        ])
37        
38        self.conversation = ConversationChain(
39            llm=self.chat,
40            memory=self.memory,
41            prompt=prompt,
42            verbose=True
43        )
44    
45    def chat(self, user_message):
46        return self.conversation.predict(input=user_message)
47    
48    def reset(self):
49        self.memory.clear()
50
51# 使用
52bot = CustomerServiceBot()
53print(bot.chat("你好"))
54print(bot.chat("这款手表防水吗？"))
55print(bot.chat("价格多少？"))

项目3: 代码审查助手​

python
1from langchain.chains import LLMChain
2from langchain.prompts import PromptTemplate
3from langchain.chat_models import ChatOpenAI
4
5class CodeReviewer:
6    def __init__(self):
7        self.llm = ChatOpenAI(model="gpt-4", temperature=0)
8        
9        self.review_prompt = PromptTemplate(
10            input_variables=["code", "language"],
11            template="""
12            请审查以下{language}代码，提供详细的反馈：
13            
14            代码：
15            ```{language}
16            {code}

请从以下方面审查：

  1. 代码质量和可读性
  2. 潜在的bug和安全问题
  3. 性能优化建议
  4. 最佳实践和改进建议

请提供结构化的审查意见。 """ )

self.chain = LLMChain(llm=self.llm, prompt=self.review_prompt)

def review(self, code, language="python"): return self.chain.run(code=code, language=language)

使用

reviewer = CodeReviewer()

code = """ def calculate_sum(numbers): total = 0 for i in range(len(numbers)): total = total + numbers[i] return total """

review = reviewer.review(code) print(review)

1### 项目4: SQL生成Agent
2
3```python
4from langchain.agents import create_sql_agent
5from langchain.agents.agent_toolkits import SQLDatabaseToolkit
6from langchain.sql_database import SQLDatabase
7from langchain.chat_models import ChatOpenAI
8
9# 连接数据库
10db = SQLDatabase.from_uri("sqlite:///example.db")
11
12# 创建SQL Agent
13llm = ChatOpenAI(model="gpt-4", temperature=0)
14toolkit = SQLDatabaseToolkit(db=db, llm=llm)
15
16agent_executor = create_sql_agent(
17    llm=llm,
18    toolkit=toolkit,
19    verbose=True,
20    agent_type="openai-functions"
21)
22
23# 自然语言查询数据库
24result = agent_executor.run(
25    "有多少用户注册于2024年？"
26)
27print(result)

LangChain高级特性​

1. Streaming输出​

python
1from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
2from langchain.chat_models import ChatOpenAI
3
4# 流式输出
5chat = ChatOpenAI(
6    streaming=True,
7    callbacks=[StreamingStdOutCallbackHandler()],
8    temperature=0.7
9)
10
11response = chat.predict("写一首关于春天的诗")
12# 输出会逐字显示

2. Callbacks监控​

python
1from langchain.callbacks import StdOutCallbackHandler
2from langchain.chains import LLMChain
3
4class CustomCallback(StdOutCallbackHandler):
5    def on_llm_start(self, serialized, prompts, **kwargs):
6        print(f"LLM开始: {prompts}")
7    
8    def on_llm_end(self, response, **kwargs):
9        print(f"LLM结束: {response}")
10    
11    def on_chain_start(self, serialized, inputs, **kwargs):
12        print(f"Chain开始: {inputs}")
13
14# 使用callback
15chain = LLMChain(
16    llm=llm,
17    prompt=prompt,
18    callbacks=[CustomCallback()]
19)

3. 缓存​

python
1from langchain.cache import InMemoryCache, SQLiteCache
2import langchain
3
4# 内存缓存
5langchain.llm_cache = InMemoryCache()
6
7# 持久化缓存
8langchain.llm_cache = SQLiteCache(database_path=".langchain.db")
9
10# 相同问题不会重复调用API
11llm("你好")  # 调用API
12llm("你好")  # 从缓存返回

4. 异步支持​

python
1import asyncio
2from langchain.chat_models import ChatOpenAI
3
4async def async_generate():
5    chat = ChatOpenAI()
6    
7    # 并发生成
8    tasks = [
9        chat.agenerate([["讲个笑话"]]),
10        chat.agenerate([["写首诗"]]),
11        chat.agenerate([["推荐本书"]])
12    ]
13    
14    results = await asyncio.gather(*tasks)
15    return results
16
17# 运行
18results = asyncio.run(async_generate())

与其他框架对比​

特性LangChainLlamaIndexSemantic Kernel
主要用途通用LLM应用数据检索企业级集成
学习曲线中等较低较高
生态系统最丰富专注RAG微软生态
Agent支持优秀基础优秀
文档质量良好优秀良好
社区活跃度非常高高中等

最佳实践​

1. 错误处理​

python
1from langchain.schema import OutputParserException
2
3try:
4    result = chain.run("你的问题")
5except OutputParserException as e:
6    print(f"输出解析错误: {e}")
7except Exception as e:
8    print(f"发生错误: {e}")

2. 提示词优化​

python
1# ❌ 不好的提示
2bad_prompt = "翻译: {text}"
3
4# ✅ 好的提示
5good_prompt = """
6请将以下文本从英文翻译成中文。
7要求：
81. 保持原意
92. 使用地道的中文表达
103. 保留专业术语
11
12文本: {text}
13
14译文:
15"""

3. 成本优化​

python
1# 根据任务复杂度选择模型
2def smart_llm_selection(task_complexity):
3    if task_complexity == "simple":
4        return ChatOpenAI(model="gpt-3.5-turbo")  # 便宜
5    elif task_complexity == "complex":
6        return ChatOpenAI(model="gpt-4")  # 贵但强大
7    
8# 使用缓存
9import langchain
10langchain.llm_cache = InMemoryCache()
11
12# 批量处理
13chain.apply([{"input": "q1"}, {"input": "q2"}])  # 比单独调用快

常见问题​

Q1: 如何处理超长文本？​

python
1from langchain.text_splitter import RecursiveCharacterTextSplitter
2from langchain.chains import MapReduceDocumentsChain, ReduceDocumentsChain
3from langchain.chains.combine_documents.stuff import StuffDocumentsChain
4
5# 1. 切分文本
6text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000)
7docs = text_splitter.create_documents([long_text])
8
9# 2. Map-Reduce处理
10map_reduce_chain = MapReduceDocumentsChain(
11    llm_chain=llm_chain,
12    reduce_documents_chain=reduce_chain
13)
14
15result = map_reduce_chain.run(docs)

Q2: 如何debug链的执行？​

python
1# 方法1: 开启verbose
2chain = LLMChain(llm=llm, prompt=prompt, verbose=True)
3
4# 方法2: 使用debug
5import langchain
6langchain.debug = True
7
8# 方法3: 自定义callback
9class DebugCallback(BaseCallbackHandler):
10    def on_chain_start(self, serialized, inputs, **kwargs):
11        print(f"开始执行: {serialized['name']}")
12        print(f"输入: {inputs}")

Q3: 如何集成自定义LLM？​

python
1from langchain.llms.base import LLM
2from typing import Optional, List
3
4class CustomLLM(LLM):
5    model_name: str = "custom-model"
6    
7    @property
8    def _llm_type(self) -> str:
9        return "custom"
10    
11    def _call(
12        self,
13        prompt: str,
14        stop: Optional[List[str]] = None
15    ) -> str:
16        # 调用你的模型API
17        response = your_model_api(prompt)
18        return response
19    
20    @property
21    def _identifying_params(self):
22        return {"model_name": self.model_name}
23
24# 使用
25custom_llm = CustomLLM()
26chain = LLMChain(llm=custom_llm, prompt=prompt)

学习资源​

官方资源​

教程课程​

社区​

总结​

LangChain是构建LLM应用的强大框架，通过本指南你已经掌握：

  • ✅ LangChain的核心概念和组件
  • ✅ Models、Prompts、Chains、Agents、Memory的使用
  • ✅ 4个完整的实战项目
  • ✅ 高级特性和最佳实践
  • ✅ 常见问题的解决方案
下一步
forum

评论区 / Comments