麻瓜教程

Appearance

LangChain 工作流 ​

LangChain 工作流是指使用 LangChain 组件构建 LLM 应用的过程和方法。本节将详细介绍 LangChain 1.2 的工作流概念和实现方式。

基本工作流 ​

LangChain 的基本工作流通常包括以下步骤：

  1. 定义需求：明确应用的功能和目标
  2. 选择模型：根据需求选择合适的语言模型
  3. 构建提示词：设计有效的提示词模板
  4. 组合组件：使用 LCEL 语法组合组件
  5. 添加记忆：根据需要添加记忆组件
  6. 集成工具：根据需要集成外部工具
  7. 评估和优化：评估应用性能并进行优化
  8. 部署应用：将应用部署到生产环境

工作流示例 ​

1. 构建简单的问答系统 ​

步骤 1：选择模型 ​

python
from langchain_openai import ChatOpenAI

# 初始化聊天模型
chat_model = ChatOpenAI(api_key="your-api-key")

步骤 2：构建提示词 ​

python
from langchain_core.prompts import ChatPromptTemplate

# 创建聊天提示词模板
chat_template = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant. Answer the user's questions based on your knowledge."),
    ("human", "{question}")
])

步骤 3：组合组件（使用 LCEL） ​

python
from langchain_core.output_parsers import StrOutputParser

# 创建输出解析器
output_parser = StrOutputParser()

# 使用 LCEL 语法创建链
chain = chat_template | chat_model | output_parser

# 运行链
response = chain.invoke({"question": "What is LangChain?"})
print(response)

2. 构建带有记忆的聊天机器人 ​

步骤 1：选择模型 ​

python
from langchain_openai import ChatOpenAI

# 初始化聊天模型
chat_model = ChatOpenAI(api_key="your-api-key")

步骤 2：构建提示词 ​

python
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 创建聊天提示词模板
chat_template = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant. Maintain context from previous messages."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

步骤 3：添加记忆 ​

python
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

# 创建消息历史存储
store = {}

def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

步骤 4：组合组件 ​

python
from langchain_core.output_parsers import StrOutputParser

# 创建输出解析器
output_parser = StrOutputParser()

# 使用 LCEL 语法创建基础链
base_chain = chat_template | chat_model | output_parser

# 创建带记忆的链
chain_with_history = RunnableWithMessageHistory(
    base_chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

# 运行链
response1 = chain_with_history.invoke(
    {"input": "What is LangChain?"},
    config={"configurable": {"session_id": "user_123"}}
)
print("Response 1:", response1)

response2 = chain_with_history.invoke(
    {"input": "Can you give me an example of how to use it?"},
    config={"configurable": {"session_id": "user_123"}}
)
print("Response 2:", response2)

3. 构建使用工具的代理 ​

步骤 1：选择模型 ​

python
from langchain_openai import OpenAI

# 初始化语言模型
llm = OpenAI(api_key="your-api-key")

步骤 2：集成工具 ​

python
from langchain.agents import load_tools

# 加载工具
tools = load_tools(["serpapi", "llm-math"], llm=llm)

步骤 3：初始化代理 ​

python
from langchain.agents import AgentType, initialize_agent

# 初始化代理
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 运行代理
response = agent.run("What is the current temperature in New York?")
print(response)

4. 构建基于文档的问答系统 ​

步骤 1：加载和处理文档 ​

python
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter

# 加载文档
loader = TextLoader("path/to/document.txt")
documents = loader.load()

# 分割文档
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.split_documents(documents)

步骤 2：创建向量存储 ​

python
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

# 初始化嵌入模型
embeddings = OpenAIEmbeddings(api_key="your-api-key")

# 创建向量存储
vectorstore = FAISS.from_documents(texts, embeddings)

步骤 3：构建检索链 ​

python
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI

# 初始化聊天模型
chat_model = ChatOpenAI(api_key="your-api-key")

# 创建检索链
qa_chain = RetrievalQA.from_chain_type(
    llm=chat_model,
    chain_type="stuff",
    retriever=vectorstore.as_retriever()
)

# 运行检索链
response = qa_chain.invoke({"query": "What is the main topic of the document?"})
print(response["result"])

工作流最佳实践 ​

1. 模块化设计 ​

  • 将应用分解为独立的组件
  • 使用函数和类封装逻辑
  • 保持组件的单一职责

2. 提示词优化 ​

  • 清晰明确地表达任务要求
  • 使用示例引导模型输出
  • 控制提示词长度，避免超出模型上下文窗口

3. 错误处理 ​

  • 添加异常处理机制
  • 实现重试逻辑
  • 监控和记录错误

4. 性能优化 ​

  • 合理使用缓存
  • 优化模型调用频率
  • 选择合适的模型参数

5. 评估和迭代 ​

  • 定义明确的评估指标
  • 定期评估应用性能
  • 根据评估结果进行迭代优化

高级工作流 ​

1. 多模型协作 ​

LangChain 支持多模型协作，不同的模型可以负责不同的任务：

python
from langchain_openai import ChatOpenAI, OpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel

# 初始化不同的模型
chat_model = ChatOpenAI(api_key="your-api-key", model_name="gpt-4")
llm = OpenAI(api_key="your-api-key", model_name="gpt-3.5-turbo-instruct")

# 创建输出解析器
output_parser = StrOutputParser()

# 创建第一个链（使用 GPT-4 进行复杂推理）
prompt1 = PromptTemplate(
    template="Analyze the following text and identify the key points: {text}",
    input_variables=["text"]
)
chain1 = prompt1 | chat_model | output_parser

# 创建第二个链（使用 GPT-3.5 进行总结）
prompt2 = PromptTemplate(
    template="Summarize the following analysis in a concise way: {analysis}",
    input_variables=["analysis"]
)
chain2 = prompt2 | llm | output_parser

# 使用 RunnableParallel 并行执行
parallel_chain = RunnableParallel(
    analysis=chain1,
    text=lambda x: x["text"]
)

# 组合链
combined_chain = (
    parallel_chain
    | (lambda x: {"analysis": x["analysis"], "text": x["text"]})
    | chain2
)

# 运行链
response = combined_chain.invoke({"text": "LangChain is a framework for building LLM applications. It provides tools and components for working with language models, including prompts, chains, agents, tools, memory, and vector stores."})
print("Summary:", response)

2. 事件驱动工作流 ​

LangChain 支持事件驱动的工作流，通过回调机制响应各种事件：

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

# 创建自定义回调处理器
class MyCallbackHandler(BaseCallbackHandler):
    def on_llm_start(self, *args, **kwargs):
        print("LLM 开始运行...")
    
    def on_llm_end(self, *args, **kwargs):
        print("LLM 运行结束...")

# 初始化聊天模型
chat_model = ChatOpenAI(api_key="your-api-key")

# 创建聊天提示词模板
chat_template = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant"),
    ("human", "{question}")
])

# 创建输出解析器
output_parser = StrOutputParser()

# 使用 LCEL 语法创建链
chain = chat_template | chat_model | output_parser

# 运行链
response = chain.invoke(
    {"question": "What is LangChain?"},
    config={"callbacks": [MyCallbackHandler()]}
)
print(response)

总结 ​

LangChain 工作流是构建 LLM 应用的核心方法，通过灵活组合各种组件，可以实现从简单到复杂的各种 LLM 应用。在 LangChain 1.2 中，推荐使用 LCEL 语法来创建链，使用 RunnableWithMessageHistory 来管理对话历史。

掌握 LangChain 的工作流概念和最佳实践，将帮助您更有效地构建功能强大、性能优良的 LLM 应用。

在接下来的章节中，我们将详细介绍如何实现各种具体的 LLM 应用，帮助您进一步掌握 LangChain 的使用技巧。

Copyright © 2025-2026 京公网安备 11010102000001号 京ICP备2024078128号-1