【Langchain+Streamlit】旅游聊天機(jī)器人_langchain streamlit-CSDN博客

視頻講解地址：【Langchain Agent】帶推銷功能的旅游聊天機(jī)器人_嗶哩嗶哩_bilibili

體驗(yàn)地址：? ?http://101.33.225.241:8503/

github地址：GitHub - jerry1900/langchain_chatbot: langchain+streamlit打造的一個(gè)有memory的旅游聊天機(jī)器人，可以和你聊旅游相關(guān)的事兒

?

????????之前，我們介紹如何打算一款簡單的旅游聊天機(jī)器人，而且之前我們介紹了SalesGPT，我們看能不能把這兩個(gè)東西結(jié)合起來，讓我們的旅游聊天機(jī)器人具有推銷產(chǎn)品的功能。我們先來看看效果：

? ? ? ? 首先，你可以和機(jī)器人閑聊關(guān)于旅游的事兒（如果你問的問題和旅游無關(guān)的話，會(huì)提示你只回答旅游問題）?；其次，當(dāng)你連續(xù)詢問有關(guān)同一個(gè)地點(diǎn)時(shí)（比如北京），機(jī)器人會(huì)檢查自己的本地知識庫，看看產(chǎn)品庫里有沒有相關(guān)的旅游產(chǎn)品，如果有的話就推薦給客戶，如果沒有就輸出一個(gè)空的字符串，用戶是沒有感知的，我們來看一下是如何實(shí)現(xiàn)的。

1. 項(xiàng)目結(jié)構(gòu)

? ? ? ? 我們是在原來項(xiàng)目基礎(chǔ)上逐步疊加的，主要增加了一個(gè)agent.py、my_tools.py、stages.py等文件。我們這次的項(xiàng)目是使用poetry來管理和運(yùn)行的：

????????項(xiàng)目結(jié)構(gòu)如圖：

? ? ? ? 我們新加了一個(gè)產(chǎn)品文件用于存儲(chǔ)旅游產(chǎn)品，下面是三個(gè)產(chǎn)品中的一個(gè)：

產(chǎn)品名稱：北京一日游
產(chǎn)品價(jià)格：299元
產(chǎn)品內(nèi)容：{
北京作為中國的首都和歷史文化名城，有許多令人著迷的景點(diǎn)和活動(dòng)。以下是一個(gè)充滿活力和文化的北京一日游的建議：
早上：天安門廣場： 開始您的一日游，您可以前往天安門廣場，這是世界上最大的城市廣場之一，也是中國的政治中心。您可以欣賞到天安門城樓，參觀升國旗儀式（早上升旗時(shí)間）。
故宮博物院： 天安門廣場北側(cè)就是故宮，這是中國最大、最完整的古代皇家宮殿建筑群。在這里，您可以領(lǐng)略中國古代皇家建筑的壯麗和深厚的歷史文化。
中午：午餐： 您可以選擇在附近的餐館品嘗地道的北京菜，比如炸醬面、北京烤鴨等。
下午：頤和園： 中午過后，您可以前往頤和園，這是中國最大的皇家園林之一，也是清代的皇家園林。園內(nèi)有美麗的湖泊、精致的建筑和獨(dú)特的風(fēng)景。
什剎海： 在下午的最后時(shí)段，您可以前往什剎海地區(qū)，這里是一個(gè)古老而又時(shí)尚的區(qū)域，有著許多酒吧、咖啡館和特色小店。您可以漫步在湖邊，欣賞夕陽下的美景，體驗(yàn)北京的悠閑氛圍。
晚上：王府井步行街： 晚上，您可以前往王府井步行街，這是北京最繁華的購物街之一，有著各種商店、餐館和娛樂場所。您可以嘗試美食、購物或者觀賞街頭表演。
京劇表演： 如果時(shí)間允許，您還可以觀看一場京劇表演，京劇是中國傳統(tǒng)戲曲的代表之一，有著悠久的歷史和獨(dú)特的藝術(shù)魅力。
}

? 2. chat.py的改動(dòng)，新增了歡迎詞，添加了Agent構(gòu)造的方法

? ? ? ? 這里構(gòu)造一個(gè)專門負(fù)責(zé)提示詞的Agent（其實(shí)就是一個(gè)LLMChain），并構(gòu)造一個(gè)負(fù)責(zé)會(huì)話和判斷功能的ConversationAgent，讓這個(gè)agent初始化并構(gòu)造一個(gè)負(fù)責(zé)判斷階段的內(nèi)部agent，我們把他們都要放到session里：

#需要國內(nèi)openai開發(fā)賬號的請聯(lián)系微信 15652965525if "welcome_word" not in st.session_state:st.session_state.welcome_word = welcome_agent()st.session_state.messages.append({'role': 'assistant', 'content': st.session_state.welcome_word['text']})st.session_state.agent = ConversationAgent()st.session_state.agent.seed_agent()st.session_state.agent.generate_stage_analyzer(verbose=True)

? ? ? ? 在用戶輸入后的每一步，先進(jìn)行一下階段判斷，然后調(diào)用agent的human_step方法，再調(diào)用agent的step()方法，完成一輪對話：

??

st.session_state.agent.determine_conversation_stage(prompt)st.session_state.agent.human_step(prompt)response = st.session_state.agent.step()

???

3. welcome_agent

? ? ? ? 這個(gè)比較簡單，就是一個(gè)咱們學(xué)習(xí)過無數(shù)遍的一個(gè)簡單的chain：

def welcome_agent():llm = OpenAI(temperature=0.6,# openai_api_key=os.getenv("OPENAI_API_KEY"),openai_api_key=st.secrets['api']['key'],# base_url=os.getenv("OPENAI_BASE_URL")base_url=st.secrets['api']['base_url'])prompt = PromptTemplate.from_template(WELCOME_TEMPLATE)chain = LLMChain(llm=llm,prompt=prompt,verbose=True,)response = chain.invoke("簡短的歡迎詞")return response

? ? ? ? 這里我們希望每次調(diào)用它的時(shí)候，可以得到一些不一樣的、有創(chuàng)意的歡迎詞，所以我們的temperature調(diào)的比較高，這樣它可能生成一些有創(chuàng)意的歡迎詞。

4. ConversationAgent類

? ? ? ? 這個(gè)類是我們的核心類，里面有很多屬性和方法，我們用python的dir()方法來看一下它里面的結(jié)構(gòu)：

from agent import ConversationAgentagent = ConversationAgent()
print(dir(ConversationAgent))

? ? ? ? 里面以_開頭的是Object基本類自帶的屬性和方法，其他的是我們構(gòu)造的屬性和方法：

['__annotations__', '__class__', '__delattr__', '__dict__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__le__', '__lt__', '__module__', '__ne__', '__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', '__weakref__', '_respond_without_tools', 'conversation_agent_with_tool', 'conversation_agent_without_tool', 'conversation_history', 'conversation_stage_id', 'current_conversation_stage', 'determine_conversation_stage', 'fake_step', 'generate_stage_analyzer', 'get_tools', 'human_step', 'llm', 'recommend_product', 'retrieve_conversation_stage', 'seed_agent', 'show_chat_history', 'stage_analyzer_chain', 'step']

? ? ? ? 我們先來看類的屬性和一些簡單的方法，注意我們這里構(gòu)造了一個(gè)llm，之后下面有很多方法會(huì)用到這個(gè)llm：

class ConversationAgent():stage_analyzer_chain: StageAnalyzerChain = Field(...)conversation_agent_without_tool = Field()conversation_agent_with_tool = Field()conversation_history = []conversation_stage_id: str = "1"current_conversation_stage: str = CONVERSATION_STAGES.get("1")llm = OpenAI(temperature=0,openai_api_key=st.secrets['api']['key'],base_url=st.secrets['api']['base_url'])def seed_agent(self):self.conversation_history.clear()print("——Seed Successful——")def show_chat_history(self):return self.conversation_historydef retrieve_conversation_stage(self, key):return CONVERSATION_STAGES.get(key)

? ? ? ? 我們繼續(xù)來看：

    def fake_step(self):input_text = self.conversation_history[-1]ai_message = self._respond_with_tools(str(input_text), verbose=True)print(ai_message,type(ai_message['output']))def step(self):input_text = self.conversation_history[-1]print(str(input_text)+'input_text****')if int(self.conversation_stage_id) == 1:ai_message = self._respond_without_tools(str(input_text),verbose=True)else:chat_message = self._respond_without_tools(str(input_text), verbose=True)recommend_message = self.recommend_product()print(recommend_message,len(recommend_message))if len(recommend_message)<=5:ai_message = chat_messageelse:ai_message = chat_message + '\n\n' + recommend_message# output_dic = self._respond_with_tools(str(input_text),verbose=True)# ai_message = str(output_dic['output'])print(ai_message,type(ai_message))ai_message = "AI:"+str(ai_message)self.conversation_history.append(ai_message)# print(f"——系統(tǒng)返回消息'{ai_message}'，并添加到history里——")return ai_message.lstrip('AI:')

? ? ? ? fake_step是一個(gè)模擬輸出的方法，不用管，測試的時(shí)候用；step方法是接收用戶的輸入，從聊天記錄里取出來（input_text = self.conversation_history[-1]） ，然后再根據(jù)不同的對話階段進(jìn)行不同的邏輯，如果是第二個(gè)階段推銷階段，那么就調(diào)用recommend_product方法去生成一個(gè)推銷產(chǎn)品的信息，并把兩條信息拼接起來。

????????

    def human_step(self,input_text):human_message = input_texthuman_message = "用戶: " + human_messageself.conversation_history.append(human_message)# print(f"——用戶輸入消息'{human_message}'，并添加到history里——")return human_message

? ? ? ? human_step方法比較簡單，就是把用戶的輸入掛到conversation_history聊天記錄里。然后是構(gòu)造階段判斷的agent和階段判斷的方法，這些都是模仿SalesGPT里的，做了一些調(diào)整：

    def generate_stage_analyzer(self,verbose: bool = False):self.stage_analyzer_chain = StageAnalyzerChain.from_llm(llm=self.llm,verbose=verbose)print("成功構(gòu)造一個(gè)StageAnalyzerChain")def determine_conversation_stage(self,question):self.question = questionprint('-----進(jìn)入階段判斷方法-----')self.conversation_stage_id = self.stage_analyzer_chain.run(conversation_history=self.conversation_history,question=self.question)print(f"Conversation Stage ID: {self.conversation_stage_id}")print(type(self.conversation_stage_id))self.current_conversation_stage = self.retrieve_conversation_stage(self.conversation_stage_id)print(f"Conversation Stage: {self.current_conversation_stage}")

? ? ? ? ?然后是_respond_without_tools這么一個(gè)內(nèi)部的方法，它在step里被調(diào)用：

    def _respond_without_tools(self,input_text,verbose: bool = False):self.conversation_agent_without_tool = ConversationChain_Without_Tool.from_llm(llm=self.llm,verbose=verbose)response = self.conversation_agent_without_tool.run(question = input_text,conversation_history=self.conversation_history,)return response

? ? ? ? 最后是get_tools方法和recommend_product方法，這里也都是模仿了SalesGPT里的寫法：

    def get_tools(self):file_path = r'C:\Users\Administrator\langchain_chatbot\product.txt'knowledge_base = build_knowledge_base(file_path)tools = get_tools(knowledge_base)return toolsdef recommend_product(self,verbose =True):tools = self.get_tools()prompt = CustomPromptTemplateForTools(template=RECOMMEND_TEMPLATE,tools_getter=lambda x: tools,# This omits the `agent_scratchpad`, `tools`, and `tool_names` variables because those are generated dynamically# This includes the `intermediate_steps` variable because that is neededinput_variables=["intermediate_steps",  # 這是在調(diào)用tools時(shí)，會(huì)產(chǎn)生的中間變量，是一個(gè)list里面的一個(gè)tuple，一個(gè)是action，一個(gè)是observation"conversation_history",],)llm_chain = LLMChain(llm=self.llm, prompt=prompt, verbose=verbose)tool_names = [tool.name for tool in tools]# WARNING: this output parser is NOT reliable yet## It makes assumptions about output from LLM which can break and throw an erroroutput_parser = SalesConvoOutputParser()recommend_agent = LLMSingleActionAgent(llm_chain=llm_chain,output_parser=output_parser,stop=["\nObservation:"],allowed_tools=tool_names,)sales_agent_executor = AgentExecutor.from_agent_and_tools(agent=recommend_agent, tools=tools, verbose=verbose, max_iterations=5)inputs = {"conversation_history": "\n".join(self.conversation_history),}response = sales_agent_executor.invoke(inputs)return str(response['output'])

?5. chain.py

? ? ? ? chain這里有三個(gè)類，差異在于使用模板的不同還有部分傳參的不同，這里寫的有點(diǎn)冗余了，大家可以自己優(yōu)化一下：

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from template import STAGE_ANALYZER_INCEPTION_PROMPT,BASIC_TEMPLATE,RECOMMEND_TEMPLATEclass StageAnalyzerChain(LLMChain):"""通過查看聊天記錄判斷是否要轉(zhuǎn)向推薦和銷售階段."""@classmethoddef from_llm(cls, llm, verbose: bool = True) -> LLMChain:"""Get the response parser."""stage_analyzer_inception_prompt_template = STAGE_ANALYZER_INCEPTION_PROMPTprompt = PromptTemplate(template=stage_analyzer_inception_prompt_template,input_variables=["conversation_history","question"],)return cls(prompt=prompt, llm=llm, verbose=verbose)class ConversationChain_Without_Tool(LLMChain):#當(dāng)用戶沒有明確的感興趣話題時(shí)，用這個(gè)chain和用戶閑聊@classmethoddef from_llm(cls, llm, verbose: bool = True) -> LLMChain:"""Get the response parser."""conversation_without_tools_template = BASIC_TEMPLATEprompt = PromptTemplate(template=conversation_without_tools_template,input_variables=["conversation_history",],)return cls(prompt=prompt, llm=llm, verbose=verbose)class Recommend_Product(LLMChain):#當(dāng)用戶有明確的感興趣話題時(shí)，用這個(gè)chain查詢產(chǎn)品庫，看是否命中，如果命中則生成一個(gè)產(chǎn)品推薦信息@classmethoddef from_llm(cls, llm, verbose: bool = True) -> LLMChain:"""Get the response parser."""conversation_without_tools_template = RECOMMEND_TEMPLATEprompt = PromptTemplate(template=conversation_without_tools_template,input_variables=["conversation_history",],)return cls(prompt=prompt, llm=llm, verbose=verbose)

6. my_tools.py

? ? ? ?這個(gè)文件里有有很多，是我把SalesGPT里的一些文件改寫拿過來用的，有一些根據(jù)實(shí)際項(xiàng)目需要進(jìn)行的微調(diào)：

import re
from typing import Unionfrom langchain.agents import Tool
from langchain.chains import RetrievalQA
from langchain.text_splitter import CharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from typing import Callable
from langchain.prompts.base import StringPromptTemplate
from langchain.agents.agent import AgentOutputParser
from langchain.agents.conversational.prompt import FORMAT_INSTRUCTIONS
from langchain.schema import AgentAction, AgentFinish  # OutputParserExceptiondef build_knowledge_base(filepath):with open(filepath, "r", encoding='utf-8') as f:product_catalog = f.read()text_splitter = CharacterTextSplitter(chunk_size=100, chunk_overlap=10)texts = text_splitter.split_text(product_catalog)llm = ChatOpenAI(temperature=0)embeddings = OpenAIEmbeddings()docsearch = Chroma.from_texts(texts, embeddings, collection_name="product-knowledge-base")knowledge_base = RetrievalQA.from_chain_type(llm=llm, chain_type="stuff", retriever=docsearch.as_retriever())return knowledge_basedef get_tools(knowledge_base):# we only use one tool for now, but this is highly extensible!tools = [Tool(name="ProductSearch",func=knowledge_base.invoke,description="查詢產(chǎn)品庫，輸入應(yīng)該是'請介紹一下**的旅游產(chǎn)品'",)]print('tools構(gòu)造正常')return toolsclass CustomPromptTemplateForTools(StringPromptTemplate):# The template to usetemplate: strtools_getter: Callabledef format(self, **kwargs) -> str:# Get the intermediate steps (AgentAction, Observation tuples)# Format them in a particular wayintermediate_steps = kwargs.pop("intermediate_steps")thoughts = ""for action, observation in intermediate_steps:thoughts += action.logthoughts += f"\nObservation: {observation}\nThought: "# Set the agent_scratchpad variable to that valueprint('——thoughts——:'+thoughts+'\n End of ——thoughts——')kwargs["agent_scratchpad"] = thoughtstools = self.tools_getter([])# Create a tools variable from the list of tools providedkwargs["tools"] = "\n".join([f"{tool.name}: {tool.description}" for tool in tools])# Create a list of tool names for the tools providedkwargs["tool_names"] = ", ".join([tool.name for tool in tools])print('prompt構(gòu)造正常')return self.template.format(**kwargs)class SalesConvoOutputParser(AgentOutputParser):ai_prefix: str = "AI"  # change for salesperson_nameverbose: bool = Truedef get_format_instructions(self) -> str:return FORMAT_INSTRUCTIONSdef parse(self, text: str) -> Union[AgentAction, AgentFinish]:if self.verbose:print("TEXT")print(text)print("-------")if f"{self.ai_prefix}:" in text:if "Do I get the answer?YES." in text:print('判斷Agent是否查到結(jié)果，yes')return AgentFinish({"output": text.split(f"{self.ai_prefix}:")[-1].strip()}, text)else:print('判斷Agent是否查到結(jié)果，no')return AgentFinish({"output": {}}, text)regex = r"Action: (.*?)[\n]*Action Input: (.*)"match = re.search(regex, text)if not match:## TODO - this is not entirely reliable, sometimes results in an error.return AgentFinish({"output": "I apologize, I was unable to find the answer to your question. Is there anything else I can help with?"},text,)# raise OutputParserException(f"Could not parse LLM output: `{text}`")action = match.group(1)action_input = match.group(2)print('output_paserser構(gòu)造正常')return AgentAction(action.strip(), action_input.strip(" ").strip('"'), text)@propertydef _type(self) -> str:return "sales-agent"

7. 結(jié)束語

? ? ? ? 整個(gè)項(xiàng)目就是把之前的兩個(gè)項(xiàng)目進(jìn)行了一個(gè)組合拼裝，在這個(gè)過程中可以更好地理解Sales

GPT這個(gè)項(xiàng)目，以及多Agent是怎么運(yùn)行的。