优秀的编程知识分享平台

网站首页 > 技术文章 正文

LangChain源码逐行解密之LLMs(二)

nanyue 2024-08-04 16:52:40 技术文章 9 ℃

18.3 base.py源码逐行剖析

现在我们要聚焦于源代码中的大语言模型部分。如图18-3所示，LangChain提供了许多语言模型的选择。 Gavin大咖微信：NLP_Matrix_Space

图18- 3 LangChain的llms目录

如图18-4所示，整个LangChain的模块化设计非常出色，你可以看到大模型（llms）、索引（indexes）、嵌入（embeddings）、评估（evaluation）等功能，这些都是非常好的模块化设计。

图18- 4 LangChain的模块化设计

Gavin大咖微信：NLP_Matrix_Space

另一方面，LangChain的结构设计也非常出色。直觉上，你会想要查看它的源代码，因为它有很多实现。当然，你的核心或出发点应该是查看base.py，只要你有基本的编程经验，就不会在这个地方遇到问题。因为base.py定义了它的协议，并提供了一些通用的操作或基本的工具类，最重要的是它提供了接口。其他模块的源代码要遵循这个接口，因为我们将模型放入LangChain或Agent的整个框架中，当框架调用你的代码时，是根据接口来调用的。

如果是一个语言模型，语言模型涉及到生命周期。当你调用语言模型时，需要进行初始化，LangChain的官网文档中清晰地介绍了这一点，大家可以更好地理解。LangChain提供了一个回调系统，允许挂接语言模型应用程序的各个阶段。这对于日志记录、监视、流式传输和其他任务非常有用。 Gavin大咖微信：NLP_Matrix_Space

  1. class BaseCallbackHandler:
  2. """可用于处理langchain回调的基本回调处理程序."""
  3. def on_llm_start(
  4. self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
  5. ) -> Any:
  6. """LLM开始运行时运行."""
  7. def on_chat_model_start(
  8. self, serialized: Dict[str, Any], messages: List[List[BaseMessage]], **kwargs: Any
  9. ) -> Any:
  10. """当聊天模型开始运行时运行"""
  11. def on_llm_new_token(self, token: str, **kwargs: Any) -> Any:
  12. """使用新的LLM标记运行。仅在启用流时可用."""
  13. def on_llm_end(self, response: LLMResult, **kwargs: Any) -> Any:
  14. """LLM结束运行时运行."""
  15. ...

以上代码定义了一个BaseCallbackHandler的基础回调处理器类。该类用于处理来自LangChain的回调。BaseCallbackHandler类包含了一些生命周期方法，每个方法都代表了一个不同的回调事件。

  • on_llm_start方法：当语言模型开始运行时调用。
  • on_chat_model_start方法：当聊天模型开始运行时调用。
  • on_llm_new_token方法：当语言模型有新的标记可用时调用。这个方法只在启用了流式传输时才可用。它接收一个标记字符串作为参数。
  • on_llm_end方法：当语言模型结束运行时调用。

回到语言模型的base.py代码，llms的base.py的代码实现：

  1. """大型语言模型的基本接口."""
  2. from __future__ import annotations
  3. import asyncio
  4. import inspect
  5. import json
  6. import logging
  7. import warnings
  8. from abc import ABC, abstractmethod
  9. from pathlib import Path
  10. from typing import (
  11. Any,
  12. Callable,
  13. Dict,
  14. List,
  15. Mapping,
  16. Optional,
  17. Sequence,
  18. Tuple,
  19. Type,
  20. Union,
  21. )
  22. import yaml
  23. from pydantic import Field, root_validator, validator
  24. from tenacity import (
  25. before_sleep_log,
  26. retry,
  27. retry_base,
  28. retry_if_exception_type,
  29. stop_after_attempt,
  30. wait_exponential,
  31. )
  32. import langchain
  33. from langchain.callbacks.base import BaseCallbackManager
  34. from langchain.callbacks.manager import (
  35. AsyncCallbackManager,
  36. AsyncCallbackManagerForLLMRun,
  37. CallbackManager,
  38. CallbackManagerForLLMRun,
  39. Callbacks,
  40. )
  41. from langchain.load.dump import dumpd
  42. from langchain.schema import (
  43. Generation,
  44. LLMResult,
  45. PromptValue,
  46. RunInfo,
  47. )
  48. from langchain.schema.language_model import BaseLanguageModel
  49. from langchain.schema.messages import AIMessage, BaseMessage, get_buffer_string
  50. ...
  51. def get_prompts(
  52. params: Dict[str, Any], prompts: List[str]
  53. ) -> Tuple[Dict[int, List], str, List[int], List[str]]:
  54. """获取已缓存的提示."""
  55. llm_string = str(sorted([(k, v) for k, v in params.items()]))
  56. missing_prompts = []
  57. missing_prompt_idxs = []
  58. existing_prompts = {}
  59. for i, prompt in enumerate(prompts):
  60. if langchain.llm_cache is not None:
  61. cache_val = langchain.llm_cache.lookup(prompt, llm_string)
  62. if isinstance(cache_val, list):
  63. existing_prompts[i] = cache_val
  64. else:
  65. missing_prompts.append(prompt)
  66. missing_prompt_idxs.append(i)
  67. return existing_prompts, llm_string, missing_prompt_idxs, missing_prompts
  68. ...

  70. def update_cache(
  71. existing_prompts: Dict[int, List],
  72. llm_string: str,
  73. missing_prompt_idxs: List[int],
  74. new_results: LLMResult,
  75. prompts: List[str],
  76. ) -> Optional[dict]:
  77. """更新缓存并获取LLM输出"""
  78. for i, result in enumerate(new_results.generations):
  79. existing_prompts[missing_prompt_idxs[i]] = result
  80. prompt = prompts[missing_prompt_idxs[i]]
  81. if langchain.llm_cache is not None:
  82. langchain.llm_cache.update(prompt, llm_string, result)
  83. llm_output = new_results.llm_output
  84. return llm_output

以上代码的第37行至第42行，要特别注意的第一个方面是管理器，它提供了一些方法来管理生命周期，你会看到与回调相关的内容，尤其是CallbackManager，这里还有异步的方式，因为有时候你可能想在后台执行一些任务，或者提高性能，异步操作非常重要。在作者看来，所有的服务器类编写都应该使用异步的方式，除非你处理的是主线程，不执行这一步，其他的任务都无法执行，否则的话，都应该是异步的方式，在使用Python编写代码时，Python的基本设计是单线程，为了实现多任务，Python引入了Coroutine（协程）的概念，同时还有一些线程和进程的方式，这些是Python编程的基础内容。

以上代码的第54行至第70行，第二个要注意的方面是get_prompts方法，这非常重要，因为与语言模型进行交互时，你是通过提示词（prompt）与它进行对话，从语言模型的角度来看，你所有的输入实际上都是提示词。例如：我想让系统教我关于托福阅读的内容，可以输入信息："Can you teach me about TOEFL reading?"（“你能教我托福阅读吗？”），这部分内容是提示词的一部分，但是它的核心是背后的提示词技术，提示词可以设计一些内部的思考过程，基于用户的输入和系统的设计，这涉及到很多复杂的内容，例如，自我评估（self-criticism）可以使用两个模型，进行同行评审（Peer Review），这意味着有两个模型，比如GPT-4、Llama或者其他一些模型，它们之间相互促进，这不是非此即彼的关系，而是提供了更多的视角，让对方给出评估，基于这种评估进行改进，这是一个不断循环的过程。这就是一个同行评审的流程。当然，作者在“TOEFL on ChatGPT”项目背后，还有很多其他的内容，这也是作者最近一两个月与许多机构和大学合作的一个核心要点。

要强调的一个关键点是，如果你想深入了解系统内部，在模型进行每一步输入之前的提示词，并进行调优，那么get_prompts方法是一个非常重要的方法。get_prompts函数接收两个参数：params和prompts。params是一个字典，存储了一些参数信息，prompts是一个字符串列表，包含了多个提示。首先将params字典中的键值对按照键进行排序，并将其转换为字符串llm_string。然后，遍历prompts列表中的每个提示。如果langchain.llm_cache不为None，则函数尝试通过调用langchain.llm_cache.lookup(prompt, llm_string)来查找缓存中的值。如果返回的cache_val是一个列表，说明该提示已经存在于缓存中，将该提示及其索引添加到existing_prompts字典中。否则，如果返回的cache_val不是列表，说明该提示缺失于缓存中，函数将该提示添加到missing_prompts列表中，并记录其索引到missing_prompt_idxs列表中。然后，函数返回四个值：existing_prompts（已存在于缓存的提示及其索引的字典）、llm_string（经过排序后的params的字符串表示）、missing_prompt_idxs（缺失的提示的索引列表）和missing_prompts（缺失的提示列表），这对于管理提示词信息是很重要的。 Gavin大咖微信：NLP_Matrix_Space

update_cache函数的作用是在使用语言模型进行生成时，更新缓存以提高效率并获取语言模型的输出结果。它通过缓存已生成的提示和结果，以便在以后的调用中可以直接从缓存中获取结果，而无需重新生成。函数接收以下参数：existing_prompts （现有提示的字典）、 llm_string （LLM字符串）、 missing_prompt_idxs （缺失提示的索引列表）、 new_results （LLM结果对象）和 prompts （提示列表），通过遍历 new_results.generations 中的结果，并将其存储到 existing_prompts 字典中，使用缺失提示的索引作为键。然后，根据相应的索引从 prompts 列表中获取提示，并在LLM缓存对象 langchain.llm_cache 不为None时更新缓存，将提示、LLM字符串和结果作为参数传递给缓存的 update 方法。然后，函数返回 new_results.llm_output 作为语言模型的输出。这些都是工具方法。

以上代码第81行，我们将使用enumerate(new_results.generations)进行遍历，大家对于enumerate应该都很熟悉，它会生成一个类似于生成器的对象，可以通过不断获取其中的内容来进行遍历。

以上代码第84行，其中langchain.llm_cache是基于一个BaseCache进行了封装。

langchain的__init__.py的代码实现：

  1. verbose: bool = False
  2. debug: bool = False
  3. llm_cache: Optional[BaseCache] = None

langchain的cache.py的BaseCache代码实现：

  1. class BaseCache(ABC):
  2. """缓存的基本接口."""
  3. @abstractmethod
  4. def lookup(self, prompt: str, llm_string: str) -> Optional[RETURN_VAL_TYPE]:
  5. """根据提示和llm_string查找"""
  6. @abstractmethod
  7. def update(self, prompt: str, llm_string: str, return_val: RETURN_VAL_TYPE) -> None:
  8. """基于提示和llm_string更新缓存."""
  9. @abstractmethod
  10. def clear(self, **kwargs: Any) -> None:
  11. """清除可以接受其他关键字参数的缓存"""

BaseCache 类定义了缓存的基本接口，是一个抽象基类（ABC），包含了三个抽象方法： lookup、update和clear，包括查找、更新和清空等操作，为缓存的实现提供了统一的规范。

如图18-5所示，LangChain提供了子类的一些实现，大家可以看一下相关的代码。

图18- 5 LangChain的缓存

例如，InMemoryCache子类可以实现自己的方法。

langchain的cache.py的InMemoryCache代码实现：

  1. class InMemoryCache(BaseCache):
  2. """将事物存储在内存中的缓存"""
  3. def __init__(self) -> None:
  4. """使用空缓存初始化."""
  5. self._cache: Dict[Tuple[str, str], RETURN_VAL_TYPE] = {}
  6. def lookup(self, prompt: str, llm_string: str) -> Optional[RETURN_VAL_TYPE]:
  7. """根据提示和llm_string查找."""
  8. return self._cache.get((prompt, llm_string), None)
  9. def update(self, prompt: str, llm_string: str, return_val: RETURN_VAL_TYPE) -> None:
  10. ""基于提示和llm_string更新缓存."""
  11. self._cache[(prompt, llm_string)] = return_val
  12. def clear(self, **kwargs: Any) -> None:
  13. """清除缓存."""
  14. self._cache = {}

在InMemoryCache的场景中，最简单的方式就是使用一个字典（dictionary），可以使用键值对进行操作，这非常重要的原因是你需要进行更新（update）和清空（clear）等相关操作，特别是查找（lookup），可以快速地检索。所以，你可以将它看作是一个字典，使用get方法来获取值，然后使用update方法进行更新。在更新时，你需要输入一个键，其中包括提示（prompt）和我们大型语言模型字符串的内容。至于clear方法，它非常简单粗暴地将缓存清空了。

我们回到base.py代码，BaseLLM类继承至BaseLanguageModel类，同时也继承了ABC抽象基类。在开发阶段，可以把verbose设置为True。设置了回调管理器callbacks和callback_manager。

Tags：

猜你喜欢

最近发表
标签列表