基础链

链（Chains）是 LangChain 中的核心概念之一，它将多个组件组合在一起，形成一个完整的工作流。在 LangChain 1.2 中，推荐使用 LCEL（LangChain Expression Language）语法来构建链。

安装依赖

在开始使用基础链之前，需要安装以下依赖：

pip install langchain langchain-openai python-dotenv

什么是基础链？

基础链是将提示词模板和语言模型组合在一起的简单链。在 LangChain 1.2 中，使用管道操作符 | 来组合组件，形成链式调用。

简单链

使用 LCEL 语法构建最简单的链，将提示词模板和语言模型组合在一起。

初始化简单链

from langchain_core.prompts import PromptTemplate
from langchain_openai import OpenAI
from langchain_core.output_parsers import StrOutputParser

# 创建提示词模板
template = "What is a good name for a {product} company that makes {description}?"
prompt = PromptTemplate(template=template, input_variables=["product", "description"])

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

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

# 使用 LCEL 语法创建链
chain = prompt | llm | output_parser

使用简单链

# 运行链
response = chain.invoke({"product": "tech", "description": "smartphones"})
print(response)

# 批量运行
responses = chain.batch([
    {"product": "tech", "description": "smartphones"},
    {"product": "food", "description": "organic vegetables"}
])
for i, result in enumerate(responses):
    print(f"Response {i+1}: {result}")

# 异步运行
async_response = await chain.ainvoke({"product": "tech", "description": "smartphones"})
print(async_response)

聊天链

聊天链专门用于聊天模型，它将聊天提示词模板和聊天模型组合在一起。

初始化聊天链

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

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

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

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

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

使用聊天链

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

带记忆的链

在 LangChain 1.2 中，使用 RunnableWithMessageHistory 来管理对话历史。

安装依赖

pip install langchain-community

初始化带记忆的链

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

# 创建提示词模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

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

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

# 创建基础链
chain = prompt | llm | output_parser

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

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

# 创建带记忆的链
chain_with_history = RunnableWithMessageHistory(
    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?"},
    config={"configurable": {"session_id": "user_123"}}
)
print("Response 2:", response2)

# 查看历史记录
history = get_session_history("user_123")
print("History:", history.messages)

带输出解析器的链

链可以与输出解析器结合使用，以解析模型的输出。

初始化带输出解析器的链

from langchain_core.prompts import PromptTemplate
from langchain_openai import OpenAI
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field

# 定义输出结构
class CompanyInfo(BaseModel):
    company_name: str = Field(description="The name of the company")
    tagline: str = Field(description="A catchy tagline for the company")

# 创建输出解析器
output_parser = JsonOutputParser(pydantic_object=CompanyInfo)

# 创建提示词模板
template = """What is a good name for a {product} company that makes {description}?

{format_instructions}
"""
prompt = PromptTemplate(
    template=template,
    input_variables=["product", "description"],
    partial_variables={"format_instructions": output_parser.get_format_instructions()}
)

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

# 创建链
chain = prompt | llm | output_parser

使用带输出解析器的链

# 运行链
response = chain.invoke({"product": "tech", "description": "smartphones"})
print("Company name:", response["company_name"])
print("Tagline:", response["tagline"])

自定义 Runnable

除了使用内置的链类型，还可以创建自定义的 Runnable。

创建自定义 Runnable

from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_core.prompts import PromptTemplate
from langchain_openai import OpenAI
from langchain_core.output_parsers import StrOutputParser

# 创建提示词模板
prompt = PromptTemplate(
    template="What is a good name for a {product} company?",
    input_variables=["product"]
)

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

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

# 创建自定义处理函数
def add_prefix(input_str: str) -> str:
    return f"Company Name Suggestion: {input_str}"

# 使用 RunnableLambda 创建自定义组件
custom_runnable = RunnableLambda(add_prefix)

# 组合链
chain = prompt | llm | output_parser | custom_runnable

# 使用自定义链
response = chain.invoke({"product": "tech"})
print(response)

使用 RunnablePassthrough

from langchain_core.runnables import RunnablePassthrough

# 创建链，保留原始输入
chain = (
    {"product": RunnablePassthrough(), "description": lambda x: x["description"]}
    | prompt
    | llm
    | output_parser
)

response = chain.invoke({"product": "tech", "description": "smartphones"})
print(response)

并行链

使用 LCEL 可以轻松地创建并行链。

初始化并行链

from langchain_core.runnables import RunnableParallel

# 创建两个并行的链
chain1 = PromptTemplate.from_template("What is a good name for a {product} company?") | llm | output_parser
chain2 = PromptTemplate.from_template("What is a good slogan for a {product} company?") | llm | output_parser

# 创建并行链
parallel_chain = RunnableParallel(
    company_name=chain1,
    slogan=chain2
)

# 使用并行链
response = parallel_chain.invoke({"product": "tech"})
print("Company name:", response["company_name"])
print("Slogan:", response["slogan"])

总结

在 LangChain 1.2 中，推荐使用 LCEL 语法来构建链。主要特点包括：

1. 管道操作符 |：使用管道操作符组合组件
2. invoke/batch/ainvoke：统一的调用接口
3. RunnableWithMessageHistory：新的记忆管理方式
4. RunnableParallel：轻松创建并行链
5. RunnableLambda：创建自定义处理逻辑

在实际应用中，您可以根据具体需求选择合适的链类型，并根据需要添加记忆、输出解析器等组件，构建功能强大的 LLM 应用。