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

构建LangChain应用程序的示例代码：61、如何使用 LangChain 和 LangSmith 优化链

news 2025/7/21 11:20:09

本示例介绍如何使用 LangChain 和 LangSmith 优化链。

设置

我们将为 LangSmith 设置环境变量，并加载相关数据

import osos.environ["LANGCHAIN_PROJECT"] = "movie-qa"
# 设置 LANGCHAIN_PROJECT 环境变量为 "movie-qa"

import pandas as pd
# 导入 pandas 库

df = pd.read_csv("data/imdb_top_1000.csv")
# 从 CSV 文件中读取 IMDB 前 1000 部电影的数据

df["Released_Year"] = df["Released_Year"].astype(int, errors="ignore")
# 将 "Released_Year" 列转换为整数类型，忽略错误

创建初始检索链

我们将使用自查询检索器

from langchain.schema import Document
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddingsembeddings = OpenAIEmbeddings()
# 导入必要的库并创建 OpenAIEmbeddings 对象

records = df.to_dict("records")
documents = [Document(page_content=d["Overview"], metadata=d) for d in records]
# 将数据框转换为字典列表，并创建 Document 对象列表

vectorstore = Chroma.from_documents(documents, embeddings)
# 使用文档和嵌入创建 Chroma 向量存储

from langchain.chains.query_constructor.base import AttributeInfo
from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain_openai import ChatOpenAImetadata_field_info = [AttributeInfo(name="Released_Year",description="The year the movie was released",type="int",),AttributeInfo(name="Series_Title",description="The title of the movie",type="str",),AttributeInfo(name="Genre",description="The genre of the movie",type="string",),AttributeInfo(name="IMDB_Rating", description="A 1-10 rating for the movie", type="float"),
]
document_content_description = "Brief summary of a movie"
llm = ChatOpenAI(temperature=0)
retriever = SelfQueryRetriever.from_llm(llm, vectorstore, document_content_description, metadata_field_info, verbose=True
)
# 创建元数据字段信息列表，设置文档内容描述，创建 ChatOpenAI 对象，
# 并使用这些信息创建 SelfQueryRetriever 对象

from langchain_core.runnables import RunnablePassthrough
# 导入 RunnablePassthrough

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
# 导入 StrOutputParser 和 ChatPromptTemplate

prompt = ChatPromptTemplate.from_template("""Answer the user's question based on the below information:Information:{info}Question: {question}"""
)
generator = (prompt | ChatOpenAI() | StrOutputParser()).with_config(run_name="generator"
)
# 创建聊天提示模板和生成器

chain = (RunnablePassthrough.assign(info=(lambda x: x["question"]) | retriever) | generator
)
# 创建完整的链，包括检索和生成步骤

运行示例

通过链运行示例。这可以是手动的，也可以使用示例列表或生产流量

chain.invoke({"question": "what is a horror movie released in early 2000s"})
# 使用链调用一个示例问题

标注

现在，转到 LangSmith 并将这些示例标注为正确或不正确

创建数据集

我们现在可以从这些运行中创建数据集。

我们要做的是找到标记为正确的运行，然后从中获取子链。具体来说，是查询生成器子链和最终生成步骤

from langsmith import Clientclient = Client()
# 创建 LangSmith 客户端

runs = list(client.list_runs(project_name="movie-qa",execution_order=1,filter="and(eq(feedback_key, 'correctness'), eq(feedback_score, 1))",)
)len(runs)
# 获取标记为正确的运行列表并打印数量

gen_runs = []
query_runs = []
for r in runs:gen_runs.extend(list(client.list_runs(project_name="movie-qa",filter="eq(name, 'generator')",trace_id=r.trace_id,)))query_runs.extend(list(client.list_runs(project_name="movie-qa",filter="eq(name, 'query_constructor')",trace_id=r.trace_id,)))
# 从正确的运行中提取生成器运行和查询构造器运行

runs[0].inputs
# 打印第一个运行的输入

runs[0].outputs
# 打印第一个运行的输出

query_runs[0].inputs
# 打印第一个查询运行的输入

query_runs[0].outputs
# 打印第一个查询运行的输出

gen_runs[0].inputs
# 打印第一个生成运行的输入

gen_runs[0].outputs
# 打印第一个生成运行的输出

创建数据集

我们现在可以为查询生成和最终生成步骤创建数据集。
我们这样做是因为：(1) 我们可以检查数据点，(2) 如果需要，我们可以编辑它们，(3) 我们可以随时间添加到它们

client.create_dataset("movie-query_constructor")inputs = [r.inputs for r in query_runs]
outputs = [r.outputs for r in query_runs]client.create_examples(inputs=inputs, outputs=outputs, dataset_name="movie-query_constructor"
)
# 创建查询构造器数据集

client.create_dataset("movie-generator")inputs = [r.inputs for r in gen_runs]
outputs = [r.outputs for r in gen_runs]client.create_examples(inputs=inputs, outputs=outputs, dataset_name="movie-generator")
# 创建生成器数据集

使用少量示例

我们现在可以下载数据集并在未来的链中使用它们作为少量示例

examples = list(client.list_examples(dataset_name="movie-query_constructor"))
# 获取查询构造器数据集的示例列表

import jsondef filter_to_string(_filter):if "operator" in _filter:args = [filter_to_string(f) for f in _filter["arguments"]]return f"{_filter['operator']}({','.join(args)})"else:comparator = _filter["comparator"]attribute = json.dumps(_filter["attribute"])value = json.dumps(_filter["value"])return f"{comparator}({attribute}, {value})"
# 定义一个函数将过滤器转换为字符串

model_examples = []for e in examples:if "filter" in e.outputs["output"]:string_filter = filter_to_string(e.outputs["output"]["filter"])else:string_filter = "NO_FILTER"model_examples.append((e.inputs["query"],{"query": e.outputs["output"]["query"], "filter": string_filter},))
# 创建模型示例列表

retriever1 = SelfQueryRetriever.from_llm(llm,vectorstore,document_content_description,metadata_field_info,verbose=True,chain_kwargs={"examples": model_examples},
)
# 使用模型示例创建新的检索器

chain1 = (RunnablePassthrough.assign(info=(lambda x: x["question"]) | retriever1) | generator
)
# 创建新的链

chain1.invoke({"question": "what are good action movies made before 2000 but after 1997?"}
)
# 使用新链调用一个示例问题

总结

本文档介绍了如何使用 LangChain 和 LangSmith 来优化问答链。主要步骤包括:

设置环境和加载数据
创建初始检索链
运行示例并进行标注
创建数据集
使用少量示例进行优化

扩展知识

LangChain：是一个用于开发由语言模型驱动的应用程序的框架。它提供了一套工具和抽象，使得构建复杂的 AI 应用变得更加简单。
LangSmith：是一个开发平台，用于构建、测试和监控基于 LLM 的应用程序。它提供了可视化和分析工具，帮助开发者优化他们的 LLM 应用。
自查询检索器（SelfQueryRetriever）：这是一种高级检索器，能够理解自然语言查询并将其转换为结构化查询，以便从向量存储中检索相关文档。
少样本学习（Few-shot learning）：这是一种机器学习技术，模型可以从很少的训练样本中学习执行新任务。在本文中，我们使用少量示例来改进查询构造和生成过程。
向量存储：这是一种特殊类型的数据库，专门用于存储和检索向量嵌入。在本文中，我们使用 Chroma 作为向量存储。

这些技术和工具共同工作，可以创建更智能、更高效的问答系统，特别是在处理特定领域（如电影信息）的查询时。

查看全文

http://www.lryc.cn/news/396768.html