2025-09-12 langchain学习-04 使用MessagesPlaceholder制作多轮对话模型

from langchain.prompts.chat import ChatPromptTemplate,MessagesPlaceholder
from langchain.prompts import SystemMessagePromptTemplate,HumanMessagePromptTemplate
from langchain.schema import AIMessage,HumanMessage
 
 #提示词模板
chat_template=ChatPromptTemplate.from_messages(
  [
    SystemMessagePromptTemplate.from_template("你的工作是{work},可以回答用户向你提出的专业相关的问题"),
    MessagesPlaceholder(variable_name="history"),
    HumanMessagePromptTemplate.from_template("{text}")
  ]
)

#模拟之前的历史对话
chat_history=[
  HumanMessage("介绍一下五代十国"),
  AIMessage("""
**时间：** 大约公元907年 - 960年
**地点：** 中国
**背景：** 唐朝灭亡后，中国陷入了长期的分裂和战乱。
**主要内容：**
*   **五代：** 指的是在中原地区相继建立的五个朝代，分别是：
    *   后梁 (907年 - 923年)
    *   后唐 (923年 - 936年)
    *   后晋 (936年 - 947年)
    *   后汉 (947年 - 951年)
    *   后周 (951年 - 960年)
    这五个朝代都建立于黄河流域，统治着中原地区，更迭非常迅速，战乱频繁。
*   **十国：** 指的是在五代之外，同时期存在的十个割据政权，主要分布在南方和山西等地，分别是：
    *   吴 (902年 - 937年)南唐 (937年 - 975年)
    *   吴越 (907年 - 978年)
    *   前蜀 (907年 - 925年)
    *   后蜀 (934年 - 965年)
    *   南汉 (917年 - 971年)
    *   楚 (907年 - 951年)
    *   闽 (909年 - 945年)
    *   荆南 (南平) (924年 - 963年)
    *   北汉 (951年 - 979年)
    这十个政权相对稳定一些，经济发展也较好。五代十国是中国历史上一个动荡不安的时期，政权更迭频繁，战乱不断。但同时也为后来的宋朝统一中国奠定了基础，并在一定程度上促进了区域文化的发展。
""")
]

prompt=chat_template.format_prompt(
  work="历史老师",
  history=chat_history,
  text="他们分别哪些人建立的国家？"
).to_messages()

print(prompt)

import os
from zhipuai import ZhipuAI
from typing import Any,Iterator,List,Optional

#对话模型基类
from langchain.chat_models.base import BaseChatModel

#提示词模板
from langchain.prompts.chat import ChatPromptTemplate,MessagesPlaceholder
from langchain.prompts import SystemMessagePromptTemplate,HumanMessagePromptTemplate

#消息种类
from langchain.schema import(
  AIMessage,
  BaseMessage,
  ChatResult,
  ChatGeneration,
  HumanMessage,
  SystemMessage
)

#流式输出返回类型
from langchain.schema.output import ChatGenerationChunk
from langchain.schema.messages import AIMessageChunk

#过程管理
from langchain_core.callbacks.manager import CallbackManagerForLLMRun

zhipu_api_key=os.getenv("ZHIPU_API_KEY")

def Msg2Dict(message:List[BaseMessage])->List[dict]:
  result=[]
  for msg in message:
    if isinstance(msg,AIMessage):
      role="assistant"
    elif isinstance(msg,SystemMessage):
      role="system"
    elif isinstance(msg,HumanMessage):
      role="user"
    result.append({"role":role,"content":msg.content})
  return result

class ChatGLM3(BaseChatModel):
  client:ZhipuAI=None

  def __init__(
      self,
      api_key:str=zhipu_api_key,
      **kwargs
  ):
    super().__init__(**kwargs)
    self.client=ZhipuAI(api_key=api_key)

  @property
  def _llm_type(self)->str:
    return "ChatGLM3"
  
  def _generate(
      self,
      message:List[BaseMessage],
      stop:Optional[List[str]]=None,
      run_manager:Optional[CallbackManagerForLLMRun]=None,
      **kwargs
  )->ChatResult:
    messages=Msg2Dict(message=message)
    response=self.client.chat.completions.create(
      model="glm-3-turbo",
      messages=messages,
      **kwargs
    )
    result=response.choices[0].message.content
    return ChatResult(generations=ChatGeneration(message=AIMessage(content=result)))
  
  def _stream(
      self,
      message:List[BaseMessage],
      stop:Optional[List[str]]=None,
      run_manager:Optional[CallbackManagerForLLMRun]=None,
      stream=True,
      **kwargs
  )->Iterator[ChatGenerationChunk]:
    messages=Msg2Dict(message=message)
    response=self.client.chat.completions.create(
      model="glm-3-turbo",
      messages=messages,
      stream=True,
      **kwargs
    )
    for chunk in response:
      result=chunk.choices[0].delta.content
      yield ChatGenerationChunk(message=AIMessageChunk(content=result))
  
def main():
  glm=ChatGLM3()

  subject="历史"
  print(f"{subject}学家：你好，我是一名专业的{subject}学家，你可以向我询问有关{subject}的问题，回复\"exit\"或\"quit\"结束会话")

  chat_template=ChatPromptTemplate.from_messages(
    [
      SystemMessagePromptTemplate.from_template("你是一名专业的{subject}学家，可以用简单明了的语言回答用户的问题"),
      MessagesPlaceholder(variable_name="history"),
      HumanMessagePromptTemplate.from_template("{text}")
    ]
  )

  chat_history=[]

  while True:
    user_input=input("你：")
    try:
      if user_input.lower() in ["exit","quit"]:
        print(f"{subject}学家：再见！")
        break
      prompt=chat_template.format_prompt(
        subject=subject,
        history=chat_history,
        text=user_input
      ).to_messages()

      print(f"{subject}学家：",end="",flush=True)
      full_response=""
      for chunk in glm.stream(prompt):
        print(chunk.content,end="",flush=True)
        full_response+=chunk.content
      
      chat_history.append(HumanMessage(user_input))
      chat_history.append(AIMessage(full_response))
    except KeyboardInterrupt:
      print("再见！")
      break

if __name__=='__main__':
  main()