Langchain模块Chain组件基本用法

 2025/12/15 

Chain是一种接口，将多个组件（llm、prompt等）串联在一起，以实现更复杂的应用或功能

模块化：将复杂的应用拆解成多个更简单，更易于管理和维护的模块，每个模块都可以独立进行开发和测试
可重用性：创建可以重复使用的组件和功能
可扩展性：可以轻松添加、修改、删除链中的组件，以满足应用的新需求或改变现有功能

API异步调用

import asyncio
import time

from langchain_deepseek import ChatDeepSeek
from langchain.prompts import PromptTemplate
from langchain.schema.output_parser import StrOutputParser
from dotenv import load_dotenv

load_dotenv()

template="我养了一只{pet}，叫什么好呢？回答不超过50个字"
# 同步函数
def generate_serially():
    llm = ChatDeepSeek(model="deepseek-chat",streaming=True,temperature=0.7)
    prompt = PromptTemplate(
        input_variables=['pet'],
        template=template,
    )
    chain = prompt | llm | StrOutputParser()
    for _ in range(5):
        response = chain.invoke({"pet": "布偶"})
        print(response)

# 异步函数
async def generate_async(chain):
    response = await chain.ainvoke({"pet": "边牧"})
    print(response)
    
async def generate_concurrentlt():
    llm = ChatDeepSeek(model="deepseek-chat",streaming=True,temperature=0.7)
    prompt = PromptTemplate(
        input_variables=['pet'],
        template=template,
    )
    chain = prompt | llm | StrOutputParser()
    tasks = [generate_async(chain) for _ in range (5)]
    await asyncio.gather(*tasks)
    
# 计算异步时长
async def main():
    await generate_concurrentlt()
s = time.perf_counter()
asyncio.run(main())
elapsed = time.perf_counter() -s
print(elapsed)

# 4.085857541998848

# 计算同步时长
s = time.perf_counter()
generate_serially()
elapsed = time.perf_counter() -s
print(elapsed)
# 11.405498398002237

debug调试模式

from langchain_deepseek import ChatDeepSeek
from langchain.prompts import PromptTemplate
from langchain.schema.output_parser import StrOutputParser
from dotenv import load_dotenv
from langchain.callbacks.tracers import ConsoleCallbackHandler

# 全局调试
from langchain.globals import set_debug
set_debug(True)

load_dotenv()

template="我养了一只{pet}，叫什么好呢？回答不超过50个字"
    
llm = ChatDeepSeek(model="deepseek-chat",streaming=True,temperature=0.7)
prompt = PromptTemplate(
    input_variables=['pet'],
    template=template,
)
chain = prompt | llm | StrOutputParser()

# 全局链调试
# response = chain.invoke({"pet": "边牧"},config={'callbacks': [ConsoleCallbackHandler()]})
# print(response)

# 针对特定组件进行调试(模板)
# prompt = PromptTemplate(
#     input_variables=['pet'],
#     template=template,
# ).with_config({"callbacks": [ConsoleCallbackHandler()]})

# 针对特定组件进行调试(模型)
# response = chain.invoke({"pet": "边牧"},config={'callbacks': [ConsoleCallbackHandler()]})
# print(response)

# 正常流程保留的输出
response = chain.invoke({"pet": "边牧"})
print(response)

引用外部链接LangChain-hub

langchain-hub是一个集中存储和分享LangChain可重用组件的仓库，如提示词、链和代理。是官方组件库

有这个组件，很多时候就不需要手动去写prompt和chain了，直接用别人写好的就可以

官网：https://smith.langchain.com/hub

# 引用外部prompt
from langsmith import Client
from langchain_deepseek import ChatDeepSeek
from langchain.schema.output_parser import StrOutputParser
from langchain.callbacks.tracers import ConsoleCallbackHandler
import os
from dotenv import load_dotenv

# # 全局调试
# from langchain.globals import set_debug
# set_debug(True)

load_dotenv()
client = Client(api_key=os.environ["LANGSMITH_API_KEY"])
# 懒提示语模板：根据提示生成写作指南；变量(lazy_prompt、task)
prompt_lazy = client.pull_prompt("hardkothari/prompt-maker", include_model=True) 
# 文章生成模板：根据提示生成文章：变量(target_audience受众定位、text上下文)
prompt_generate = client.pull_prompt("hardkothari/blog-generator", include_model=True) 

# 使用懒提示语模板生成写作指南
chain = prompt_lazy | ChatDeepSeek(model="deepseek-chat") | StrOutputParser()
result = chain.invoke(
    {
        "lazy_prompt": "写一篇关于AI的文章",
        "task": "使用中文写一个吸引人的博客文章，不超过800字"
    },
)
# 作为一位资深科技专栏作家与人工智能领域观察者，请撰写一篇关于人工智能（AI）的中文博客文章。
# **文章要求如下：**
# 1.  **目标与定位：** 文章需面向对科技感兴趣但非专业出身的普通读者，旨在激发他们对AI的兴趣与思考，而非进行艰深的技术讲解。
# 2.  **核心主题：** 聚焦于 **“AI如何正在悄然改变我们的日常生活与未来想象”** 。避免泛泛而谈，请选择一个具体、新颖且能引发共鸣的切入点（例如：AI个人助手进化史、创意产业中的AI协作者、下一代教育模式、或城市智能管理等）。
# 3.  **内容结构：**
#     *   **引人入胜的开头：** 以一个具体的生活场景、一个令人惊讶的AI应用案例或一个引人深思的问题开篇，迅速抓住读者注意力。
#     *   **主体论述：** 围绕所选切入点，展开2-3个核心段落。结合最新趋势、生动比喻和易懂的实例进行阐述，平衡介绍AI带来的机遇与引发的思考（如伦理、就业、人际关系等）。
#     *   **有力收尾：** 总结核心观点，并展望一个引人遐想的未来可能性，或提出一个开放性问题，鼓励读者参与讨论。
# 4.  **风格与语气：** 语言需流畅、生动、富有感染力。可以适当使用设问、排比等修辞手法。保持专业性与可读性的平衡，避免过于学术化或营销化。
# 5.  **格式与长度：** 文章需包含一个吸引人的标题。全文请严格控制在**800字以内**，段落清晰，适合移动端阅读。
# **请开始你的写作：**

# 将写作指南作为文章生成的上下文，生成文章
chain = prompt_generate | ChatDeepSeek(model="deepseek-chat",streaming=True) | StrOutputParser()
result = chain.invoke(
    {
        "target_audience": "普通人",
        "text": result,
    },
)
print(result)

chain使用memory

from langchain_deepseek import ChatDeepSeek
from langchain.schema.output_parser import StrOutputParser
from dotenv import load_dotenv
load_dotenv()

from langchain_core.runnables import RunnableWithMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

# 存储不同会话的memory
store = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    """获取或创建会话的历史记录"""
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# 创建prompt
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的助手"),
    MessagesPlaceholder(variable_name="history"),  # 历史消息占位符
    ("human", "{query}")
])

# 创建链
base_chain = prompt | ChatDeepSeek(model="deepseek-chat", streaming=True) | StrOutputParser()

# 包装成带memory的链
chain_with_memory = RunnableWithMessageHistory(
    base_chain,
    get_session_history,            # 获取history的函数
    input_messages_key="query",     # 输入消息的键
    history_messages_key="history"  # history占位符的键
)

# 传入session_id
response = chain_with_memory.invoke(
    {"query": "你好，我叫小明"},
    config={"configurable": {"session_id": "user_123"}}  # 配置session_id
)

# 同一session_id会记住历史
next_response = chain_with_memory.invoke(
    {"input": "我刚才说我叫什么？"},
    config={"configurable": {"session_id": "user_123"}} # 指定相同的session_id
)
print(next_response)
# 你刚才说你叫**小明**！我记得很清楚呢～有什么需要我帮忙的吗？ 😄