19.9 semantic_similarity.py源码逐行剖析

LengthBasedExampleSelector类比较简单，我们看一下semantic_similarity.py具体是怎么操作的。Gavin大咖微信：NLP_Matrix_Space

semantic_similarity.py的代码实现：

1. """基于SemanticSimilarity选择示例的示例选择器."""
2. from __future__ import annotations
3. from typing import Any, Dict, List, Optional, Type
4. from pydantic import BaseModel, Extra
5. from langchain.embeddings.base import Embeddings
6. from langchain.prompts.example_selector.base import BaseExampleSelector
7. from langchain.vectorstores.base import VectorStore
8. def sorted_values(values: Dict[str, str]) -> List[Any]:
9. """返回字典中按键排序的值列表"""
10. return [values[val] for val in sorted(values)]
11. class SemanticSimilarityExampleSelector(BaseExampleSelector, BaseModel):
12. """基于SemanticSimilarity选择示例的示例选择器"""
13. vectorstore: VectorStore
14. """VectorStore包含有关示例的信息"""
15. k: int = 4
16. """要选择的示例数."""
17. example_keys: Optional[List[str]] = None
18. """用于筛选示例的可选键"""
19. input_keys: Optional[List[str]] = None
20. """用于筛选输入的可选键。如果提供，则搜索基于输入变量而不是所有变量."""
21. class Config:
22. """pydantic对象的配置"""
23. extra = Extra.forbid
24. arbitrary_types_allowed = True
25. def add_example(self, example: Dict[str, str]) -> str:
26. """将新示例添加到vectorstore."""
27. if self.input_keys:
28. string_example = " ".join(
29. sorted_values({key: example[key] for key in self.input_keys})
30. )
31. else:
32. string_example = " ".join(sorted_values(example))
33. ids = self.vectorstore.add_texts([string_example], metadatas=[example])
34. return ids[0]
35. def select_examples(self, input_variables: Dict[str, str]) -> List[dict]:
36. """根据语义相似性选择要使用的示例"""
37. # Get the docs with the highest similarity.
38. if self.input_keys:
39. input_variables = {key: input_variables[key] for key in self.input_keys}
40. query = " ".join(sorted_values(input_variables))
41. example_docs = self.vectorstore.similarity_search(query, k=self.k)
42. # 从元数据中获取示例
43. # 这假设示例存储在元数据中
44. examples = [dict(e.metadata) for e in example_docs]
45. # 如果提供了示例键，则将示例过滤到这些键。
46. if self.example_keys:
47. examples = [{k: eg[k] for k in self.example_keys} for eg in examples]
48. return examples
49. @classmethod
50. def from_examples(
51. cls,
52. examples: List[dict],
53. embeddings: Embeddings,
54. vectorstore_cls: Type[VectorStore],
55. k: int = 4,
56. input_keys: Optional[List[str]] = None,
57. **vectorstore_cls_kwargs: Any,
58. ) -> SemanticSimilarityExampleSelector:
59. """使用示例列表和嵌入创建k-shot示例选择器
60. 基于查询相似性动态重新混洗示例
61. 参数:
62. examples: 要在提示中使用的示例列表.
63. embeddings:一个初始化的嵌入API接口，例如OpenAIEmbeddings
64. vectorstore_cls: 一个向量存储DB接口类，例如FAISS.
65. k: 要选择的示例数
66. input_keys: 如果提供，则搜索基于输入变量，而不是所有变量
67. vectorstore_cls_kwargs: 包含向量存储url的可选kwargs
68. 返回:
69. ExampleSelector已实例化，由向量存储支持
70. """
71. if input_keys:
72. string_examples = [
73. " ".join(sorted_values({k: eg[k] for k in input_keys}))
74. for eg in examples
75. ]
76. else:
77. string_examples = [" ".join(sorted_values(eg)) for eg in examples]
78. vectorstore = vectorstore_cls.from_texts(
79. string_examples, embeddings, metadatas=examples, **vectorstore_cls_kwargs
80. )
81. return cls(vectorstore=vectorstore, k=k, input_keys=input_keys)
82. 
    

以上代码实现了一个基于语义相似性的示例选择器SemanticSimilarityExampleSelector，代码中导入了一些必要的模块和类，包括BaseModel、BaseExampleSelector、VectorStore、Embeddings等。定义了一个工具方法叫sorted_values，它会根据key进行一些排序操作，直接调用了sorted方法，返回按键排序的值列表。Gavin大咖微信：NLP_Matrix_Space

SemanticSimilarityExampleSelector类继承自BaseExampleSelector和BaseModel。

SemanticSimilarityExampleSelector类具有以下属性：

• vectorstore：包含有关示例的信息的向量存储。
• k：要选择的示例数，默认值为4。
• example_keys：用于筛选示例的可选键。
• input_keys：用于筛选输入的可选键。如果提供了这些键，则基于输入变量进行搜索，而不是所有变量。

SemanticSimilarityExampleSelector类具有以下方法：

• add_example方法，用于将新示例添加到向量存储中。根据是否提供了input_keys，将示例的值按键排序后进行拼接，然后将拼接后的字符串作为文本添加到向量存储中，并返回相应的标识符。上述代码中的第44行调用了vectorstore.add_texts方法。

vectorstores的base.by的代码实现：

1. class VectorStore(ABC):
2. """向量存储接口."""
3. @abstractmethod
4. def add_texts(
5. self,
6. texts: Iterable[str],
7. metadatas: Optional[List[dict]] = None,
8. **kwargs: Any,
9. ) -> List[str]:
10. """通过嵌入运行更多文本并添加到vectorstore中
11. 参数:
12. texts: 要添加到vectorstore的可迭代字符串
13. metadatas: 与文本相关联的元数据的可选列表
14. kwargs: vectorstore特定参数
15. 返回:
16. 将文本添加到向量存储中的ID列表
17. """

如图19-9所示，vectorstores可以有很多具体的实现，例如ClickHouse，对于从事大数据或者对时效性要求较高的人来说，对ClickHouse可能有更多了解。另外一种是Redis，当然还有其他一些的实现。Gavin大咖微信：NLP_Matrix_Space

图19- 9 vectorstores的具体实现

• select_examples方法，用于根据语义相似性选择要使用的示例。首先，根据输入变量的键值对按键排序后拼接成查询字符串。然后，使用向量存储的similarity_search方法，基于查询字符串找到与之相似度最高的示例文档。接下来，从示例文档的元数据中提取示例，并根据提供的example_keys对示例进行筛选。然后，返回被选中的示例列表。上述代码中的第53行调用了vectorstore.similarity_search方法。

vectorstores的base.by的similarity_search代码实现：

1. @abstractmethod
2. def similarity_search(
3. self, query: str, k: int = 4, **kwargs: Any
4. ) -> List[Document]:
5. """返回与查询最相似的文档"""

similarity_search方法是一个抽象方法，我们可以看一个similarity_search方法的具体实现，例如opensearch_vector_search.py的similarity search方法，它会调用similarity_search_with_score方法，然后调用_raw_similarity_search_with_score方法，在向量存储中进行相似性搜索，支持多种搜索类型，包括近似搜索、脚本评分和Painless脚本等，这是属于NLP的一些基础的内容。

Gavin大咖微信：NLP_Matrix_Space

opensearch_vector_search.py的similarity_search代码实现：

1. class OpenSearchVectorSearch(VectorStore):
2. ...
3. def similarity_search(
4. self, query: str, k: int = 4, **kwargs: Any
5. ) -> List[Document]:
6. """返回与查询最相似的文档
7. ...
8. """
9. docs_with_scores = self.similarity_search_with_score(query, k, **kwargs)
10. return [doc[0] for doc in docs_with_scores]

• from_examples方法，用于根据示例列表和嵌入创建基于语义相似性的示例选择器。该方法根据示例列表和输入的参数，创建一个向量存储，并将示例列表中的示例文本按键排序后拼接成字符串。然后，使用向量存储的from_texts方法将示例文本添加到向量存储中，并返回一个实例化的SemanticSimilarityExampleSelector对象。注意，这里面有一个很重要的操作叫“rehuffle”操作，基于示例与给定查询的相似性对其列表进行重新排序。示例不是按照原来的顺序显示，而是动态地重新排列，以确定与查询最相似或最相关的示例的优先级。Gavin大咖微信：NLP_Matrix_Space

接下来，我们看MaxMarginalRelevanceExampleSelector类的实现。

semantic_similarity.py的MaxMarginalRelevanceExampleSelector的代码实现：

1. class MaxMarginalRelevanceExampleSelector(SemanticSimilarityExampleSelector):
2. """ExampleSelector，根据最大边际相关性选择示例
3. 这在本文中被证明可以提高性能：
4. https://arxiv.org/pdf/2211.13892.pdf
5. """
6. fetch_k: int = 20
7. """要获取以重新排列的示例数"""
8. def select_examples(self, input_variables: Dict[str, str]) -> List[dict]:
9. """根据语义相似性选择要使用的示例."""
10. # 获取相似度最高的文档
11. if self.input_keys:
12. input_variables = {key: input_variables[key] for key in self.input_keys}
13. query = " ".join(sorted_values(input_variables))
14. example_docs = self.vectorstore.max_marginal_relevance_search(
15. query, k=self.k, fetch_k=self.fetch_k
16. )
17. # 从元数据中获取示例。
18. # 这假设示例存储在元数据中。
19. examples = [dict(e.metadata) for e in example_docs]
20. # 如果提供了示例键，则将示例过滤到这些键
21. if self.example_keys:
22. examples = [{k: eg[k] for k in self.example_keys} for eg in examples]
23. return examples
24. @classmethod
25. def from_examples(
26. cls,
27. examples: List[dict],
28. embeddings: Embeddings,
29. vectorstore_cls: Type[VectorStore],
30. k: int = 4,
31. input_keys: Optional[List[str]] = None,
32. fetch_k: int = 20,
33. **vectorstore_cls_kwargs: Any,
34. ) -> MaxMarginalRelevanceExampleSelector:
35. """使用示例列表和嵌入创建k-shot示例选择器
36. 基于查询相似性动态重新混洗示例
37. 参数:
38. examples: 要在提示中使用的示例列表
39. embeddings: 一个小型嵌入API接口，例如OpenAIEmbeddings
40. vectorstore_cls:一个向量存储DB接口类，例如FAISS
41. k:要选择的示例数
42. input_keys: 如果提供，则搜索基于输入变量，而不是所有变量
43. vectorstore_cls_kwargs: 包含向量存储url的可选kwargs
44. 返回:
45. ExampleSelector已实例化，由向量存储支持。
46. """
47. if input_keys:
48. string_examples = [
49. " ".join(sorted_values({k: eg[k] for k in input_keys}))
50. for eg in examples
51. ]
52. else:
53. string_examples = [" ".join(sorted_values(eg)) for eg in examples]
54. vectorstore = vectorstore_cls.from_texts(
55. string_examples, embeddings, metadatas=examples, **vectorstore_cls_kwargs
56. )
57. return cls(vectorstore=vectorstore, k=k, fetch_k=fetch_k, input_keys=input_keys)

以上代码是MaxMarginalRelevanceExampleSelector的示例选择器类，用于根据最大边际相关性选择示例。该类实现了select_examples方法和from_examples方法。

select_examples方法根据语义相似性选择要使用的示例。首先，根据输入变量获取查询文本。然后，使用向量存储的max_marginal_relevance_search方法获取与查询最相似的示例文档。接下来，从示例文档的元数据中提取示例。如果提供了示例键，将仅保留指定键的示例。然后，返回选择的示例列表。Gavin大咖微信：NLP_Matrix_Space

from_examples方法，用于基于示例列表和嵌入创建k-shot示例选择器。它使用查询相似性动态重新混洗示例。首先，将示例列表转换为字符串形式，并根据输入键进行排序和连接。然后，使用向量存储的from_texts方法将字符串示例转换为向量表示，并将示例元数据作为附加信息传递。然后，实例化MaxMarginalRelevanceExampleSelector类，并返回该实例。

Gavin大咖微信：NLP_Matrix_Space

19.10 ngram_overlap.py源码逐行剖析

接下来，我们基于ngram重叠分数的示例选择器NgramOverlapExampleSelector。

ngram_overlap.py的代码实现：

1. """根据ngram重叠分数（sentence_bleu分数）选择并排序示例
2. https://www.nltk.org/_modules/nltk/translate/bleu_score.html
3. https://aclanthology.org/P02-1040.pdf
4. """
5. from typing import Dict, List
6. import numpy as np
7. from pydantic import BaseModel, root_validator
8. from langchain.prompts.example_selector.base import BaseExampleSelector
9. from langchain.prompts.prompt import PromptTemplate
10. def ngram_overlap_score(source: List[str], example: List[str]) -> float:
11. """计算源和示例的ngram重叠分数作为sentence_bleu分数
12. 将sentence_bleu与method1平滑函数和自动重加权一起使用。
13. 返回介于0.0和1.0之间（包括0.0和1.0）的浮点值。
14. https://www.nltk.org/_modules/nltk/translate/bleu_score.html
15. https://aclanthology.org/P02-1040.pdf
16. """
17. from nltk.translate.bleu_score import (
18. SmoothingFunction, # type: ignore
19. sentence_bleu,
20. )
21. hypotheses = source[0].split()
22. references = [s.split() for s in example]
23. return float(
24. sentence_bleu(
25. references,
26. hypotheses,
27. smoothing_function=SmoothingFunction().method1,
28. auto_reweigh=True,
29. )
30. )
31. class NGramOverlapExampleSelector(BaseExampleSelector, BaseModel):
32. """根据ngram重叠分数（sentence_bleu分数）选择并排序示例。
33. https://www.nltk.org/_modules/nltk/translate/bleu_score.html
34. https://aclanthology.org/P02-1040.pdf
35. """
36. examples: List[dict]
37. """提示模板所需的示例列表."""
38. example_prompt: PromptTemplate
39. """用于格式化示例的提示模板."""
40. threshold: float = -1.0
41. """算法停止的阈值。默认情况下设置为-1.0
42. 对于负阈值：select_examples按ngram_overlap_score对示例进行排序，但不包括任何示例。对于大于1.0的阈值：select_examples排除所有示例，并返回一个空列表。对于等于0.0的阈值：select_examples按ngram_overlap_score对示例进行排序，并排除与输入没有ngram重叠的示例。
43. """
44. @root_validator(pre=True)
45. def check_dependencies(cls, values: Dict) -> Dict:
46. """检查是否存在有效的依赖项."""
47. try:
48. from nltk.translate.bleu_score import ( # noqa: disable=F401
49. SmoothingFunction,
50. sentence_bleu,
51. )
52. except ImportError as e:
53. raise ValueError(
54. "Not all the correct dependencies for this ExampleSelect exist"
55. ) from e
56. return values
57. def add_example(self, example: Dict[str, str]) -> None:
58. """将新示例添加到列表中"""
59. self.examples.append(example)
60. def select_examples(self, input_variables: Dict[str, str]) -> List[dict]:
61. """ 通过输入返回按ngram_overlap_score排序的示例列表。
62. 降序。排除ngram_overlap_score小于或等于阈值的任何示例。
63. """
64. inputs = list(input_variables.values())
65. examples = []
66. k = len(self.examples)
67. score = [0.0] * k
68. first_prompt_template_key = self.example_prompt.input_variables[0]
69. for i in range(k):
70. score[i] = ngram_overlap_score(
71. inputs, [self.examples[i][first_prompt_template_key]]
72. )
73. while True:
74. arg_max = np.argmax(score)
75. if (score[arg_max] < self.threshold) or abs(
76. score[arg_max] - self.threshold
77. ) < 1e-9:
78. break
79. examples.append(self.examples[arg_max])
80. score[arg_max] = self.threshold - 1.0
81. return examples

以上代码是一个基于ngram重叠分数的示例选择器。ngram_overlap_score函数：计算源文本和示例之间的ngram重叠分数，使用了sentence_bleu函数来计算BLEU分数。它使用了NLTK库中的SmoothingFunction来进行平滑处理，并使用自动重加权参数。返回值为0.0到1.0之间的浮点数。

NGramOverlapExampleSelector类继承自BaseExampleSelector和BaseModel，用于选择和排序示例。

NGramOverlapExampleSelector类的属性：

• examples：一个示例列表。
• example_prompt：一个用于格式化示例的提示模板。
• threshold：一个阈值，默认值为-1.0，用于控制选择示例的停止条件。你可以根据具体需求设置阈值来控制示例选择的行为。如果你希望保留所有示例，可以使用负阈值。如果你希望排除所有示例，可以使用大于1.0的阈值。如果你希望仅选择与输入有ngram重叠的示例，可以使用等于0.0的阈值。这都是NLP的基础知识。对于负阈值（默认为-1.0）：select_examples函数会按照ngram重叠分数对示例进行排序，但不排除任何示例。对于大于1.0的阈值：select_examples函数会排除所有示例，并返回一个空列表。对于等于0.0的阈值：select_examples函数会按照ngram重叠分数对示例进行排序，并排除与输入没有ngram重叠的示例。

NGramOverlapExampleSelector类的方法：

• check_dependencies方法：用于检查是否存在所需的依赖项，包括NLTK库中的BLEU计算函数。
• add_example方法：用于将新的示例添加到示例列表中。
• select_examples方法：根据输入的文本返回按照ngram重叠分数排序的示例列表。它首先计算每个示例与输入之间的ngram重叠分数，并将分数存储在一个列表中。然后，它选择分数最高的示例，将其添加到结果列表中，并将该示例的分数设置为低于阈值。重复此过程，直到找不到分数高于阈值的示例为止。

这是LangChain框架源码中提示词（Prompts）的相关内容。只要你写Prompts，基本上都会涉及到这里面的内容。本节中介绍的每个点都在我们的应用程序中得到了应用，特别是示例选择器（Example Selector）。