赞
踩
小李哥将继续每天介绍一个基于亚马逊云科技AWS云计算平台的全球前沿AI技术解决方案,帮助大家快速了解国际上最热门的云计算平台亚马逊云科技AWS AI最佳实践,并应用到自己的日常工作里。
本次介绍用当下热门的LangChain框架的基本知识和学习亚马逊云科技热门AI模型管理服务Amazon SageMaker实操,手把手开发生成式AI应用(对话机器人和代码生成器)。本文将通过利用LangChain的提示词工程PromptTemplate模块、问答链load_qa_chain模块、RetrievalQA问答检索模块、Document文档模块、Embedding向量化模块、VectorstoreIndexCreator 索引模块、Conversation对话模块,编写一步一步的细节代码和展示LangChain应用实际样例,0基础学会AI核心技能。网页聊天AI机器人界面如下。
本方案使用 GPT-J 6B Embedding FP16作为向量模型,使用Falcon 7B Instruct BF16作为大语言模型。本架构设计全部采用了云原生Serverless架构,提供可扩展和安全的AI应用解决方案。本方案架构图如下:
LangChain 是一个专为开发者设计的开源库,旨在简化和增强语言模型的开发和部署。随着生成式AI技术的发展,如何有效地利用这些强大的语言模型成为了开发者关注的焦点。LangChain 提供了一套工具集,帮助开发者轻松地集成和管理语言模型,支持从模型选择到部署的一系列操作,使开发过程更加高效和简洁。
LangChain 可以用来构建智能客服和对话系统,通过接入强大的语言模型,能够更好地理解和回应用户的自然语言输入。这种应用场景在电商、金融等领域非常受欢迎,能够有效提升用户体验,减少人工客服的工作量。
对于需要大量内容、文案生成的行业,如媒体、市场营销等,LangChain 可以帮助开发自动化写作工具。通过定制化的语言模型,系统可以根据输入的关键词或主题自动生成高质量的文章、新闻稿或广告文案,大大提高内容生产效率。
在数据密集型行业,如金融和市场分析,LangChain 可以用来开发数据分析和报告生成工具。这些工具能够从海量数据中提取关键信息,并生成易于理解的分析报告,帮助决策者快速获取所需的洞见。
LangChain 还可以用于构建自动翻译和内容本地化系统,帮助出海企业跨越语言障碍,进入全球市场。通过结合先进的语言模型和翻译技术,可以实现高质量的文本翻译,同时保留原文的语义和风格。
1. 首先我们登录亚马逊云科技控制台,进入SageMaker服务中的Studio
2. 点击“Open”,打开SageMaker中的Jupyter Notebook
3. 打开“SageMaker JumpStart”下拉栏,选择“GPT-J 6B Embedding FP16”向量模型
4. 为模型配置底层服务器资源类型,我们选择“ml.g5.2xlarge”,复制模型节点名称并点击“Deploy”部署
5.按相同方式,我们部署另外的模型“Falcon 7B Embedding FP16”
6. 我们可以在SageMaker服务下的Endpoints界面查看部署状态,“Inservice”则表述部署完成
7. 接下来我们在SageMaker Studio中新建一个Jupyter Notebook,利用LangChain优化大模型的回复
8. 首先我们在Cell中复制以下代码,安装LangChain和相关依赖
- import warnings
- warnings.filterwarnings('ignore')
-
- !pip install --upgrade pip
- !pip install --upgrade sagemaker --quiet
- !pip install ipywidgets==7.0.0 --quiet
- !pip install langchain==0.1.11 --quiet
- !pip install faiss-cpu --quiet
9. 接下来我们导入依赖并初始化模型训练客户端
- import time
- import boto3, json
- from typing import Any, Dict, List, Optional
-
- session = boto3.Session()
- sagemaker_runtime_client = session.client("sagemaker-runtime")
10. 接下来我们开始创建一个AI问答机器人。首先定义文档内容、LangChain提示词模板PromptTemplate、和LangChain输入、输出格式处理函数ContentHandler。
- from typing import Dict
- from langchain.chains.question_answering import load_qa_chain
- from langchain_core.prompts import PromptTemplate
- from langchain_community.llms import SagemakerEndpoint
- from langchain_community.llms.sagemaker_endpoint import LLMContentHandler
- from langchain.docstore.document import Document
-
-
- example_doc_1 = """
- Amazon Bedrock is a fully managed service that offers a choice of high-performing foundation models (FMs) from
- leading AI companies like AI21 Labs, Anthropic, Cohere, Meta, Mistral AI, Stability AI, and Amazon via a single
- API, along with a broad set of capabilities you need to build generative AI applications with security, privacy,
- and responsible AI. Using Amazon Bedrock, you can easily experiment with and evaluate top FMs for your use case,
- privately customize them with your data using techniques such as fine-tuning and Retrieval Augmented Generation
- (RAG), and build agents that execute tasks using your enterprise systems and data sources. Since Amazon Bedrock
- is serverless, you don't have to manage any infrastructure, and you can securely integrate and deploy generative
- AI capabilities into your applications using the AWS services you are already familiar with.
- """
-
- docs = [
- Document(
- page_content=example_doc_1,
- )
- ]
-
- prompt_template = """Use the following pieces of context to answer the question at the end.
- {context}
- Question: {question}
- Answer:"""
- PROMPT = PromptTemplate(
- template=prompt_template, input_variables=["context", "question"]
- )
-
- class ContentHandler(LLMContentHandler):
- content_type = "application/json"
- accepts = "application/json"
-
- def transform_input(self, prompt: str, model_kwargs: Dict) -> bytes:
- input_str = json.dumps({"inputs": prompt, "parameters": model_kwargs})
- # print(input_str)
- return input_str.encode("utf-8")
-
- def transform_output(self, output: bytes) -> str:
- response_json = json.loads(output.read().decode("utf-8"))
- # print(response_json)
- return response_json[0]["generated_text"]
-
-
- content_handler = ContentHandler()
11. 定义模型参数parameters,构建调用亚马逊云科技SageMaker的问答链load_qa_chain,并利用样例问题对问LangChain问答链测试
- embedding_endpoint_name = "jumpstart-dft-<REPLACE WITH EMBEDDING ENDPOINT NAME>"
- instruct_endpoint_name = "jumpstart-dft-<REPLACE WITH INSTRUCT ENDPOINT NAME>"
-
- parameters = {
- "max_length": 200,
- "num_return_sequences": 1,
- "top_k": 250,
- "top_p": 0.95,
- "do_sample": False,
- "temperature": 1,
- }
- chain = load_qa_chain(
- llm = SagemakerEndpoint(
- endpoint_name=instruct_endpoint_name,
- region_name='us-east-1',
- model_kwargs=parameters,
- content_handler=content_handler,
- ),
- prompt=PROMPT,
- )
-
- question = "What is Amazon Bedrock?"
- chain.invoke({"input_documents": docs, "question": question}, return_only_outputs=True)
12. 下面我们将外挂知识库,在LangChain中结合RAG增强我们AI问答机器人的回复效果,得到更准确的答案。首先我们利用LangChain的SagemakerEndpointEmbeddings类定义文本向量化的规则,再用EmbeddingsContentHandler定义向量化的输入和输出格式,模型使用了SageMaker上向量模型GPT-J 6B Embedding FP16。
- # Wrap the SageMaker endpoint for the embedding model into langchain_community.embeddings.SagemakerEndpointEmbeddings.
- # This code defines a custom subclass of SagemakerEndpointEmbeddings to handle document embeddings by using a SageMaker endpoint.
- # The code also defines a custom content handler for input and output formatting.
-
- from langchain_community.embeddings import SagemakerEndpointEmbeddings
- from langchain_community.embeddings.sagemaker_endpoint import EmbeddingsContentHandler
-
-
- class SagemakerEndpointEmbeddingsJumpStart(SagemakerEndpointEmbeddings):
- def embed_documents(self, texts: List[str], chunk_size: int = 5) -> List[List[float]]:
- """Compute doc embeddings using a SageMaker Inference Endpoint.
- Args:
- texts: The list of texts to embed.
- chunk_size: The chunk size defines how many input texts will
- be grouped together as request. If None, will use the
- chunk size specified by the class.
- Returns:
- List of embeddings, one for each text.
- """
- results = []
- _chunk_size = len(texts) if chunk_size > len(texts) else chunk_size
-
- for i in range(0, len(texts), _chunk_size):
- response = self._embedding_func(texts[i : i + _chunk_size])
- print
- results.extend(response)
- return results
-
-
- class ContentHandler(EmbeddingsContentHandler):
- content_type = "application/json"
- accepts = "application/json"
-
- def transform_input(self, prompt: str, model_kwargs={}) -> bytes:
- input_str = json.dumps({"text_inputs": prompt, **model_kwargs})
- return input_str.encode("utf-8")
-
- def transform_output(self, output: bytes) -> str:
- response_json = json.loads(output.read().decode("utf-8"))
- embeddings = response_json["embedding"]
- return embeddings
-
-
- content_handler = ContentHandler()
-
- embeddings = SagemakerEndpointEmbeddingsJumpStart(
- endpoint_name=embedding_endpoint_name,
- region_name='us-east-1',
- content_handler=content_handler,
- )
13. 接下来我们预处理数据,将开源的FAQ数据集转换成DataFrame用于接下来的测试。
- import glob
- import os
- import pandas as pd
-
- all_files = glob.glob(os.path.join("rag_data/", "*.csv"))
-
- df_knowledge = pd.concat(
- (pd.read_csv(f, header=None, names=["Question", "Answer"]) for f in all_files),
- axis=0,
- ignore_index=True,
- )
- df_knowledge.drop(["Question"], axis=1, inplace=True)
- df_knowledge.head(5)
14. 下面我们导入必要的依赖,并用LangChain去读取FAQ数据集中的答案部分
- # Set up the necessary imports for building the Q&A app.
- from langchain.chains import RetrievalQA
- from langchain.indexes.vectorstore import VectorstoreIndexCreator
- from langchain_community.document_loaders.text import TextLoader
- from langchain_community.vectorstores.faiss import FAISS
- from langchain_text_splitters import CharacterTextSplitter
- from langchain_core.prompts import PromptTemplate
- from langchain.chains.question_answering import load_qa_chain
- from langchain_community.document_loaders import DataFrameLoader
-
- # Load the processed data into LangChain.
- loader = DataFrameLoader(df_knowledge,page_content_column="Answer")
- documents = loader.load()
15.接下来我们利用LangChain调用向量模型将FAQ数据集的答案部分向量化并保存,使用大语言模型对我们的实例问题进行相关性搜索得到搜索结果,这里用到了FAISS库加速RAG搜索效率。
- question = "What is Amazon SageMaker?"
- index_creator = VectorstoreIndexCreator(
- vectorstore_cls=FAISS,
- embedding=embeddings,
- text_splitter=CharacterTextSplitter(chunk_size=300, chunk_overlap=0),
- )
- index = index_creator.from_loaders([loader])
- index.query(question=question, llm=sm_llm)
16. 接下来我们介绍两个应用该方案的现实场景。
16.1 使用LangChain和LLM进行代码生成,样例代码如下,生成的代码需求是: "takes a list of numbers and returns the sum of all even numbers",“根据一个数字列表并返回所有偶数之和。”
- from langchain import PromptTemplate, LLMChain
-
- # Define SageMaker LLM
- llm = sm_llm
-
- # Define the prompt template
- prompt_template = """
- Write a Python function that {task_description}.
- """
-
- # Create the prompt template instance
- prompt = PromptTemplate(
- input_variables=["task_description"],
- template=prompt_template,
- )
-
- # Create the LLMChain instance with the prompt and LLM
- llm_chain = LLMChain(prompt=prompt, llm=llm)
-
- # Generate code based on the task description
- task_description = "takes a list of numbers and returns the sum of all even numbers"
- output = llm_chain.invoke(input=task_description)
- print(output["text"])
我们可以看到我们利用LangChain与大模型交互,根据我们的需求生成了代码。
16.2 利用LangChain ConversationChain与大模型交互构建AI对话机器人。我们这里使用 conversation.predict
方法处理用户输入的内容,利用LangChain与大模型交互生成对话回复。
- from langchain.memory import ConversationBufferMemory
- from langchain import ConversationChain
-
- parameters = {
- "max_length": 50,
- "num_return_sequences": 1,
- "top_k": 10,
- "top_p": 0.01,
- "do_sample": False,
- "temperature": 0, }
-
- sm_llm = SagemakerEndpoint(
- endpoint_name=instruct_endpoint_name,
- region_name='us-east-1',
- model_kwargs=parameters,
- content_handler=content_handler,
- )
-
- # Initialize the LLM (in this case, OpenAI's GPT-3)
- llm = sm_llm
-
- # Initialize the memory
- memory = ConversationBufferMemory()
-
- # Create the conversation chain with memory
- conversation = ConversationChain(
- llm=llm,
- memory=memory,
- verbose=False
- )
-
- def chat_with_ai():
- print("Hi! I'm an AI assistant. How can I help you today?")
- while True:
- human_input = input("Human: ")
- if human_input.lower() == 'exit':
- print("Assistant: Goodbye!")
- break # Exit the loop if user types 'exit'
-
- # Process input through the conversation chain
- response = conversation.predict(input=human_input, stop="\n\n")
-
- # Print the AI's response
- print(f"Assistant: {response}")
-
- # Call the function to start chatting
- chat_with_ai()
以上就是在亚马逊云科技上使用LangChain和大语言AI模型搭建云原生AI对话机器人,和生成代码的步骤。欢迎大家关注小李哥,未来获取更多国际前沿的生成式AI开发方案!
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。