LangChain开发全流程剖析

接下来，我们再回到“get_prompt()”方法。在这个方法中，有系统提示词（system prompts）和用户提示词（user prompts），这是从相应的文件中读取的，从“system.prompt”文件中读取系统提示词（system_template），从“user.prompt”文件中读取用户提示词（human_template）。在第7行中，我们定义了两个输入变量：“query”和“df_info”。然后，将这些提示词信息和输入变量封装在一个“ChatPromptTemplate”对象中，会调用“SystemMessagePromptTemplate.from_template()”方法和“HumanMessagePromptTemplate.from_template()”方法，并返回该对象。

其中，from_template方法是BaseStringMessagePromptTemplate的一个类方法，这个方法接受一个字符串模板“template”和一个字符串“template_format”，还有其他的关键字参数。这个方法返回一个“MessagePromptTemplateT”类型的对象。我们使用“PromptTemplate.from_template()”方法从字符串模板中创建一个PromptTemplate对象。然后，使用这个对象作为参数来创建一个“MessagePromptTemplateT”类型的对象。这个对象将包含从模板中解析出来的内容，这个源码本身非常简单。Gavin大咖微信：NLP_Matrix_Space

chat.py的from_template方法的代码实现：

class BaseStringMessagePromptTemplate(BaseMessagePromptTemplate,ABC):prompt: StringPromptTemplateadditional_kwargs: dict = Field(default_factory=dict)...@classmethoddef from_template(cls: Type[MessagePromptTemplateT],template: str,template_format: str = "f-string",**kwargs: Any,) -> MessagePromptTemplateT:prompt = PromptTemplate.from_template(template, template_format=template_format)return cls(prompt=prompt, **kwargs)...

我们有了提示词(prompt)之后，就可以基于这个提示词和模型(ChatOpenAI)来创建一个LLMChain对象。ChatOpenAI模型是一个封装器(wrapper)，它封装了OpenAI的API，并提供了一个简单的接口来使用这个API。通过使用封装器，我们可以方便地访问和使用OpenAI的模型。

app.py的代码实现：

chain_create = LLMChain(llm=ChatOpenAI(temperature=0,model_name=OpenAI.model_name,openai_api_key=OpenAI.secret_key,),prompt=get_prompt(),)

定义一个ChatOpenAI类，这是一个基于OpenAI的大型语言模型的封装器(wrapper)，提供了一个简单的接口来与OpenAI的API进行交互。 Gavin大咖微信：NLP_Matrix_Space

openai.py的ChatOpenAI的代码实现：

class ChatOpenAI(BaseChatModel):""" OpenAI聊天大语言模型的封装器。要使用，应该安装“openai”python包，并且使用API键设置的环境变量OPENAI_API_KEY。 任何有效传递给openai.create调用的参数都可以传入，即使没有显式保存在此类上。示例:.. code-block:: pythonfrom langchain.chat_models import ChatOpenAIopenai = ChatOpenAI(model_name="gpt-3.5-turbo")"""@propertydef lc_secrets(self) -> Dict[str, str]:return {"openai_api_key": "OPENAI_API_KEY"}@propertydef lc_serializable(self) -> bool:return Trueclient: Any #: :meta private:model_name: str = Field(default="gpt-3.5-turbo", alias="model")"""要使用的模型名称."""temperature: float = 0.7"""要使用的采样温度."""model_kwargs: Dict[str, Any] = Field(default_factory=dict)"""保留任何对未明确指定的“create”调用有效的模型参数"""openai_api_key: Optional[str] = None""" API请求的基本URL路径，如果未使用代理或服务模拟器，请保留为空."""openai_api_base: Optional[str] = Noneopenai_organization: Optional[str] = None# 支持OpenAI的显式代理openai_proxy: Optional[str] = NoneRequest_timeout: Optional[Union[float, Tuple[float, float]]] = None"""请求OpenAI完成API超时。默认值为600秒"""max_retries: int = 6"""生成时要进行的最大重试次数."""streaming: bool = False"""是否流式传输结果"""n: int = 1"""为每个提示生成的聊天补全数"""max_tokens: Optional[int] = None"""要生成的最大标记数"""tiktoken_model_name: Optional[str] = None"""使用此类时传递给tiktoken的模型名称。tiktoken用于计算文档中的标记数量，以限制在一定的限制范围内。默认情况下，当设置为“无”时，这将与嵌入模型名称相同。然而，在某些情况下，可能希望使用tiktoken不支持的模型名称的Embedding类。这可能包括当使用Azure嵌入时，或者当使用许多模型提供程序中的一个时，这些模型提供程序公开了类似OpenAI的API，但具有不同的模型。在这种情况下，为了避免在调用tiktoken时出错，可以在此处指定要使用的模型名称。"""

在ChatOpenAI类中，我们定义了一些属性和参数来进行配置。其中包括模型名称(model_name)、温度(temperature)、模型参数(model_kwargs)、API密钥(openai_api_key)等。我们还可以设置一些可选参数，如代理设置(openai_proxy)、超时时间(request_timeout)、重试次数(max_retries)、最大生成标记数(max_tokens)等，这些都是你进行API调用的一些基本的参数。

chain_create是一个LLMChain实例，LLMChain类封装一个ChatOpenAI实例和一个提示词模板，它是针对LLM运行查询的链，使用的ChatOpenAI模型是一个大语言模型。 Gavin大咖微信：NLP_Matrix_Space

llm.py的LLMChain的代码实现：

class LLMChain(Chain):""" 针对LLM运行查询的链示例:.. code-block:: pythonfrom langchain import LLMChain, OpenAI, PromptTemplateprompt_template = "Tell me a {adjective} joke"prompt = PromptTemplate(input_variables=["adjective"], template=prompt_template)llm = LLMChain(llm=OpenAI(), prompt=prompt)"""

在“股价查询”的案例中，我们演示了如何使用大语言模型来生成响应，并逐步分析了这个过程，使用控制台(console)来展示日志信息，一步一步进行剖析，并讲解了提示词的两个部分：系统提示词和用户提示词。这和以上代码的示例是一样的，构建LLMChain实例的时候，传入OpenAI实例，以及提示词参数。

在app.py ask方法的第9行，chain_create实例调用run方法，run是LLMChain链的一个核心方法。

app.py的ask方法的代码实现：

...@post(path="/ask", name="ask", sync_to_thread=True)def ask(data: Annotated[Query, Body(media_type=RequestEncodingType.MULTI_PART)],) -> str:query = data.querydf_info = data.df_infochain_result = chain_create.run({"df_info": df_info,"query": query,})result = chain_result.split("```python")[-1][:-3].strip()logger.info(f'chain_result: n {chain_result}')return result

在run方法中，我们需要传入两个参数：df_info和query，在产生代码的过程中，它会参考你的信息，我们可以根据这些信息来指向正确的内容，以便更好地生成的响应（chain_result）。我们来看一下run方法的源代码，这个地方非常重要。Gavin大咖微信：NLP_Matrix_Space

chains的base.py的代码实现：

def run(self,*args: Any,callbacks: Callbacks = None,tags: Optional[List[str]] = None,metadata: Optional[Dict[str, Any]] = None,**kwargs: Any,) -> str:""" 当有单个字符串输出时执行链的方法。此方法与chain.__call__方法的主要区别，是指此方法只能用于返回单个字符串输出的链。如果Chain有更多的输出、非字符串输出，或者想将inputs/run一起返回，请使用Chain__call__。另一个区别是，此方法希望直接以位置参数或关键字参数的形式传递输入，而 Chain__call__ 期望具有所有输入的单个输入字典参数:*args: 如果链需要一个输入，则可以将其作为唯一的位置参数传入。callbacks: 用于此链运行的回调。在构造过程中，除了传递到链的回调之外，还会调用这些回调，但只有这些运行时回调才会传到对其他对象的调用tags: 传递到所有回调的字符串标记列表。在构造过程中，这些将作为附加传递给传递到链的标记，但只有这些运行时标记才会传到对其他对象的调用**kwargs: 如果链需要多个输入，则可以将它们作为关键字参数间接传递。返回:以字符串形式输出的链示例:.. code-block:: python# 假设我们有一个接受“问题”字符串的单个输入链chain.run("What's the temperature in Boise, Idaho?")# -> "博伊西的温度是..."# 假设我们有一个多输入链，它包含一个“问题”字符串和“上下文”字符串:question = "What's the temperature in Boise, Idaho?"context = "Weather report for Boise, Idaho on 07/03/23..."chain.run(question=question, context=context)# -> "博伊西的温度是...""""# 在开始时运行以确保这是可能的/已定义的_output_key = self._run_output_keyif args and not kwargs:if len(args) != 1:raise ValueError("`run` supports only one positional argument.")return self(args[0], callbacks=callbacks, tags=tags, metadata=metadata)[_output_key]if kwargs and not args:return self(kwargs, callbacks=callbacks, tags=tags, metadata=metadata)[_output_key]if not kwargs and not args:raise ValueError("`run` supported with either positional arguments or keyword arguments,"" but none were provided.")else:raise ValueError(f"`run` supported with either positional arguments or keyword arguments"f" but not both. Got args: {args} and kwargs: {kwargs}.")

run方法接受可变数量的参数和关键字参数，并且可以通过回调(callbacks)和标签(tags)进行配置。如果Chain对象只返回一个字符串输出，则可以使用该方法。如果Chain对象有多个输出、非字符串输出，或者你想将输入/运行信息与输出一起返回，则应该使用Chain.__call__方法。run方法的示例展示了如何使用该方法来运行Chain对象，并返回一个字符串输出。如果Chain对象需要多个输入，则可以将它们作为关键字参数传递给该方法。

Gavin大咖微信：NLP_Matrix_Space

以上代码中的注释提供了非常有价值的信息，在只有一个字符串输出的情况下执行链的run方法。在自然语言处理(NLP)中非常经典和流行的一种想法是，将NLP 任务都转化成文本到文本 （Text-to-Text）的任务，如果你熟悉像T5这样的NLP模型，你应该听说过这种想法，并且了解它是如何实现的。在这个基础接口中，我们可以看到一些基本的判断，这些是面向对象设计中基本的实现方式。父类中有一些通用的判断，例如，参数的问题等等，子类可以继承父类的方法。当然，在子类中调用父类的方法也是必要的。LLMChain是Chain类的子类。你只要做LangChain，看的关键肯定是run方法，当你调用run()方法时，它会运行内部的__call__方法。

....

....

....

....

接下来，我们来看一个使用openai.ChatCompletion的get_completion_from_messages的示例函数，它接受messages、model、temperature和max_tokens作为参数。

def get_completion_from_messages(messages,model="gpt-3.5-turbo",temperature=0, max_tokens=500):response = openai.ChatCompletion.create(model=model,messages=messages,temperature=temperature,max_tokens=max_tokens,)return response.choices[0].message["content"]

在上述代码的第4行，我们需要调用openai.ChatCompletion的哪个方法？请大声告诉我，是“create”方法，通过调用openai.ChatCompletion的create方法发送请求，使用指定的模型、消息、温度和最大标记数等参数。然后，将响应存储在response变量中，返回响应中第一条选择的消息内容。Gavin大咖微信：NLP_Matrix_Space

回到openai.py文件的completion_with_retry方法，self.client指的是openai.ChatCompletion。在这里，调用它的create方法，最终获取远程的结果。

openai.py的completion_with_retry方法的代码实现：

class ChatOpenAI(BaseChatModel):...def completion_with_retry(self, **kwargs: Any) -> Any:"""使用tenacity重试完成调用."""retry_decorator = self._create_retry_decorator()@retry_decoratordef _completion_with_retry(**kwargs: Any) -> Any:return self.client.create(**kwargs)return _completion_with_retry(**kwargs)

Gavin大咖微信：NLP_Matrix_Space

这个时候，你可能会感到非常兴奋和激动，因为你成功地理解了整个框架的运作方式，打通了整个框架的一切。我们通过这样一个应用程序，从用户的角度，从程序运行的角度，一步步进入框架，通过框架看见它是怎么去封装我们的OpenAI，以及怎么一步一步去调用的，

逐步了解OpenAI的封装和调用过程。在刚才的代码部分，我们调用了create方法，这正是openai.ChatCompletion作为OpenAI的原生调用方法。

我们通过一个应用程序，以小见大地展示了整个过程，带领读者一步一步深入源代码，抽丝剥茧的揭示了背后的整体布局。