最新LangChain+GLM4开发AI应用程序系列（一）：快速入门篇_langchain调用chatglm4

一、前言

要想开发出优秀的AI应用程序，除了有性能强大的大语言模型(后续简称LLM)作基础，优秀的提示词工程技术，借助检索增强生成从外部知识库获取LLM不具备的专有知识，通过智能体对接外部工具扩展LLM能力，缺一不可。此外AI应用程序通常都需要经过多轮提示以及解析输出，并有机连接成一个整体，本质上是个链的过程，如何快速高效实现也是一个问题。

基于以上这些痛点，哈里森·蔡斯 (Harrison Chase) 于2022年10月创建了LangChain，使我们能更方便和快捷的开发出功能强大的AI应用程序。可以说，LangChain是目前最强的AI应用程序开发框架之一，从Java程序员的视角来说，它的地位就相当于Java界的Spring一样。

如何学习LangChain，官网上的使用说明文档最权威，但是官网文档的一大特定或者说是弊端就是，知识点不聚焦，比较零散，初学者一上来很难抓住重点。此外LangChain迭代速度很快，网上虽然也有不少此类文章，但大部分都有所过时。

基于此，我实践并整理了最新版的LangChain结合国内性价比最高的智谱AI GLM-4大模型的系列文章，希望能给大家起到一定的参考作用，同时也欢迎共同探讨。本篇文章是第一篇，从零开始带大家快速入门LangChain开发AI应用程序。

二、LangChain简介

LangChain本质上是一个用于开发由LLM驱动的应用程序的框架。它通过连接LLM和外部一切可以利用的能力，给应用程序赋能，使得应用程序能够具备上下文意识和推理能力。

整个LangChain由六层组成，架构图如下。

最底层是协议层，包括LangChain表达语言(LCEL)和一些基础抽象类。主要是提示词模板、输出解析器、链的抽象类，通过LCEL语法以及管道的概念，来简化代码的开发。

第二层是集成组件层，包括各种大语言模型输入输出、外部知识库检索、智能体工具等第三方集成。具体的比如智谱AI大语言模型包装类、嵌入模型HuggingFaceEmbeddings、向量数据库FAISS等，都在这一层。

第三层是认知架构层，包括构成应用程序认知架构的链、智能体和检索策略。主要是链(Chains)、智能体(Agents)和检索策略的具体实现类。

第四层是应用模板层，包括各种参考应用模板。有多种可以快速运行部署的应用示例，参考价值还是比较高的。

第五层是部署层，LangServe，是一个用于将LangChain部署为REST API的库。

第六层是表现层，LangSmith，一个开发平台，允许调试、测试、评估和监控LangChain应用。

三、环境准备

在使用LangChain进行AI应用程序开发前，需要准备好相应的开发环境，包括Conda、Jupyter Notebook、使用的智谱AI GLM-4大模型。

1、安装Conda

使用Python的人都会遇到库的安装、环境的管理问题，Conda就是解决这些问题的一个工具，目前有AnaConda和MiniConda两种，都是Continuum Analytics的开源项目。这两种的区别就是：AnaConda大而全，预装了大部分科学计算库(99%其实用不上)，提供了图形界面，适合初学者；MiniConda小而精，只有命令行。个人推荐使用MiniConda，比较轻量，节省时间和空间，需要什么再安装就可以。

MiniConda的下载地址如下：https://docs.conda.io/projects/miniconda/en/latest/ ，windows和mac版本都包括了，下载好直接安装即可。

创建虚拟环境：

conda create -n langchain python=3.9
1

激活虚拟环境：

conda activate langchain
1

2、安装Jupyter Notebook

为了能更方便编程和调试，及时执行代码块，我们使用Jupyter Notebook来作为Python集成开发工具。

在虚拟环境中安装Jupyter Notebook：

conda install jupyter notebook
1

启动Jupyter notebook：

jupyter notebook
1

3、安装LangChain

在Jupyter Notebook中进行操作，安装LangChain的最小集。

pip install --upgrade langchain
1

4、调用智谱AI的GLM-4大模型

LLM的选择有多种方案。

方案一：远程调用OpenAI的ChatGPT系统API，效果较好，token花费较贵；

方案二：远程调用智谱AI的GLM-4的API，效果较好，token花费较低；

方案三：本地部署开源大语言模型ChatGLM3-6B，效果较差，不需要收费，但电脑需要有13GB以上的GPU。

综合考虑，方案二最理想。远程调用智谱AI的GLM-4的API的方式门槛最低，提示词工程的效果也比较好。目前智谱AI新注册会赠送18元金额，GLM-4是0.1元/1000tokens，实名认证的话会再送400万tokens(一个月内使用有效)，算比较经济实惠的。注册地址：https://open.bigmodel.cn/。可以注册申请API key。

下面的操作都在Jupyter Notebook中进行。

安装智谱的SDK包：

pip install zhipuai
1

由于最新的LangChain 0.1.7集成的ChatZhipuAI类和最新zhipuai SDK版本兼容性方面有问题，需要重新包装一个类。代码如下：

"""ZHIPU AI chat models wrapper."""
from __future__ import annotations

import asyncio
import logging
from functools import partial
from importlib.metadata import version
from typing import (
    Any,
    Callable,
    Dict,
    Iterator,
    List,
    Mapping,
    Optional,
    Tuple,
    Type,
    Union,
)

from langchain_core.callbacks import (
    AsyncCallbackManagerForLLMRun,
    CallbackManagerForLLMRun,
)
from langchain_core.language_models.chat_models import (
    BaseChatModel,
    generate_from_stream,
)
from langchain_core.language_models.llms import create_base_retry_decorator
from langchain_core.messages import (
    AIMessage,
    AIMessageChunk,
    BaseMessage,
    BaseMessageChunk,
    ChatMessage,
    ChatMessageChunk,
    HumanMessage,
    HumanMessageChunk,
    SystemMessage,
    SystemMessageChunk,
    ToolMessage,
    ToolMessageChunk,
)
from langchain_core.outputs import (
    ChatGeneration,
    ChatGenerationChunk,
    ChatResult,
)
from langchain_core.pydantic_v1 import BaseModel, Field
from packaging.version import parse

logger = logging.getLogger(__name__)


def is_zhipu_v2() -> bool:
    """Return whether zhipu API is v2 or more."""
    _version = parse(version("zhipuai"))
    return _version.major >= 2


def _create_retry_decorator(
    llm: ChatZhipuAI,
    run_manager: Optional[
        Union[AsyncCallbackManagerForLLMRun, CallbackManagerForLLMRun]
    ] = None,
) -> Callable[[Any], Any]:
    import zhipuai

    errors = [
        zhipuai.ZhipuAIError,
        zhipuai.APIStatusError,
        zhipuai.APIRequestFailedError,
        zhipuai.APIReachLimitError,
        zhipuai.APIInternalError,
        zhipuai.APIServerFlowExceedError,
        zhipuai.APIResponseError,
        zhipuai.APIResponseValidationError,
        zhipuai.APITimeoutError,
    ]
    return create_base_retry_decorator(
        error_types=errors, max_retries=llm.max_retries, run_manager=run_manager
    )


def convert_message_to_dict(message: BaseMessage) -> dict:
    """Convert a LangChain message to a dictionary.

    Args:
        message: The LangChain message.

    Returns:
        The dictionary.
    """
    message_dict: Dict[str, Any]
    if isinstance(message, ChatMessage):
        message_dict = {"role": message.role, "content": message.content}
    elif isinstance(message, HumanMessage):
        message_dict = {"role": "user", "content": message.content}
    elif isinstance(message, AIMessage):
        message_dict = {"role": "assistant", "content": message.content}
        if "tool_calls" in message.additional_kwargs:
            message_dict["tool_calls"] = message.additional_kwargs["tool_calls"]
            # If tool calls only, content is None not empty string
            if message_dict["content"] == "":
                message_dict["content"] = None
    elif isinstance(message, SystemMessage):
        message_dict = {"role": "system", "content": message.content}
    elif isinstance(message, ToolMessage):
        message_dict = {
            "role": "tool",
            "content": message.content,
            "tool_call_id": message.tool_call_id,
        }
    else:
        raise TypeError(f"Got unknown type {message}")
    if "name" in message.additional_kwargs:
        message_dict["name"] = message.additional_kwargs["name"]
    return message_dict


def convert_dict_to_message(_dict: Mapping[str, Any]) -> BaseMessage:
    """Convert a dictionary to a LangChain message.

    Args:
        _dict: The dictionary.

    Returns:
        The LangChain message.
    """
    role = _dict.get("role")
    if role == "user":
        return HumanMessage(content=_dict.get("content", ""))
    elif role == "assistant":
        content = _dict.get("content", "") or ""
        additional_kwargs: Dict = {}
        if tool_calls := _dict.get("tool_calls"):
            additional_kwargs["tool_calls"] = tool_calls
        return AIMessage(content=content, additional_kwargs=additional_kwargs)
    elif role == "system":
        return SystemMessage(content=_dict.get("content", ""))
    elif role == "tool":
        additional_kwargs = {}
        if "name" in _dict:
            additional_kwargs["name"] = _dict["name"]
        return ToolMessage(
            content=_dict.get("content", ""),
            tool_call_id=_dict.get("tool_call_id"),
            additional_kwargs=additional_kwargs,
        )
    else:
        return ChatMessage(content=_dict.get("content", ""), role=role)


def _convert_delta_to_message_chunk(
    _dict: Mapping[str, Any], default_class: Type[BaseMessageChunk]
) -> BaseMessageChunk:
    role = _dict.get("role")
    content = _dict.get("content") or ""
    additional_kwargs: Dict = {}
    if _dict.get("tool_calls"):
        additional_kwargs["tool_calls"] = _dict["tool_calls"]

    if role == "user" or default_class == HumanMessageChunk:
        return HumanMessageChunk(content=content)
    elif role == "assistant" or default_class == AIMessageChunk:
        return AIMessageChunk(content=content, additional_kwargs=additional_kwargs)
    elif role == "system" or default_class == SystemMessageChunk:
        return SystemMessageChunk(content=content)
    elif role == "tool" or default_class == ToolMessageChunk:
        return ToolMessageChunk(content=content, tool_call_id=_dict["tool_call_id"])
    elif role or default_class == ChatMessageChunk:
        return ChatMessageChunk(content=content, role=role)
    else:
        return default_class(content=content)


class ChatZhipuAI(BaseChatModel):
    """
    `ZHIPU AI` large language chat models API.

    To use, you should have the ``zhipuai`` python package installed.

    Example:
    .. code-block:: python

    from langchain_community.chat_models import ChatZhipuAI

    zhipuai_chat = ChatZhipuAI(
        temperature=0.5,
        api_key="your-api-key",
        model_name="glm-3-turbo",
    )

    """

    zhipuai: Any
    zhipuai_api_key: Optional[str] = Field(default=None, alias="api_key")
    """Automatically inferred from env var `ZHIPUAI_API_KEY` if not provided."""

    client: Any = Field(default=None, exclude=True)  #: :meta private:

    model_name: str = Field("glm-3-turbo", alias="model")
    """
    Model name to use.
    -glm-3-turbo:
        According to the input of natural language instructions to complete a
        variety of language tasks, it is recommended to use SSE or asynchronous
        call request interface.
    -glm-4:
        According to the input of natural language instructions to complete a
        variety of language tasks, it is recommended to use SSE or asynchronous
        call request interface.
    """

    temperature: float = Field(0.95)
    """
    What sampling temperature to use. The value ranges from 0.0 to 1.0 and cannot
    be equal to 0.
    The larger the value, the more random and creative the output; The smaller
    the value, the more stable or certain the output will be.
    You are advised to adjust top_p or temperature parameters based on application
    scenarios, but do not adjust the two parameters at the same time.
    """

    top_p: float = Field(0.7)
    """
    Another method of sampling temperature is called nuclear sampling. The value
    ranges from 0.0 to 1.0 and cannot be equal to 0 or 1.
    The model considers the results with top_p probability quality tokens.
    For example, 0.1 means that the model decoder only considers tokens from the
    top 10% probability of the candidate set.
    You are advised to adjust top_p or temperature parameters based on application
    scenarios, but do not adjust the two parameters at the same time.
    """

    request_id: Optional[str] = Field(None)
    """
    Parameter transmission by the client must ensure uniqueness; A unique
    identifier used to distinguish each request, which is generated by default
    by the platform when the client does not transmit it.
    """
    do_sample: Optional[bool] = Field(True)
    """
    When do_sample is true, the sampling policy is enabled. When do_sample is false,
    the sampling policy temperature and top_p are disabled
    """
    streaming: bool = Field(False)
    """Whether to stream the results or not."""

    model_kwargs: Dict[str, Any] = Field(default_factory=dict)
    """Holds any model parameters valid for `create` call not explicitly specified."""

    max_tokens: Optional[int] = None
    """Number of chat completions to generate for each prompt."""

    max_retries: int = 2
    """Maximum number of retries to make when generating."""

    @property
    def _identifying_params(self) -> Dict[str, Any]:
        """Get the identifying parameters."""
        return {**{"model_name": self.model_name}, **self._default_params}

    @property
    def _llm_type(self) -> str:
        """Return the type of chat model."""
        return "zhipuai"

    @property
    def lc_secrets(self) -> Dict[str, str]:
        return {"zhipuai_api_key": "ZHIPUAI_API_KEY"}

    @classmethod
    def get_lc_namespace(cls) -> List[str]:
        """Get the namespace of the langchain object."""
        return ["langchain", "chat_models", "zhipuai"]

    @property
    def lc_attributes(self) -> Dict[str, Any]:
        attributes: Dict[str, Any] = {}

        if self.model_name:
            attributes["model"] = self.model_name

        if self.streaming:
            attributes["streaming"] = self.streaming

        if self.max_tokens:
            attributes["max_tokens"] = self.max_tokens

        return attributes

    @property
    def _default_params(self) -> Dict[str, Any]:
        """Get the default parameters for calling ZhipuAI API."""
        params = {
            "model": self.model_name,
            "stream": self.streaming,
            "temperature": self.temperature,
            "top_p": self.top_p,
            "do_sample": self.do_sample,
            **self.model_kwargs,
        }
        if self.max_tokens is not None:
            params["max_tokens"] = self.max_tokens
        return params

    @property
    def _client_params(self) -> Dict[str, Any]:
        """Get the parameters used for the zhipuai client."""
        zhipuai_creds: Dict[str, Any] = {
            "request_id": self.request_id,
        }
        return {**self._default_params, **zhipuai_creds}

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        try:
            from zhipuai import ZhipuAI

            if not is_zhipu_v2():
                raise RuntimeError(
                    "zhipuai package version is too low"
                    "Please install it via 'pip install --upgrade zhipuai'"
                )

            self.client = ZhipuAI(
                api_key=self.zhipuai_api_key,  # 填写您的 APIKey
            )
        except ImportError:
            raise RuntimeError(
                "Could not import zhipuai package. "
                "Please install it via 'pip install zhipuai'"
            )

    def completions(self, **kwargs) -> Any | None:
        return self.client.chat.completions.create(**kwargs)

    async def async_completions(self, **kwargs) -> Any:
        loop = asyncio.get_running_loop()
        partial_func = partial(self.client.chat.completions.create, **kwargs)
        response = await loop.run_in_executor(
            None,
            partial_func,
        )
        return response

    async def async_completions_result(self, task_id):
        loop = asyncio.get_running_loop()
        response = await loop.run_in_executor(
            None,
            self.client.asyncCompletions.retrieve_completion_result,
            task_id,
        )
        return response

    def _create_chat_result(self, response: Union[dict, BaseModel]) -> ChatResult:
        generations = []
        if not isinstance(response, dict):
            response = response.dict()
        for res in response["choices"]:
            message = convert_dict_to_message(res["message"])
            generation_info = dict(finish_reason=res.get("finish_reason"))
            if "index" in res:
                generation_info["index"] = res["index"]
            gen = ChatGeneration(
                message=message,
                generation_info=generation_info,
            )
            generations.append(gen)
        token_usage = response.get("usage", {})
        llm_output = {
            "token_usage": token_usage,
            "model_name": self.model_name,
            "task_id": response.get("id", ""),
            "created_time": response.get("created", ""),
        }
        return ChatResult(generations=generations, llm_output=llm_output)

    def _create_message_dicts(
        self, messages: List[BaseMessage], stop: Optional[List[str]]
    ) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]:
        params = self._client_params
        if stop is not None:
            if "stop" in params:
                raise ValueError("`stop` found in both the input and default params.")
            params["stop"] = stop
        message_dicts = [convert_message_to_dict(m) for m in messages]
        return message_dicts, params

    def completion_with_retry(
        self, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any
    ) -> Any:
        """Use tenacity to retry the completion call."""

        retry_decorator = _create_retry_decorator(self, run_manager=run_manager)

        @retry_decorator
        def _completion_with_retry(**kwargs: Any) -> Any:
            return self.completions(**kwargs)

        return _completion_with_retry(**kwargs)

    async def acompletion_with_retry(
        self,
        run_manager: Optional[AsyncCallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> Any:
        """Use tenacity to retry the async completion call."""

        retry_decorator = _create_retry_decorator(self, run_manager=run_manager)

        @retry_decorator
        async def _completion_with_retry(**kwargs: Any) -> Any:
            return await self.async_completions(**kwargs)

        return await _completion_with_retry(**kwargs)

    def _generate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        stream: Optional[bool] = None,
        **kwargs: Any,
    ) -> ChatResult:
        """Generate a chat response."""

        should_stream = stream if stream is not None else self.streaming
        if should_stream:
            stream_iter = self._stream(
                messages, stop=stop, run_manager=run_manager, **kwargs
            )
            return generate_from_stream(stream_iter)

        message_dicts, params = self._create_message_dicts(messages, stop)
        params = {
            **params,
            **({"stream": stream} if stream is not None else {}),
            **kwargs,
        }
        response = self.completion_with_retry(
            messages=message_dicts, run_manager=run_manager, **params
        )
        return self._create_chat_result(response)

    async def _agenerate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        stream: Optional[bool] = False,
        **kwargs: Any,
    ) -> ChatResult:
        """Asynchronously generate a chat response."""
        should_stream = stream if stream is not None else self.streaming
        if should_stream:
            stream_iter = self._astream(
                messages, stop=stop, run_manager=run_manager, **kwargs
            )
            return generate_from_stream(stream_iter)

        message_dicts, params = self._create_message_dicts(messages, stop)
        params = {
            **params,
            **({"stream": stream} if stream is not None else {}),
            **kwargs,
        }
        response = await self.acompletion_with_retry(
            messages=message_dicts, run_manager=run_manager, **params
        )
        return self._create_chat_result(response)

    def _stream(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> Iterator[ChatGenerationChunk]:
        """Stream the chat response in chunks."""
        message_dicts, params = self._create_message_dicts(messages, stop)
        params = {**params, **kwargs, "stream": True}

        default_chunk_class = AIMessageChunk
        for chunk in self.completion_with_retry(
            messages=message_dicts, run_manager=run_manager, **params
        ):
            if not isinstance(chunk, dict):
                chunk = chunk.dict()
            if len(chunk["choices"]) == 0:
                continue
            choice = chunk["choices"][0]
            chunk = _convert_delta_to_message_chunk(
                choice["delta"], default_chunk_class
            )

            finish_reason = choice.get("finish_reason")
            generation_info = (
                dict(finish_reason=finish_reason) if finish_reason is not None else None
            )
            default_chunk_class = chunk.__class__
            chunk = ChatGenerationChunk(message=chunk, generation_info=generation_info)
            yield chunk
            if run_manager:
                run_manager.on_llm_new_token(chunk.text, chunk=chunk)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506

创建调用的对话大模型对象，api key需要修改成你自己的：

# 填写您自己的APIKey
ZHIPUAI_API_KEY = "..."
llm = ChatZhipuAI(
    temperature=0.1,
    api_key=ZHIPUAI_API_KEY,
    model_name="glm-4",
)
1
2
3
4
5
6
7

四、AI应用程序开发入门

环境准备好之后，就可以开始使用LangChain进行AI应用程序开发了。LangChain其实就是通过各种链(Chain)将外部数据和计算连接到LLM，从而构建面向AI的应用程序。AI应用程序开发的入门，介绍了简单LLM链、检索链、智能体三个示例程序，通过一步一步操作，让大家快速入门。

1、简单LLM链(LLM Chain)

1）上面已经安装并创建了智谱AI大模型，现在可以直接调用它。

llm.invoke("langsmith如何帮助测试?")
1

结果：

AIMessage(content='LangSmith 是一个旨在帮助开发者将大语言模型（LLM）应用程序从原型阶段推进到生产阶段的全栈开发平台。在测试方面，LangSmith 提供了一系列功能和工具来支持开发者对 LLM 应用进行全面的测试，确保其性能、可靠性和稳定性。以下是 LangSmith 在测试方面如何提供帮助的几个方面：\n\n1. **调试工具**：LangSmith 为开发者提供了调试工具，可以帮助他们理解模型的输出以及其在特定输入下的行为。这包括错误追踪和日志分析功能，使得开发者可以快速定位并解决模型在运行时出现的问题。\n\n2. **跟踪功能**：它允许开发者跟踪模型的使用情况，包括性能指标和错误率，从而帮助开发者监控模型在实际使用中的表现。\n\n3. **集成测试框架**：LangSmith 可能提供了集成测试框架，使得开发者可以编写测试用例来验证模型的预期行为。这包括单元测试和集成测试，以确保模型在各种条件下都能正确运行。\n\n4. **Prompt 优化**：在生成式语言模型中，Prompt（提示）的设计对输出质量至关重要。LangSmith 可能提供了Prompt优化工具，帮助开发者测试和比较不同Prompt对模型输出的影响。\n\n5. **模型评估**：LangSmith 可能包括模型评估功能，这允许开发者测试模型的性能，并与其他模型或数据集进行比较，以确保模型能够满足生产环境中的需求。\n\n6. **监控和指标**：它可能提供了实时监控功能，帮助开发者收集和分析模型的性能指标，如响应时间、错误率等，从而持续监控模型的健康状况。\n\n通过这些功能，LangSmith 帮助开发者确保他们的 LLM 应用程序在推向生产之前经过了严格的测试，能够提供稳定和高质量的输出。这对于确保企业级应用的可靠性、一致性和用户满意度至关重要。')
1

2）在LangChain中，可以使用提示词模板，将原始用户输入转换为更好的 LLM 输入。提示词模板是LangChain的一大提效法宝。

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是世界级的技术文档作者。"),
    ("user", "{input}")
])
1
2
3
4
5
6

然后可以将提示词、LLM组合成一个简单的LLM链。Chain是LangChain的核心，最新的版本使用了Unix经典的管道方式来连接，相比之前简洁了很多。

chain = prompt | llm
1

调用LLM链。

chain.invoke({"input": "langsmith如何帮助测试?"})
1

结果：

AIMessage(content='Langsmith是一个机器学习模型，旨在帮助开发人员创建、理解和测试自然语言处理（NLP）模型。作为一个技术文档作者，我可以为你提供一个关于Langsmith如何帮助测试的概述：\n\n1. **自动化测试生成**：Langsmith可以自动生成用于测试NLP模型的示例输入和预期输出。这可以节省开发人员的时间，并确保测试覆盖了各种可能的输入情况。\n\n2. **测试用例优化**：Langsmith能够识别测试用例中的冗余或低效部分，并提出优化的建议。这有助于提高测试的效率和准确性。\n\n3. **模型行为分析**：通过分析NLP模型的行为，Langsmith可以帮助识别模型的潜在问题和缺陷。这有助于在模型部署之前发现并修复问题。\n\n4. **测试结果评估**：Langsmith可以评估测试结果，提供关于模型性能的详细反馈。这有助于开发人员了解模型的表现，并确定是否需要进一步的改进。\n\n5. **集成与协作**：Langsmith可以与其他开发工具和平台集成，以便在模型的开发和测试过程中提供无缝的支持。这有助于提高团队协作和开发效率。\n\n总的来说，Langsmith可以作为一个强大的工具，帮助开发人员创建高质量的NLP模型，并通过自动化和优化测试过程来提高测试效率和准确性。')
1

3）可以看到，输出结果是一个AIMessage消息对象。很多情况使用字符串要方便得多，可以添加一个字符串输出解析器来将聊天消息转换为字符串。输出解析器也是LangChain的一大提效法宝。

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是世界级的技术文档作者。"),
    ("user", "{input}")
])

output_parser = StrOutputParser()
chain = prompt | llm | output_parser
chain.invoke({"input": "langsmith如何帮助测试?"})
1
2
3
4
5
6
7
8
9
10

输出结果变成了字符串类型：

'Langsmith是一个机器学习模型，旨在帮助开发人员创建、理解和测试自然语言处理（NLP）模型。作为一个技术文档作者，我可以为你提供一个关于Langsmith如何帮助测试的概述：\n\n1. **自动化测试生成**：Langsmith可以自动生成用于测试NLP模型的示例输入和预期输出。这可以节省开发人员的时间，并确保测试覆盖了各种可能的输入情况。\n\n2. **测试用例优化**：Langsmith能够识别测试用例中的冗余或低效部分，并提出优化的建议。这有助于提高测试的效率和准确性。\n\n3. **模型行为分析**：通过分析NLP模型的行为，Langsmith可以帮助识别模型的潜在问题和缺陷。这有助于在模型部署之前发现并修复问题。\n\n4. **测试结果评估**：Langsmith可以评估测试结果，提供关于模型性能的详细反馈。这有助于开发人员了解模型的表现，并确定是否需要进一步的改进。\n\n5. **集成与协作**：Langsmith可以与其他开发工具和平台集成，如Jenkins、GitHub等，以便在模型开发和测试过程中实现自动化和协作。\n\n总的来说，Langsmith作为一个AI工具，旨在简化NLP模型的测试过程，提高测试的质量和效率，并帮助开发人员更快地构建和部署可靠的NLP解决方案。'
1

本示例通过提示词模板、LLM、输出解析器，以管道的方式组成一个链，可以快速的调用AI大模型，实现一个简单的AI应用程序。

2、检索链(Retrieval Chain)

之前简单LLM链示例里问的问题（“langsmith 如何帮助测试？”），完全是依赖大语言模型已有的知识来进行回答。当我们有更专业更准确的知识时，可以通过检索的方式从外部获取最相关的知识，然后作为背景知识传递给大语言模型，来获得更精准的结果。

1）首先从互联网获取数据

需要先安装BeautifulSoup：

pip install beautifulsoup4
1

导入并使用WebBaseLoader。官网原示例的"https://docs.smith.langchain.com/overview" 这个地址是错误的，改成：“https://docs.smith.langchain.com”

from langchain_community.document_loaders import WebBaseLoader
loader = WebBaseLoader("https://docs.smith.langchain.com")

docs = loader.load()
1
2
3
4

2）接下来需要使用嵌入模型进行向量化，再存储到向量数据库。

因为OpenAIEmbeddings嵌入模型需要和OpenAI ChatGPT配套使用。我们换成更通用的HuggingFaceEmbeddings。

from langchain_community.embeddings import HuggingFaceEmbeddings
embeddings = HuggingFaceEmbeddings()
1
2

现在可以使用此嵌入模型将文档提取到向量存储中了，为了简单起见，使用了本地向量数据库FAISS 。

首先安装FAISS所需的软件包：

pip install faiss-cpu
1

然后在向量数据库中建立索引：

from langchain_community.vectorstores import FAISS
from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter()
documents = text_splitter.split_documents(docs)
vector = FAISS.from_documents(documents, embeddings)
1
2
3
4
5
6

3）创建一个检索链。该链将接受传入的问题，查找相关文档，然后将这些文档与原始问题一起传递给LLM并要求其回答原始问题。

首先，建立一个文档链，该链接受问题和检索到的文档并生成答案。

from langchain.chains.combine_documents import create_stuff_documents_chain

prompt = ChatPromptTemplate.from_template("""仅根据所提供的上下文回答以下问题:

<context>
{context}
</context>

问题: {input}""")

document_chain = create_stuff_documents_chain(llm, prompt)
1
2
3
4
5
6
7
8
9
10
11

使用检索器动态选择最相关的文档，并将其传递给检索链。

from langchain.chains import create_retrieval_chain

retriever = vector.as_retriever()
retrieval_chain = create_retrieval_chain(retriever, document_chain)
1
2
3
4

调用检索链，得到答案。

response = retrieval_chain.invoke({"input": "langsmith如何帮助测试?"})
print(response["answer"])
1
2

结果：

根据上下文信息，LangSmith 是一个用于构建生产级大型语言模型（LLM）应用程序的平台。它可以让用户调试、测试、评估和监控基于任何LLM框架构建的链和智能代理。LangSmith 与 LangChain 无缝集成，LangChain 是一个用于与 LLMs 一起构建的开源框架。

具体来说，LangSmith 帮助测试的方式可能包括：

1. **调试**：提供调试工具和界面，帮助开发者发现并修复代码中的错误。
2. **监控**：在模型运行时监控其性能，确保其按照预期工作，并及时发现异常。
3. **评估**：提供评估功能，让开发者可以测试智能代理的响应和行为是否符合预期。
4. **集成**：与 LangChain 无缝集成，使得测试基于 LangChain 构建的 LLM 应用更为方便。

开发者可以使用 LangSmith 提供的文档和指南来学习如何使用这些测试功能。此外，LangSmith 可能还提供了一些示例和教程，帮助用户更好地理解和应用测试工具。

请注意，上述解释是基于提供的上下文信息推测的，具体langsmith如何帮助测试的详细信息需要查阅其官方文档或联系其技术支持以获得更准确的答案。
1
2
3
4
5
6
7
8
9
10
11
12

本示例首先获取外部互联网页面数据，经过嵌入模型对数据进行向量化，存储到向量数据库，当用户输入提示词时，到向量数据库中获取相关信息作为背景知识一起输入给LLM，最后LLM输出更准确的信息。通过这种检索增强生成的方式，可以方便的构建一些面向特定私有化知识领域的专用AI应用程序。后续会有一篇文章专门深入的介绍检索增强生成方面的应用。

3、智能体(Agent)

所谓智能体，实际上是指预先提供多种工具(tools)，智能体调用LLM根据我们的问题决定使用某种工具，进而调用工具获得需要的信息，再把需要的信息发送给LLM，获得最终结果。工具可以是LangChain内置的工具，也可以是自定义的工具，比如通过网络进行搜索获取信息的工具、数学计算的工具、我们自己定义的特定功能函数等。

在这个示例中，我们给智能体提供了两个工具：一个是刚才创建的有关 LangSmith的检索器，另一个是能够回答最新信息的搜索工具。

1）首先，为刚才创建的检索器设置一个工具，名字叫"langsmith_search"：

from langchain.tools.retriever import create_retriever_tool

retriever_tool = create_retriever_tool(
    retriever,
    "langsmith_search",
    "搜索有关LangSmith的信息。关于LangSmith的任何问题，您都可以使用这个工具",
)
1
2
3
4
5
6
7

2）对于另一个搜索工具，需要使用到Tavily，需要用到API 密钥，从https://app.tavily.com/home网站上可以免费申请。

然后设置为环境变量：

export TAVILY_API_KEY=...
1

创建搜索工具：

from langchain_community.tools.tavily_search import TavilySearchResults

search = TavilySearchResults()
1
2
3

3）创建智能体使用的工具列表：

tools = [retriever_tool, search]
1

4）接下来可以创建智能体来使用这些工具。

首先安装langchain hub，这个网站上提供了很多提示词模板，可以直接使用。

pip install langchainhub
1

获取提示词和创建智能体执行器。这里做了些优化，create_openai_functions_agent换成了create_openai_tools_agent，“hwchase17/openai-functions-agent"换成了"hwchase17/openai-tools-agent”。

from langchain import hub
from langchain.agents import create_openai_tools_agent
from langchain.agents import AgentExecutor

# Get the prompt to use - you can modify this!
prompt = hub.pull("hwchase17/openai-tools-agent")
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
1
2
3
4
5
6
7
8

4）现在可以调用智能体了。先向它询问有关 LangSmith 的问题：

agent_executor.invoke({"input": "langsmith如何帮助测试?"})
1

结果如下，可以看到LLM先根据问题确定要使用"langsmith_search"工具，然后LangChain调用"langsmith_search"工具从向量数据库中获得相关信息，最后返回信息给LLM得到最终结果。

> Entering new AgentExecutor chain...

Invoking: `langsmith_search` with `{'query': 'langsmith如何帮助测试'}`


LangSmith | 
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/weixin_40725706/article/detail/344116