当前位置:   article > 正文

【Langchain多Agent实践】一个有推销功能的旅游聊天机器人

langchain多agent

【Langchain+Streamlit】旅游聊天机器人_langchain streamlit-CSDN博客

视频讲解地址:【Langchain Agent】带推销功能的旅游聊天机器人_哔哩哔哩_bilibili

体验地址:   http://101.33.225.241:8503/

github地址:GitHub - jerry1900/langchain_chatbot: langchain+streamlit打造的一个有memory的旅游聊天机器人,可以和你聊旅游相关的事儿

 

        之前,我们介绍如何打算一款简单的旅游聊天机器人,而且之前我们介绍了SalesGPT,我们看能不能把这两个东西结合起来,让我们的旅游聊天机器人具有推销产品的功能。我们先来看看效果:

        首先,你可以和机器人闲聊关于旅游的事儿(如果你问的问题和旅游无关的话,会提示你只回答旅游问题) ;其次,当你连续询问有关同一个地点时(比如北京),机器人会检查自己的本地知识库,看看产品库里有没有相关的旅游产品,如果有的话就推荐给客户,如果没有就输出一个空的字符串,用户是没有感知的,我们来看一下是如何实现的。

1. 项目结构

        我们是在原来项目基础上逐步叠加的,主要增加了一个agent.py、my_tools.py、stages.py等文件。我们这次的项目是使用poetry来管理和运行的:

        项目结构如图:

        我们新加了一个产品文件用于存储旅游产品,下面是三个产品中的一个:

产品名称:北京一日游
产品价格:299元
产品内容:{
北京作为中国的首都和历史文化名城,有许多令人着迷的景点和活动。以下是一个充满活力和文化的北京一日游的建议:
早上:天安门广场: 开始您的一日游,您可以前往天安门广场,这是世界上最大的城市广场之一,也是中国的政治中心。您可以欣赏到天安门城楼,参观升国旗仪式(早上升旗时间)。
故宫博物院: 天安门广场北侧就是故宫,这是中国最大、最完整的古代皇家宫殿建筑群。在这里,您可以领略中国古代皇家建筑的壮丽和深厚的历史文化。
中午:午餐: 您可以选择在附近的餐馆品尝地道的北京菜,比如炸酱面、北京烤鸭等。
下午:颐和园: 中午过后,您可以前往颐和园,这是中国最大的皇家园林之一,也是清代的皇家园林。园内有美丽的湖泊、精致的建筑和独特的风景。
什刹海: 在下午的最后时段,您可以前往什刹海地区,这里是一个古老而又时尚的区域,有着许多酒吧、咖啡馆和特色小店。您可以漫步在湖边,欣赏夕阳下的美景,体验北京的悠闲氛围。
晚上:王府井步行街: 晚上,您可以前往王府井步行街,这是北京最繁华的购物街之一,有着各种商店、餐馆和娱乐场所。您可以尝试美食、购物或者观赏街头表演。
京剧表演: 如果时间允许,您还可以观看一场京剧表演,京剧是中国传统戏曲的代表之一,有着悠久的历史和独特的艺术魅力。
}

  2. chat.py的改动,新增了欢迎词,添加了Agent构造的方法

        这里构造一个专门负责提示词的Agent(其实就是一个LLMChain),并构造一个负责会话和判断功能的ConversationAgent,让这个agent初始化并构造一个负责判断阶段的内部agent,我们把他们都要放到session里:

  1. #需要国内openai开发账号的请联系微信 15652965525
  2. if "welcome_word" not in st.session_state:
  3. st.session_state.welcome_word = welcome_agent()
  4. st.session_state.messages.append({'role': 'assistant', 'content': st.session_state.welcome_word['text']})
  5. st.session_state.agent = ConversationAgent()
  6. st.session_state.agent.seed_agent()
  7. st.session_state.agent.generate_stage_analyzer(verbose=True)

        在用户输入后的每一步,先进行一下阶段判断,然后调用agent的human_step方法,再调用agent的step()方法,完成一轮对话:

  

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

   

3. welcome_agent

        这个比较简单,就是一个咱们学习过无数遍的一个简单的chain:

  1. def welcome_agent():
  2. llm = OpenAI(
  3. temperature=0.6,
  4. # openai_api_key=os.getenv("OPENAI_API_KEY"),
  5. openai_api_key=st.secrets['api']['key'],
  6. # base_url=os.getenv("OPENAI_BASE_URL")
  7. base_url=st.secrets['api']['base_url']
  8. )
  9. prompt = PromptTemplate.from_template(WELCOME_TEMPLATE)
  10. chain = LLMChain(
  11. llm=llm,
  12. prompt=prompt,
  13. verbose=True,
  14. )
  15. response = chain.invoke("简短的欢迎词")
  16. return response

        这里我们希望每次调用它的时候,可以得到一些不一样的、有创意的欢迎词,所以我们的temperature调的比较高,这样它可能生成一些有创意的欢迎词。

4. ConversationAgent类

        这个类是我们的核心类,里面有很多属性和方法,我们用python的dir()方法来看一下它里面的结构:

  1. from agent import ConversationAgent
  2. agent = ConversationAgent()
  3. print(dir(ConversationAgent))

        里面以_开头的是Object基本类自带的属性和方法,其他的是我们构造的属性和方法:

['__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']

        我们先来看类的属性和一些简单的方法,注意我们这里构造了一个llm,之后下面有很多方法会用到这个llm:

  1. class ConversationAgent():
  2. stage_analyzer_chain: StageAnalyzerChain = Field(...)
  3. conversation_agent_without_tool = Field()
  4. conversation_agent_with_tool = Field()
  5. conversation_history = []
  6. conversation_stage_id: str = "1"
  7. current_conversation_stage: str = CONVERSATION_STAGES.get("1")
  8. llm = OpenAI(
  9. temperature=0,
  10. openai_api_key=st.secrets['api']['key'],
  11. base_url=st.secrets['api']['base_url']
  12. )
  13. def seed_agent(self):
  14. self.conversation_history.clear()
  15. print("——Seed Successful——")
  16. def show_chat_history(self):
  17. return self.conversation_history
  18. def retrieve_conversation_stage(self, key):
  19. return CONVERSATION_STAGES.get(key)

        我们继续来看:

  1. def fake_step(self):
  2. input_text = self.conversation_history[-1]
  3. ai_message = self._respond_with_tools(str(input_text), verbose=True)
  4. print(ai_message,type(ai_message['output']))
  5. def step(self):
  6. input_text = self.conversation_history[-1]
  7. print(str(input_text)+'input_text****')
  8. if int(self.conversation_stage_id) == 1:
  9. ai_message = self._respond_without_tools(str(input_text),verbose=True)
  10. else:
  11. chat_message = self._respond_without_tools(str(input_text), verbose=True)
  12. recommend_message = self.recommend_product()
  13. print(recommend_message,len(recommend_message))
  14. if len(recommend_message)<=5:
  15. ai_message = chat_message
  16. else:
  17. ai_message = chat_message + '\n\n' + recommend_message
  18. # output_dic = self._respond_with_tools(str(input_text),verbose=True)
  19. # ai_message = str(output_dic['output'])
  20. print(ai_message,type(ai_message))
  21. ai_message = "AI:"+str(ai_message)
  22. self.conversation_history.append(ai_message)
  23. # print(f"——系统返回消息'{ai_message}',并添加到history里——")
  24. return ai_message.lstrip('AI:')

        fake_step是一个模拟输出的方法,不用管,测试的时候用;step方法是接收用户的输入,从聊天记录里取出来(input_text = self.conversation_history[-1]) ,然后再根据不同的对话阶段进行不同的逻辑,如果是第二个阶段推销阶段,那么就调用recommend_product方法去生成一个推销产品的信息,并把两条信息拼接起来。

        

  1. def human_step(self,input_text):
  2. human_message = input_text
  3. human_message = "用户: " + human_message
  4. self.conversation_history.append(human_message)
  5. # print(f"——用户输入消息'{human_message}',并添加到history里——")
  6. return human_message

        human_step方法比较简单,就是把用户的输入挂到conversation_history聊天记录里。然后是构造阶段判断的agent和阶段判断的方法,这些都是模仿SalesGPT里的,做了一些调整:

  1. def generate_stage_analyzer(self,verbose: bool = False):
  2. self.stage_analyzer_chain = StageAnalyzerChain.from_llm(
  3. llm=self.llm,
  4. verbose=verbose
  5. )
  6. print("成功构造一个StageAnalyzerChain")
  7. def determine_conversation_stage(self,question):
  8. self.question = question
  9. print('-----进入阶段判断方法-----')
  10. self.conversation_stage_id = self.stage_analyzer_chain.run(
  11. conversation_history=self.conversation_history,
  12. question=self.question
  13. )
  14. print(f"Conversation Stage ID: {self.conversation_stage_id}")
  15. print(type(self.conversation_stage_id))
  16. self.current_conversation_stage = self.retrieve_conversation_stage(
  17. self.conversation_stage_id
  18. )
  19. print(f"Conversation Stage: {self.current_conversation_stage}")

         然后是_respond_without_tools这么一个内部的方法,它在step里被调用:

  1. def _respond_without_tools(self,input_text,verbose: bool = False):
  2. self.conversation_agent_without_tool = ConversationChain_Without_Tool.from_llm(
  3. llm=self.llm,
  4. verbose=verbose
  5. )
  6. response = self.conversation_agent_without_tool.run(
  7. question = input_text,
  8. conversation_history=self.conversation_history,
  9. )
  10. return response

        最后是get_tools方法和recommend_product方法,这里也都是模仿了SalesGPT里的写法:

  1. def get_tools(self):
  2. file_path = r'C:\Users\Administrator\langchain_chatbot\product.txt'
  3. knowledge_base = build_knowledge_base(file_path)
  4. tools = get_tools(knowledge_base)
  5. return tools
  6. def recommend_product(self,verbose =True):
  7. tools = self.get_tools()
  8. prompt = CustomPromptTemplateForTools(
  9. template=RECOMMEND_TEMPLATE,
  10. tools_getter=lambda x: tools,
  11. # This omits the `agent_scratchpad`, `tools`, and `tool_names` variables because those are generated dynamically
  12. # This includes the `intermediate_steps` variable because that is needed
  13. input_variables=[
  14. "intermediate_steps", # 这是在调用tools时,会产生的中间变量,是一个list里面的一个tuple,一个是action,一个是observation
  15. "conversation_history",
  16. ],
  17. )
  18. llm_chain = LLMChain(llm=self.llm, prompt=prompt, verbose=verbose)
  19. tool_names = [tool.name for tool in tools]
  20. # WARNING: this output parser is NOT reliable yet
  21. ## It makes assumptions about output from LLM which can break and throw an error
  22. output_parser = SalesConvoOutputParser()
  23. recommend_agent = LLMSingleActionAgent(
  24. llm_chain=llm_chain,
  25. output_parser=output_parser,
  26. stop=["\nObservation:"],
  27. allowed_tools=tool_names,
  28. )
  29. sales_agent_executor = AgentExecutor.from_agent_and_tools(
  30. agent=recommend_agent, tools=tools, verbose=verbose, max_iterations=5
  31. )
  32. inputs = {
  33. "conversation_history": "\n".join(self.conversation_history),
  34. }
  35. response = sales_agent_executor.invoke(inputs)
  36. return str(response['output'])

 5. chain.py

        chain这里有三个类,差异在于使用模板的不同还有部分传参的不同,这里写的有点冗余了,大家可以自己优化一下:

  1. from langchain.chains import LLMChain
  2. from langchain.prompts import PromptTemplate
  3. from template import STAGE_ANALYZER_INCEPTION_PROMPT,BASIC_TEMPLATE,RECOMMEND_TEMPLATE
  4. class StageAnalyzerChain(LLMChain):
  5. """通过查看聊天记录判断是否要转向推荐和销售阶段."""
  6. @classmethod
  7. def from_llm(cls, llm, verbose: bool = True) -> LLMChain:
  8. """Get the response parser."""
  9. stage_analyzer_inception_prompt_template = STAGE_ANALYZER_INCEPTION_PROMPT
  10. prompt = PromptTemplate(
  11. template=stage_analyzer_inception_prompt_template,
  12. input_variables=[
  13. "conversation_history",
  14. "question"
  15. ],
  16. )
  17. return cls(prompt=prompt, llm=llm, verbose=verbose)
  18. class ConversationChain_Without_Tool(LLMChain):
  19. #当用户没有明确的感兴趣话题时,用这个chain和用户闲聊
  20. @classmethod
  21. def from_llm(cls, llm, verbose: bool = True) -> LLMChain:
  22. """Get the response parser."""
  23. conversation_without_tools_template = BASIC_TEMPLATE
  24. prompt = PromptTemplate(
  25. template=conversation_without_tools_template,
  26. input_variables=[
  27. "conversation_history",
  28. ],
  29. )
  30. return cls(prompt=prompt, llm=llm, verbose=verbose)
  31. class Recommend_Product(LLMChain):
  32. #当用户有明确的感兴趣话题时,用这个chain查询产品库,看是否命中,如果命中则生成一个产品推荐信息
  33. @classmethod
  34. def from_llm(cls, llm, verbose: bool = True) -> LLMChain:
  35. """Get the response parser."""
  36. conversation_without_tools_template = RECOMMEND_TEMPLATE
  37. prompt = PromptTemplate(
  38. template=conversation_without_tools_template,
  39. input_variables=[
  40. "conversation_history",
  41. ],
  42. )
  43. return cls(prompt=prompt, llm=llm, verbose=verbose)

6. my_tools.py

       这个文件里有有很多,是我把SalesGPT里的一些文件改写拿过来用的,有一些根据实际项目需要进行的微调:

  1. import re
  2. from typing import Union
  3. from langchain.agents import Tool
  4. from langchain.chains import RetrievalQA
  5. from langchain.text_splitter import CharacterTextSplitter
  6. from langchain_community.vectorstores import Chroma
  7. from langchain_openai import ChatOpenAI, OpenAIEmbeddings
  8. from typing import Callable
  9. from langchain.prompts.base import StringPromptTemplate
  10. from langchain.agents.agent import AgentOutputParser
  11. from langchain.agents.conversational.prompt import FORMAT_INSTRUCTIONS
  12. from langchain.schema import AgentAction, AgentFinish # OutputParserException
  13. def build_knowledge_base(filepath):
  14. with open(filepath, "r", encoding='utf-8') as f:
  15. product_catalog = f.read()
  16. text_splitter = CharacterTextSplitter(chunk_size=100, chunk_overlap=10)
  17. texts = text_splitter.split_text(product_catalog)
  18. llm = ChatOpenAI(temperature=0)
  19. embeddings = OpenAIEmbeddings()
  20. docsearch = Chroma.from_texts(
  21. texts, embeddings, collection_name="product-knowledge-base"
  22. )
  23. knowledge_base = RetrievalQA.from_chain_type(
  24. llm=llm, chain_type="stuff", retriever=docsearch.as_retriever()
  25. )
  26. return knowledge_base
  27. def get_tools(knowledge_base):
  28. # we only use one tool for now, but this is highly extensible!
  29. tools = [
  30. Tool(
  31. name="ProductSearch",
  32. func=knowledge_base.invoke,
  33. description="查询产品库,输入应该是'请介绍一下**的旅游产品'",
  34. )
  35. ]
  36. print('tools构造正常')
  37. return tools
  38. class CustomPromptTemplateForTools(StringPromptTemplate):
  39. # The template to use
  40. template: str
  41. tools_getter: Callable
  42. def format(self, **kwargs) -> str:
  43. # Get the intermediate steps (AgentAction, Observation tuples)
  44. # Format them in a particular way
  45. intermediate_steps = kwargs.pop("intermediate_steps")
  46. thoughts = ""
  47. for action, observation in intermediate_steps:
  48. thoughts += action.log
  49. thoughts += f"\nObservation: {observation}\nThought: "
  50. # Set the agent_scratchpad variable to that value
  51. print('——thoughts——:'+thoughts+'\n End of ——thoughts——')
  52. kwargs["agent_scratchpad"] = thoughts
  53. tools = self.tools_getter([])
  54. # Create a tools variable from the list of tools provided
  55. kwargs["tools"] = "\n".join(
  56. [f"{tool.name}: {tool.description}" for tool in tools]
  57. )
  58. # Create a list of tool names for the tools provided
  59. kwargs["tool_names"] = ", ".join([tool.name for tool in tools])
  60. print('prompt构造正常')
  61. return self.template.format(**kwargs)
  62. class SalesConvoOutputParser(AgentOutputParser):
  63. ai_prefix: str = "AI" # change for salesperson_name
  64. verbose: bool = True
  65. def get_format_instructions(self) -> str:
  66. return FORMAT_INSTRUCTIONS
  67. def parse(self, text: str) -> Union[AgentAction, AgentFinish]:
  68. if self.verbose:
  69. print("TEXT")
  70. print(text)
  71. print("-------")
  72. if f"{self.ai_prefix}:" in text:
  73. if "Do I get the answer?YES." in text:
  74. print('判断Agent是否查到结果,yes')
  75. return AgentFinish(
  76. {"output": text.split(f"{self.ai_prefix}:")[-1].strip()}, text)
  77. else:
  78. print('判断Agent是否查到结果,no')
  79. return AgentFinish({"output": {}}, text)
  80. regex = r"Action: (.*?)[\n]*Action Input: (.*)"
  81. match = re.search(regex, text)
  82. if not match:
  83. ## TODO - this is not entirely reliable, sometimes results in an error.
  84. return AgentFinish(
  85. {
  86. "output": "I apologize, I was unable to find the answer to your question. Is there anything else I can help with?"
  87. },
  88. text,
  89. )
  90. # raise OutputParserException(f"Could not parse LLM output: `{text}`")
  91. action = match.group(1)
  92. action_input = match.group(2)
  93. print('output_paserser构造正常')
  94. return AgentAction(action.strip(), action_input.strip(" ").strip('"'), text)
  95. @property
  96. def _type(self) -> str:
  97. return "sales-agent"

7. 结束语

        整个项目就是把之前的两个项目进行了一个组合拼装,在这个过程中可以更好地理解Sales

GPT这个项目,以及多Agent是怎么运行的。

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

闽ICP备14008679号