2025-09-11 langchain学习-02 用langchain对大语言模型做自定义集成

import os
from langchain.llms.base import LLM
from langchain_core.callbacks.manager import CallbackManagerForLLMRun
from langchain_core.outputs import GenerationChunk
from typing import Any,List,Iterator,Optional
from zhipuai import ZhipuAI

zhipu_api_key=os.getenv("ZHIPU_API_KEY")

class ChatGLM3(LLM):
  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 _call(
      self,
      prompt:str,
      stop:Optional[List[str]]=None,
      run_manager:Optional[CallbackManagerForLLMRun]=None,
      **kwargs:Any
  )->str:
    history=kwargs.get("history",[])
    messages=history+[{"role":"user","content":prompt}]
    respone=self.client.chat.completions.create(
      model='glm-3-turbo',
      messages=messages
    )
    result=respone.choices[0].message.content
    return result
  
  def _stream(
      self,
      prompt:str,
      stop:Optional[List[str]]=None,
      run_manager:Optional[CallbackManagerForLLMRun]=None,
      **kwargs:Any
  )->Iterator[GenerationChunk]:
    history=kwargs.get("history",[])
    messages=history+[{"role":"user","content":prompt}]
    respone=self.client.chat.completions.create(
      model='glm-3-turbo',
      messages=messages,
      stream=True
    )
    for chunk in respone:
      result=chunk.choices[0].delta.content
      yield GenerationChunk(text=result)