当前位置：首页 > article >正文

运用 LangChain 编排任务处理流水线，实现多轮对话场景

article 2025/2/12 16:52:47

在LangChain中，任务处理流水线和多轮对话的场景可以通过LangChain的链式结构（Chain）和会话记忆（Conversation Memory）来实现。LangChain框架旨在简化与大语言模型（如 GPT-3）进行交互的流程，提供了灵活的工具来处理任务和对话。接下来，我将通过一个示例来展示如何使用LangChain来编排任务处理流水线，并实现多轮对话的场景。

1. 安装LangChain

首先，确保已经安装了LangChain库：

pip install langchain

2. 基础设置

在使用LangChain时，我们通常需要配置一个语言模型（比如OpenAI的GPT模型）。以下是设置的基础结构：

from langchain.llms import OpenAI
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory
from langchain.prompts import PromptTemplate

# 使用 OpenAI GPT 作为语言模型
llm = OpenAI(model="text-davinci-003", temperature=0.7)

# 设置对话记忆，允许LangChain在多轮对话中保持上下文
memory = ConversationBufferMemory()

# 创建一个ConversationChain（对话链）
conversation_chain = ConversationChain(llm=llm, memory=memory)

3. 定义任务流水线

接下来，我们通过LangChain的链式结构来定义一个任务处理流水线。可以在流水线中加入多个步骤，按顺序处理。

例如，我们定义一个简单的流水线，首先进行用户输入的处理，然后通过调用语言模型生成回复。

from langchain.prompts import PromptTemplate

# 定义一个模板，指导对话模型的回答
template = "用户说: {input_text}\n助手:"
prompt = PromptTemplate(input_variables=["input_text"], template=template)

# 使用模板和语言模型创建一个链式结构
prompt_chain = prompt | llm

def process_pipeline(user_input):
    # 通过链式结构生成回复
    response = prompt_chain.invoke({"input_text": user_input})
    return response

4. 实现多轮对话场景

在多轮对话中，记忆会保存用户和系统之间的对话记录。每次用户输入时，系统都会从记忆中获取上下文，确保对话的连贯性。

# 对话过程中的多轮交互
def run_conversation():
    while True:
        user_input = input("请输入您的问题：")
        if user_input.lower() == "退出":
            break
        response = conversation_chain.run(user_input)
        print(f"助手回答：{response}")

# 启动对话
run_conversation()

5. 任务处理流水线与多轮对话结合

如果需要根据不同任务或不同的对话情境调整响应内容，可以通过LangChain的Chain来创建更复杂的多步骤任务处理流水线。

假设我们有一个多轮对话场景，其中在用户提问后，系统需要先分析问题的类型（如技术、天气、新闻等），然后根据类型生成适当的响应。

from langchain.prompts import PromptTemplate

# 定义不同任务类型的模板
task_templates = {
    "技术": "用户询问技术问题: {input_text}\n助手回答技术问题:",
    "天气": "用户询问天气情况: {input_text}\n助手回答天气问题:",
    "新闻": "用户询问最新新闻: {input_text}\n助手回答新闻问题:"
}

def get_task_prompt(user_input):
    # 判断问题的类型，选择合适的模板
    if "技术" in user_input:
        task_type = "技术"
    elif "天气" in user_input:
        task_type = "天气"
    elif "新闻" in user_input:
        task_type = "新闻"
    else:
        task_type = "其他"

    # 根据问题类型选择对应的模板
    template = task_templates.get(task_type, task_templates["其他"])
    return PromptTemplate(input_variables=["input_text"], template=template)

# 创建对话链与任务处理链
def process_pipeline_with_task(user_input):
    prompt = get_task_prompt(user_input)
    task_chain = prompt | llm
    response = task_chain.invoke({"input_text": user_input})
    return response

# 测试
print(process_pipeline_with_task("今天天气如何？"))