LangChain开发 2

from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory

model = ChatTongyi(model="qwen3-max")

prompt = PromptTemplate.from_template(
  "你需要根据会话历史回应用户问题。对话历史：{chat_history}，用户提问：{input}，请回答！"
)
str_parser = StrOutputParser()
base_chain = prompt | model | str_parser


store = {} # key: session, value: class

# 实现通过会话ID获取InMemoryChatMessageHistory实例对象

def get_history(session_id):
  if session_id not in store:
    store[session_id] = InMemoryChatMessageHistory()
  return store[session_id]

# 创建一个新的链，对原有的链增强功能，自动附加历史消息

conversation_chain = RunnableWithMessageHistory(
  base_chain, # 被增强的原有chain
  get_history, # 通货会话ID获取InMemoryChatMessageHistory类对象
  input_messages_key="input", # 用户输入的占位符
  history_messages_key="chat_history", # 历史消息在模板中的占位符
)

if __name__ == "__main__":
  # 固定格式：添加langchain配置，为当前程序配置所属的session_id
  session_config = {
    "configurable": {
      "session_id": "user_001"
    }
  }
  res = conversation_chain.invoke({"input": "小明有两个猫"}, session_config)
  print("第 1 次执行!", res)
  res = conversation_chain.invoke({"input": "小王有三只狗"}, session_config)
  print("第 2 次执行!", res)
  res = conversation_chain.invoke({"input": "一共有多少宠物?"}, session_config)
  print("第 3 次执行!", res)