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)

import os, json
from langchain_core.messages import message_to_dict, messages_from_dict
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.prompts import PromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory

# message_to_dict: 单个消息对象（BaseMessage类） -> 字典
# message_from_dict: [字典、字典...] -> [消息、消息...]
# AIMessage、HumanMessage、SystemMessage都是BaseMessage的子类

'''FileChatMessageHistory'''

class FileChatMessageHistory(BaseChatMessageHistory):
  def __init__(self, session_id, storage_path):
    self.session_id = session_id
    self.storage_path = storage_path
    self.file_path = os.path.join(self.storage_path, self.session_id)
    
    # 确保文件夹是存在的，若不存在会创建
    os.makedirs(os.path.dirname(self.file_path), exist_ok=True)

  def add_messages(self, messages):
    # Sequence序列，类似list,tuple
    all_messages = list(self.messages) # 已有的消息列表
    all_messages.extend(messages) # 新老融合

    # 将数据同步写入本地文件中
    # 类对象写入文件: 01编码
    # 将01编码写成字典，再转成json，转字符串
    new_messages = [message_to_dict(message) for message in all_messages]
    with open(self.file_path, "w", encoding="utf-8") as f:
      json.dump(new_messages, f, ensure_ascii=False)
  
  @property # property装饰器将messages方法变成成员属性用
  def messages(self):
    try:
      with open(self.file_path, "r", encoding="utf-8") as f:
        messages_data = json.load(f)
        return messages_from_dict(messages_data)
    except FileNotFoundError:
      return []
    
  def clear(self):
    with open(self.file_path, "w", encoding="utf-8") as f:
      json.dump([], f) 
      
model = ChatTongyi(
  model="qwen3-max",
  dashscope_api_key=os.environ.get("DASHSCOPE_API_KEY", "")
)

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


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

def get_history(session_id):
  return FileChatMessageHistory(session_id, "./chat_history")

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

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)

from langchain_community.document_loaders import CSVLoader

loader = CSVLoader(
  file_path="./data/stu.csv"
)

# load 文件
documents = loader.load()

# 打印数据
for document in documents:
  print(document)