AI笔记08：LangChain：多任务应用开发

• Langchain提供了一套工具、组件和接口，简化了创建LLM应用的过程。
• 官方解释：they use an LLM to determine which actions to take and in what order. An action can either be using a tool and observing its output, or returning to the user.

LangChain是由多个组件组成的：

• Models：模型，列如GPT-4o
• Prompts：提示，包括提示管理、提示优化和提示序列化
• Memory：记忆，用来保存和模型交互时的上下文
• Indexes：索引，用于结构化文档，方便和模型交互，如果要构建自己的知识库，就需要各种类型文档的加载，转换，长文本切割，文本向量计算，向量索引存储查询等
• Chains：链，一系列对各种组件的调用
• Agents：代理，决定模型采取哪些行动，执行并且观察流程，直到完成为止

在LangChain中集成了一些常用的tools：

• Apify 、ArXivAPI Tool、 AWS Lambda API、 Shell Tool、 Bing Search、 ChatGPT Plugins、DuckDuckGo Search、File System Tools、Google Places、Google Search、Google Serper API、GradioTools、GraphQL tool、HuggingFace Tools、Human as a tool、IFTTT WebHooks、Metaphor Search、Call the API、Use Metaphor as a tool、OpenWeatherMap API、Python REPL、Requests、 SceneXplain、Search Tools、 SearxNG Search API、 SerpAPI、Twilio、Wikipedia、Wolfram Alpha、YouTubeSearchTool、Zapier Natural Language Actions API、Example with SimpleSequentialChain。

1、ReAct范式

• ReAct: Synergizing Reasoning and Acting，将推理和动作相结合，克服LLM胡言乱语的问题，同时提高了结果的可解释性和可信赖度。
• 由LLM 选择工具。执行工具后，将输出结果返回给LLM。不断重复上述过程，直到达到停止条件，一般是LLM 自己认为找到答案了。

示例Python代码：

import re
from typing import List, Union
import textwrap
import time

from langchain.agents import initialize_agent, AgentType, Tool
from langchain_community.llms import Tongyi
from langchain.prompts import PromptTemplate
from langchain.llms.base import BaseLLM

# 输出结果显示，每行最多60字符，每个字符显示停留0.1秒（动态显示效果）
def output_response(response: str) -> None:
    """动态显示响应结果"""
    if not response:
        exit(0)
    # 每行最多60个字符
    for line in textwrap.wrap(response, width=60):
        for word in line.split():
            for char in word:
                print(char, end="", flush=True)
                time.sleep(0.1)  # 每个字符之间延迟0.1秒
            print(" ", end="", flush=True)  # 单词之间添加空格
        print()  # 每行结束后换行
    # 遇到这里，这个问题的回答就结束了
    print("----------------------------------------------------------------")

# 定义了LLM的Prompt Template
CONTEXT_QA_TMPL = """
根据以下提供的信息，回答用户的问题
信息：{context}

问题：{query}
"""
CONTEXT_QA_PROMPT = PromptTemplate(
    input_variables=["query", "context"],
    template=CONTEXT_QA_TMPL,
)

# 模拟公司产品和公司介绍的数据源
class TeslaDataSource:
    def __init__(self, llm: BaseLLM):
        self.llm = llm

    # 工具1：产品描述
    def find_product_description(self, product_name: str) -> str:
        """模拟公司产品的数据库"""
        product_info = {
            "Model 3": "具有简洁、动感的外观设计，流线型车身和现代化前脸。定价23.19-33.19万",
            "Model Y": "在外观上与Model 3类似，但采用了更高的车身和更大的后备箱空间。定价26.39-36.39万",
            "Model X": "拥有独特的翅子门设计和更加大胆的外观风格。定价89.89-105.89万",
        }
          # 彻底清理输入，去除所有可能的额外文本
        # 1. 分割换行符并取第一部分
        # 2. 分割空格并过滤掉非产品名称相关的词
        # 3. 移除所有可能的特殊字符
        cleaned_name = product_name.split('
')[0].strip()
        
        # 尝试直接匹配
        if cleaned_name in product_info:
            return product_info[cleaned_name]
        
        # 不区分大小写匹配
        cleaned_name_lower = cleaned_name.lower()
        for key, value in product_info.items():
            if key.lower() == cleaned_name_lower:
                return value
                
        # 尝试部分匹配，看是否包含产品名称关键词
        for key, value in product_info.items():
            if any(part.lower() in cleaned_name_lower for part in key.split()):
                return value

    # 工具2：公司介绍
    def find_company_info(self, query: str) -> str:
        """模拟公司介绍文档数据库，让llm根据信息回答问题"""
        context = """
        特斯拉最知名的产品是电动汽车，其中包括Model S、Model 3、Model X和Model Y等多款车型。
        特斯拉以其技术创新、高性能和领先的自动驾驶技术而闻名。公司不断推动自动驾驶技术的研发，并在车辆中引入了各种驾驶辅助功能，如自动紧急制动、自适应巡航控制和车道保持辅助等。
        """
        # prompt模板 = 上下文context + 用户的query
        prompt = CONTEXT_QA_PROMPT.format(query=query, context=context)
        # 使用LLM进行推理
        return self.llm(prompt)

# 设置通义千问API密钥
DASHSCOPE_API_KEY = 'sk-882e296067**********cf27e6e20f3ec0'

if __name__ == "__main__":
    # 定义LLM
    llm = Tongyi(model_name="qwen-turbo", dashscope_api_key=DASHSCOPE_API_KEY)
    
    # 自有数据源
    tesla_data_source = TeslaDataSource(llm)
    
    # 定义工具
    tools = [
        Tool(
            name="查询产品名称",
            func=tesla_data_source.find_product_description,
            description="通过产品名称找到产品描述时用的工具，输入的是产品名称",
        ),
        Tool(
            name="公司相关信息",
            func=tesla_data_source.find_company_info,
            description="当用户询问公司相关的问题，可以通过这个工具了解公司信息",
        ),
    ]
    
    # 使用ReAct模式初始化Agent
    agent = initialize_agent(
        tools=tools,
        llm=llm,
        agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
        verbose=True,
        handle_parsing_errors=True,
    )

    # 主过程：可以一直提问下去，直到Ctrl+C
    while True:
        try:
            user_input = input("请输入您的问题：")
            response = agent.run(user_input)
            output_response(response)
        except KeyboardInterrupt:
            print("
程序已退出")
            break
        except Exception as e:
            print(f"发生错误: {e}")
            continue

2、LCEL构建任务链

• LCEL是LangChain推出的链式表达式语言，支持用”|”操作符将各类单元（如Prompt、LLM、Parser等）组合。
• 每个”|”左侧的输出会自动作为右侧的输入，实现数据流式传递。

优势：

• 代码简洁，逻辑清晰，易于多步任务编排。
• 支持多分支、条件、并行等复杂链路。
• 易于插拔、复用和调试每个子任务。

典型用法：

• 串联：`A | B | C`，A的输出传给B，B的输出传给C。
• 分支：`{“x”: A, “y”: B}`，并行执行A和B。
• 支持流式：如`.stream()` 方法可边生成边打印。

示例Python代码：

from langchain_core.prompts import ChatPromptTemplate
from langchain_community.llms import Tongyi  # 导入通义千问Tongyi模型
from langchain_core.output_parsers import StrOutputParser
import dashscope
import os

# 从环境变量获取 dashscope 的 API Key
api_key = os.environ.get('DASHSCOPE_API_KEY')
dashscope.api_key = api_key

# stream=True 让LLM支持流式输出
llm = Tongyi(model_name="qwen-turbo", dashscope_api_key=api_key, stream=True)

# 定义三个子任务：翻译->处理->回译
translate_to_en = ChatPromptTemplate.from_template("Translate this to English: {input}") | llm | StrOutputParser()
process_text = ChatPromptTemplate.from_template("Analyze this text: {text}") | llm | StrOutputParser()
translate_to_cn = ChatPromptTemplate.from_template("Translate this to Chinese: {output}") | llm | StrOutputParser()

# 组合成多任务链
workflow = {"text": translate_to_en} | process_text | translate_to_cn
#workflow.invoke({"input": "北京有哪些好吃的地方，简略回答不超过200字"})

# 使用stream方法，边生成边打印
for chunk in workflow.stream({"input": "北京有哪些好吃的地方，简略回答不超过200字"}):
    print(chunk, end="", flush=True)
print()  # 换行