2025-09-13 langchain学习-06 输出解析器：csv解析器

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

#输出解析器
from langchain.output_parsers import CommaSeparatedListOutputParser


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"
    else:
      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:Any
  ):
    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:Any
  )->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(result))])
  
  def _stream(
      self,
      message:List[BaseMessage],
      stop:Optional[List[str]]=None,
      run_manager:Optional[CallbackManagerForLLMRun]=None,
      **kwargs:Any
  )->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))

from langchain.output_parser import CommaSeparatedListOutputParser

def main()
    #创建解析器实例
    output_parser=CommaSeparatedListOutputParser()
    #获取格式化指令
    format_instructions=out_parser.get_format_instructions()
    #创建包含格式化指令的提示词模板
    chat_template=ChatPromptTemplate.from_message(
        [
            SystemPromptTemplate.from_template("你是一个提示词生成小助手，可以根据用户输入的主题给出关键词。\n{instructions}"),
            HumanPromptTemplate.from_temolate("以{subject}为主题给出五个关键词")
        ]
    )
    #将格式化指令注入其中
    final_template=chat_template.partial(instructions=format_instructions)
    
    #形成提示词
    Prompt=final_template.format_prompt(subject="RAG技术").to_messages()
    
    #下面放进glm里看看结果
    glm=ChatGLM3()
    
    full_response="
    for chunk in glm.stream(prompt):
        print(chunk.content,end="",Flush=True)
        full_response+=chunk.content
    print()
    print(output_parser.parse(full_response))
    
if __name__=='__main__':
    main()