在使用LangChain進行AI應用程序開發前，需要準備好相應的開發環境，包括Conda、Jupyter Notebook、使用的智譜AI GLM-4大模型。
使用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激活虛擬環境：
conda activate langchain為了能更方便編程和調試，及時執行代碼塊，我們使用Jupyter Notebook來作為Python集成開發工具。
在虛擬環境中安裝Jupyter Notebook：
conda install jupyter notebook啟動Jupyter notebook：
jupyter notebook在Jupyter Notebook中進行操作，安裝LangChain的最小集。
pip install --upgrade langchainLLM的選擇有多種方案。
方案一：遠程調用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由于最新的LangChain 0.1.7集成的ChatZhipuAI類和最新zhipuai SDK版本兼容性方面有問題，需要重新包裝一個類。代碼如下：
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)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](http://m.dlbhg.com/wiki/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)創建調用的對話大模型對象，api key需要修改成你自己的：
# 填寫您自己的APIKey
ZHIPUAI_API_KEY = "..."
llm = ChatZhipuAI(
temperature=0.1,
api_key=ZHIPUAI_API_KEY,
model_name="glm-4",
)環境準備好之后，就可以開始使用LangChain進行應用程序開發的入門，介紹了簡單LLM鏈、檢索鏈、智能體三個示例程序，通過一步一步操作，讓大家快速入門。
1）上面已經安裝并創建了智譜AI大模型，現在可以直接調用它。
llm.invoke("langsmith如何幫助測試?")結果：
AIMessage(content='LangSmith 是一個旨在幫助開發者將大語言模型（LLM）應用程序從原型階段推進到生產階段的全棧開發平臺。在測試方面，LangSmith 提供了一系列功能和工具來支持開發者對 LLM 應用進行全面的測試，確保其性能、可靠性和穩定性。以下是 LangSmith 在測試方面如何提供幫助的幾個方面：nn1. **調試工具**：LangSmith 為開發者提供了調試工具，可以幫助他們理解模型的輸出以及其在特定輸入下的行為。這包括錯誤追蹤和日志分析功能，使得開發者可以快速定位并解決模型在運行時出現的問題。nn2. **跟蹤功能**：它允許開發者跟蹤模型的使用情況，包括性能指標和錯誤率，從而幫助開發者監控模型在實際使用中的表現。nn3. **集成測試框架**：LangSmith 可能提供了集成測試框架，使得開發者可以編寫測試用例來驗證模型的預期行為。這包括單元測試和集成測試，以確保模型在各種條件下都能正確運行。nn4. **Prompt 優化**：在生成式語言模型中，Prompt（提示）的設計對輸出質量至關重要。LangSmith 可能提供了Prompt優化工具，幫助開發者測試和比較不同Prompt對模型輸出的影響。nn5. **模型評估**：LangSmith 可能包括模型評估功能，這允許開發者測試模型的性能，并與其他[模型或數據](http://m.dlbhg.com/blog/yq-deepspeed-chat-model-and-data)集進行比較，以確保模型能夠滿足生產環境中的需求。nn6. **監控和指標**：它可能提供了實時監控功能，幫助開發者收集和分析模型的性能指標，如響應時間、錯誤率等，從而持續監控模型的健康狀況。nn通過這些功能，LangSmith 幫助開發者確保他們的 LLM 應用程序在推向生產之前經過了嚴格的測試，能夠提供穩定和高質量的輸出。這對于確保企業級應用的可靠性、一致性和用戶滿意度至關重要。')2）在LangChain中，可以使用提示詞模板，將原始用戶輸入轉換為更好的 LLM 輸入。提示詞模板是LangChain的一大提效法寶。
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
prompt = ChatPromptTemplate.from_messages([
("system", "你是世界級的技術文檔作者。"),
("user", "{input}")
])然后可以將提示詞、LLM組合成一個簡單的LLM鏈。Chain是LangChain的核心，最新的版本使用了Unix經典的管道方式來連接，相比之前簡潔了很多。
chain = prompt | llm調用LLM鏈。
chain.invoke({"input": "langsmith如何幫助測試?"})結果：
AIMessage(content='Langsmith是一個機器學習模型，旨在幫助開發人員創建、理解和測試自然語言處理（NLP）模型。作為一個技術文檔作者，我可以為你提供一個關于Langsmith如何幫助測試的概述：nn1. **自動化測試生成**：Langsmith可以自動生成用于測試NLP模型的示例輸入和預期輸出。這可以節省開發人員的時間，并確保測試覆蓋了各種可能的輸入情況。nn2. **測試用例優化**：Langsmith能夠識別測試用例中的冗余或低效部分，并提出優化的建議。這有助于提高測試的效率和準確性。nn3. **模型行為分析**：通過分析NLP模型的行為，Langsmith可以幫助識別模型的潛在問題和缺陷。這有助于在模型部署之前發現并修復問題。nn4. **測試結果評估**：Langsmith可以評估測試結果，提供關于模型性能的詳細反饋。這有助于開發人員了解模型的表現，并確定是否需要進一步的改進。nn5. **集成與協作**：Langsmith可以與其他開發工具和平臺集成，以便在模型的開發和測試過程中提供無縫的支持。這有助于提高團隊協作和開發效率。nn總的來說，Langsmith可以作為一個強大的工具，幫助開發人員創建高質量的NLP模型，并通過自動化和優化測試過程來提高測試效率和準確性。')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如何幫助測試?"})輸出結果變成了字符串類型：
'Langsmith是一個機器學習模型，旨在幫助開發人員創建、理解和測試自然語言處理（NLP）模型。作為一個技術文檔作者，我可以為你提供一個關于Langsmith如何幫助測試的概述：nn1. **自動化測試生成**：Langsmith可以自動生成用于測試NLP模型的示例輸入和預期輸出。這可以節省開發人員的時間，并確保測試覆蓋了各種可能的輸入情況。nn2. **測試用例優化**：Langsmith能夠識別測試用例中的冗余或低效部分，并提出優化的建議。這有助于提高測試的效率和準確性。nn3. **模型行為分析**：通過分析NLP模型的行為，Langsmith可以幫助識別模型的潛在問題和缺陷。這有助于在模型部署之前發現并修復問題。nn4. **測試結果評估**：Langsmith可以評估測試結果，提供關于模型性能的詳細反饋。這有助于開發人員了解模型的表現，并確定是否需要進一步的改進。nn5. **集成與協作**：Langsmith可以與其他開發工具和平臺集成，如Jenkins、GitHub等，以便在模型開發和測試過程中實現自動化和協作。nn總的來說，Langsmith作為一個AI工具，旨在簡化NLP模型的測試過程，提高測試的質量和效率，并幫助開發人員更快地構建和部署可靠的NLP解決方案。'本示例通過提示詞模板、LLM、輸出解析器，以管道的方式組成一個鏈，可以快速的調用AI大模型，實現一個簡單的AI應用程序。
之前簡單LLM鏈示例里問的問題（“langsmith 如何幫助測試？”），完全是依賴大語言模型已有的知識來進行回答。當我們有更專業更準確的知識時，可以通過檢索的方式從外部獲取最相關的知識，然后作為背景知識傳遞給大語言模型，來獲得更精準的結果。
1）首先從互聯網獲取數據
需要先安裝BeautifulSoup：
pip install beautifulsoup4導入并使用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()2）接下來需要使用嵌入模型進行向量化，再存儲到向量數據庫。
因為OpenAIEmbeddings嵌入模型需要和OpenAI ChatGPT配套使用。我們換成更通用的HuggingFaceEmbeddings。
from langchain_community.embeddings import HuggingFaceEmbeddingsembeddings = HuggingFaceEmbeddings()
現在可以使用此嵌入模型將文檔提取到向量存儲中了，為了簡單起見，使用了本地向量數據庫FAISS 。
首先安裝FAISS所需的軟件包：
pip install faiss-cpu然后在向量數據庫中建立索引：
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)3）創建一個檢索鏈。該鏈將接受傳入的問題，查找相關文檔，然后將這些文檔與原始問題一起傳遞給LLM并要求其回答原始問題。
首先，建立一個文檔鏈，該鏈接受問題和檢索到的文檔并生成答案。
from langchain.chains.combine_documents import create_stuff_documents_chain
prompt = ChatPromptTemplate.from_template("""僅根據所提供的上下文回答以下問題:
{context}
問題: {input}""")
document_chain = create_stuff_documents_chain(llm, prompt)使用檢索器動態選擇最相關的文檔，并將其傳遞給檢索鏈。
from langchain.chains import create_retrieval_chain
retriever = vector.as_retriever()
retrieval_chain = create_retrieval_chain(retriever, document_chain)調用檢索鏈，得到答案。
response = retrieval_chain.invoke({"input": "langsmith如何幫助測試?"})
print(response["answer"])結果：
根據上下文信息，LangSmith 是一個用于構建生產級大型語言模型（LLM）應用程序的平臺。它可以讓用戶調試、測試、評估和監控基于任何[LLM框架構](http://m.dlbhg.com/blog/summary-of-large-model-techniques)建的鏈和智能代理。LangSmith 與 [LangChain 無縫集成](http://m.dlbhg.com/blog/wx-openai-assistants-api-minimal-intro-with-langchain-integration)，LangChain 是一個用于與 LLMs 一起構建的開源框架。
具體來說，LangSmith 幫助測試的方式可能包括：
1. **調試**：提供調試工具和界面，幫助開發者發現并修復代碼中的錯誤。
2. **監控**：在模型運行時監控其性能，確保其按照預期工作，并及時發現異常。
3. **評估**：提供評估功能，讓開發者可以測試智能代理的響應和行為是否符合預期。
4. **集成**：與 LangChain 無縫集成，使得測試基于 LangChain 構建的 LLM 應用更為方便。
開發者可以使用 LangSmith 提供的文檔和指南來學習如何使用這些測試功能。此外，LangSmith 可能還提供了一些示例和教程，幫助用戶更好地理解和應用測試工具。
請注意，上述解釋是基于提供的上下文信息推測的，具體langsmith如何幫助測試的詳細信息需要查閱其官方文檔或聯系其技術支持以獲得更準確的答案。本示例首先獲取外部互聯網頁面數據，經過嵌入模型對數據進行向量化，存儲到向量數據庫，當用戶輸入提示詞時，到向量數據庫中獲取相關信息作為背景知識一起輸入給LLM，最后LLM輸出更準確的信息。通過這種檢索增強生成的方式，可以方便的構建一些面向特定私有化知識領域的專用AI應用程序。后續會有一篇文章專門深入的介紹檢索增強生成方面的應用。
所謂智能體，實際上是指預先提供多種工具(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的任何問題，您都可以使用這個工具",
)2）對于另一個搜索工具，需要使用到Tavily，需要用到API 密鑰，從https://app.tavily.com/home網站上可以免費申請。
然后設置為環境變量：
export TAVILY_API_KEY=...創建搜索工具：
from langchain_community.tools.tavily_search import TavilySearchResults
search = TavilySearchResults()3）創建智能體使用的工具列表：
–
tools = [retriever_tool, search]4）接下來可以創建智能體來使用這些工具。
首先安裝langchain hub，這個網站上提供了很多提示詞模板，可以直接使用。
pip install langchainhub獲取提示詞和創建智能體執行器。這里做了些優化，create_openaifunctionsagent換成了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)4）現在可以調用智能體了。先向它詢問有關 LangSmith 的問題：
agent_executor.invoke({"input": "langsmith如何幫助測試?"})結果如下，可以看到LLM先根據問題確定要使用"langsmith_search"工具，然后LangChain調用"langsmith_search"工具從向量數據庫中獲得相關信息，最后返回信息給LLM得到最終結果。
> Entering new AgentExecutor chain...
Invoking: langsmith_search with {'query': 'langsmith如何幫助測試'}
LangSmith | ?? LangSmith
Skip to main content?? LangSmith DocsLangChain [API key](http://m.dlbhg.com/blog/api-key-what-is-an-api-key/), and configure your environment.Pricing: Learn about the pricing model for LangSmith.Self-Hosting: Learn about self-hosting options for LangSmith.Tracing: Learn about the tracing capabilities of LangSmith.Evaluation: Learn about the evaluation capabilities of LangSmith.Prompt Hub Learn about the Prompt Hub, a prompt management tool built into LangSmith.Additional ResourcesLangSmith Cookbook: A collection of tutorials and end-to-end walkthroughs using LangSmith.LangChain Python: Docs for the Python LangChain library.LangChain [Python API](http://m.dlbhg.com/blog/sa-creating-python-apis) Reference: documentation to review the core APIs of LangChain.LangChain JS: Docs for the TypeScript LangChain libraryDiscord: Join us on our Discord to discuss all things LangChain!Contact SalesIf you're interested in enterprise security and admin features, special deployment options, or access for large teams, reach out to speak with sales.NextUser GuideIntroductionQuick StartNext StepsAdditional ResourcesCommunityDiscordTwitterGitHubDocs CodeLangSmith SDKPythonJS/TSMoreHomepageBlogCopyright ? 2024 LangChain, Inc.LangSmith 是一個用于構建生產級 LLM 應用程序的平臺。它可以讓您調試、測試、評估和監控基于任何 LLM 框架構建的鏈和智能代理，并且可以無縫地與 LangChain 集成，LangChain 是構建 LLM 的首選開源框架。LangSmith 是由 LangChain 開發的，LangChain 是 LLM 應用程序生命周期各個階段的 workflow 支持，例如調試、測試、評估和監控。此外，LangSmith 還提供了 Prompt Hub，這是一個內置的提示管理工具，可以讓您輕松地管理和維護提示。
> Finished chain.
[31]:
{'input': 'langsmith如何幫助測試?',
'output': 'LangSmith 是一個用于構建生產級 LLM 應用程序的平臺。它可以讓您調試、測試、評估和監控基于任何 LLM 框架構建的鏈和智能代理，并且可以無縫地與 LangChain 集成，LangChain 是構建 LLM 的首選開源框架。LangSmith 是由 LangChain 開發的，LangChain 是 LLM 應用程序生命周期各個階段的 workflow 支持，例如調試、測試、評估和監控。此外，LangSmith 還提供了 Prompt Hub，這是一個內置的提示管理工具，可以讓您輕松地管理和維護提示。'}5）再詢問下天氣情況：
agent_executor.invoke({"input": "舊金山的天氣怎么樣?"})結果如下，可以看到LLM先根據問題確定要使用"tavily_search_results_json"工具，然后LangChain調用"tavily_search_results_json"工具從互聯網上獲得相關信息，最后返回信息給LLM得到最終結果。
> Entering new AgentExecutor chain...
Invoking: tavily_search_results_json with {'query': '舊金山的天氣'}
[{'url': 'https://tianqi.2345.com/america_united-states/san-francisco-ca/', 'content': '2345天氣預報提供舊金山天氣預報，未來舊金山15天天氣，通過2345天氣預報詳細了解舊金山天氣預報以及舊金山周邊各地區未來15天、30天天氣情況，溫度，空氣質量，降水，風力，氣壓，紫外線強度等! ... 您使用的瀏覽器版本過低! ...'}, {'url': 'https://zh.weather-forecast.com/locations/San-Francisco/forecasts/latest', 'content': '鄰近舊金山的實時氣象站n*注意：并非所有舊金山附近的氣象站會同時更新，我們只顯示最近氣象站的報告 被視為當前的。還包括在可接受的時間范圍內舊金山 附近的任何船舶（SYNOP）提交的天氣報告。Read Moren舊金山 Location mapn加利福尼亞州天氣地圖n加利福尼亞州天氣地圖n舊金山的天氣照片n上傳一幅 舊金山的天氣照片 | 上傳另一個城市n目的地舊金山附近城市的天氣預報：n? 2024 Meteo365.com Very mild (max 14°C on Fri morning, min 8°C on Wed night). Very mild (max 14°C on Fri morning, min 8°C on Wed night). Very mild (max 11°C on Sat afternoon, min 7°C on Thu night). Very mild (max 15°C on Sat afternoon, min 10°C on Mon night).'}, {'url': 'http://www.weather.com.cn/weather/401640100.shtml', 'content': '【舊金山天氣】舊金山天氣預報,天氣預報一周,天氣預報15天查詢 預報 美國>舊金山（Sanfrancisco） 18:00更新 數據來源 中央氣象臺 今天 周末 7天 8-15天 10日（今天） 晴轉陰 16 / 8℃ 4-5級轉3-4級 11日（明天） 小雨轉陰 14 / 7℃ <3級 12日（后天） 陰 13 / 8℃ <3級 13日（周二） 陰 13 / 7℃ <3級 14日（周三） 小雨轉陰 13 / 8℃ <3級 15日（周四） 多云轉陰 14 / 8℃ Finished chain.
[25]:
{'input': '舊金山的天氣怎么樣?',
'output': '舊金山現在的天氣我無法獲取，但是根據tavily_search_results_json的結果，最近舊金山的天氣情況如下：今天是晴天，最高溫度為16°C，最低溫度為8°C。明天有小雨，最高溫度為14°C，最低溫度為7°C。后天天氣陰，最高溫度為13°C，最低溫度為8°C。'}本示例通過給智能體提供多種工具(有關LangSmith的檢索器、回答最新信息的搜索工具)，智能體調用LLM根據我們的問題決定使用某種工具，進而調用工具獲得需要的信息，再把需要的信息發送給LLM，獲得最終結果。通過這種工具增強的方式，可以構建集成特定應用的AI應用程序，比如在應用程序中集成在線購物、自定義的數據計算等實用功能。后續會有一篇文章專門深入的介紹智能體工具方面的應用。
本文是最新版LangChain使用智譜AI的GLM-4大模型開發AI應用程序系列的第一篇文章-快速入門篇。
首先對LangChain做了介紹，LangChain是哈里森·蔡斯 (Harrison Chase) 于2022年10月創建的，對于AI應用程序開發來說，它的地位就相當于Java界的Spring。LangChain的整體架構分為六層，它的核心概念就是鏈，鏈接外部一切能鏈接的東西賦予LLM力量。
然后從零基礎開始介紹了LangChain環境的安裝和配置，包括Conda、Jupyter Notebook、LangChain，以及智譜AI GLM-4在LangChain的最新調用API。
最后再通過LLM鏈、檢索鏈、智能體三個經典的示例，帶大家一步一步的快速上手了提示詞模板、輸出解析器、管道、LLM鏈、檢索鏈、智能體等功能的使用和開發。
希望能給大家起到一定的參考作用，同時也歡迎共同探討。后續文章會繼續深入分析檢索鏈、智能體等的使用，敬請期待。
文章轉自微信公眾號@頂尖程序員