1. ChatOpenAI #

• ChatOpenAI是一个用于与 OpenAI 聊天模型（如 GPT-3.5、GPT-4、GPT-4o 等）进行对话交互的核心类。
• 它封装了与 OpenAI 聊天接口的调用流程，让开发者可以像使用普通 Python 类一样便捷地发送消息并获取模型回复。
• 您只需指定使用的模型名称，即可实例化 ChatOpenAI 对象，然后通过 .invoke() 方法传入用户消息，获取 AI 的回复内容，非常适合用于问答、对话机器人、智能助手等场景。

1.1. 1.openai.py #

1.openai.py

#from langchain_openai import ChatOpenAI
# 从 langchain 导入 ChatOpenAI 类
from smartchain.chat_models import ChatOpenAI
# 创建一个 ChatOpenAI 实例，使用 "gpt-4o" 模型
llm = ChatOpenAI(model="gpt-4o")
# 调用模型，传入用户消息，返回 AI 的回复结果
result = llm.invoke("你好，你是谁")
# 打印 AI 回复的内容
print(result.content)

1.2. init.py #

smartchain/init.py

1.3. messages.py #

smartchain/messages.py

# 定义基础消息类
class BaseMessage:
    # 基础消息类的文档字符串
    """基础消息类"""

    # 初始化方法，content为消息内容，其余为可选参数
    def __init__(self, content: str, **kwargs):
        # 初始化消息
        # content: 消息内容
        # **kwargs: 其他可选参数
        self.content = content  # 保存消息内容
        self.type = kwargs.get("type", "base")  # 获取消息类型，默认为"base"
        for key, value in kwargs.items():  # 遍历所有其他参数
            if key != "type":  # 排除type参数
                setattr(self, key, value)  # 设置为成员变量

    # 定义当对象被str()或print时的输出内容
    def __str__(self):
        return self.content  # 返回消息内容

    # 定义对象的官方字符串表示，用于debug
    def __repr__(self):
        return f"{self.__class__.__name__}(content={self.content!r})"  # 返回类名和内容

# 定义用户消息类，继承自BaseMessage
class HumanMessage(BaseMessage):
    # 用户消息类的文档字符串
    """用户消息"""

    # 初始化方法，调用父类构造方法，并指定type为"human"
    def __init__(self, content: str, **kwargs):
        # 调用父类构造方法，并固定type为"human"
        super().__init__(content, type="human", **kwargs)

# 定义AI消息类，继承自BaseMessage
class AIMessage(BaseMessage):
    # AI消息类的文档字符串
    """AI 消息"""

    # 初始化方法，调用父类构造方法，并指定type为"ai"
    def __init__(self, content: str, **kwargs):
        # 调用父类构造方法，并固定type为"ai"
        super().__init__(content, type="ai", **kwargs)

1.4. chat_models.py #

smartchain/chat_models.py

# 导入操作系统相关模块
import os

# 导入 openai 模块
import openai

# 从 .messages 模块导入 AIMessage、HumanMessage 和 SystemMessage 类
from .messages import AIMessage, HumanMessage, SystemMessage

# 定义与 OpenAI 聊天模型交互的类
class ChatOpenAI:

    # 初始化方法
    def __init__(self, model: str = "gpt-4o", **kwargs):
        # 初始化 ChatOpenAI 类
        """
        初始化 ChatOpenAI

        Args:
            model: 模型名称，如 "gpt-4o"
            **kwargs: 其他参数（如 temperature, max_tokens 等）
        """
        # 设置模型名称
        self.model = model
        # 获取 api_key，优先从参数获取，否则从环境变量获取
        self.api_key = kwargs.get("api_key") or os.getenv("OPENAI_API_KEY")
        # 如果没有提供 api_key，则抛出异常
        if not self.api_key:
            raise ValueError("需要提供 api_key 或设置 OPENAI_API_KEY 环境变量")
        # 保存除 api_key 之外的其他参数，用于 API 调用
        self.model_kwargs = {k: v for k, v in kwargs.items() if k != "api_key"}
        # 创建 OpenAI 客户端实例
        self.client = openai.OpenAI(api_key=self.api_key)

    # 调用模型生成回复的方法
    def invoke(self, input, **kwargs):
        # 调用模型生成回复
        """
        调用模型生成回复

        Args:
            input: 输入内容，可以是字符串或消息列表
            **kwargs: 额外的 API 参数

        Returns:
            AIMessage: AI 的回复消息
        """
        # 将输入数据转换为消息格式
        messages = self._convert_input(input)
        # 构建 API 请求参数字典
        params = {
            "model": self.model,
            "messages": messages,
            **self.model_kwargs,
            **kwargs
        }
        # 使用 OpenAI 客户端发起 chat.completions.create 调用获取回复
        response = self.client.chat.completions.create(**params)
        # 取出返回结果中的第一个选项
        choice = response.choices[0]
        # 获取消息内容
        content = choice.message.content or ""
        # 返回一个 AIMessage 对象
        return AIMessage(content=content)

    # 流式调用模型生成回复的方法
    def stream(self, input, **kwargs):
        # 流式调用模型生成回复
        """
        流式调用模型生成回复

        Args:
            input: 输入内容，可以是字符串或消息列表
            **kwargs: 额外的 API 参数

        Yields:
            AIMessage: AI 的回复消息块（每次产生部分内容）
        """
        # 将输入数据转换为消息格式
        messages = self._convert_input(input)
        # 构建 API 请求参数字典，启用流式输出
        params = {
            "model": self.model,
            "messages": messages,
            "stream": True,  # 启用流式输出
            **self.model_kwargs,
            **kwargs
        }
        # 使用 OpenAI 客户端发起流式调用
        stream = self.client.chat.completions.create(**params)
        # 迭代流式响应
        for chunk in stream:
            # 检查是否有内容增量
            if chunk.choices and len(chunk.choices) > 0:
                delta = chunk.choices[0].delta
                # 检查 delta 中是否有 content，如果有则发送
                if hasattr(delta, 'content') and delta.content:
                    # 产生包含部分内容的 AIMessage
                    yield AIMessage(content=delta.content)

    # 内部方法，将输入转换为 OpenAI API 需要的消息格式
    def _convert_input(self, input):
        # 将输入转换为 OpenAI API 需要的消息格式
        """
        将输入转换为 OpenAI API 需要的消息格式

        Args:
            input: 字符串、消息列表或 ChatPromptValue

        Returns:
            list[dict]: OpenAI API 格式的消息列表
        """
        # 输入为字符串时，直接封装为用户角色消息
        if isinstance(input, str):
            return [{"role": "user", "content": input}]
        else:
            # 其他输入类型，转为字符串作为 user 消息
            return [{"role": "user", "content": str(input)}]

2. ChatDeepSeek #

• ChatDeepSeek是一个用于集成 DeepSeek Chat API 的大语言模型（LLM）接口，实现方式上与 OpenAI 的聊天模型类似。
• 它通过封装底层 API 的请求、消息格式处理、以及响应解析，为用户提供了灵活调用 DeepSeek 聊天模型的能力。

主要功能

• 模型选择：支持自定义 DeepSeek 提供的模型名称（如 "deepseek-chat"）。
• API 密钥管理：可通过参数或环境变量传入 api_key，保障调用的安全性。
• 消息格式转换：自动将字符串、消息对象等不同格式的输入转换为 DeepSeek 所需的消息格式。
• 流式输出支持：支持流式响应，便于实时获取生成内容。
• 与 LangChain 风格兼容：保持使用习惯与 OpenAI 聊天模型一致，方便快速切换和集成。

常见用法

1. 实例化模型对象，指定模型名称和密钥；
2. 调用 .invoke() 或相应方法，传入用户输入，获取 AI 回复；
3. 支持额外传递温度、最大 token 数等参数自定义生成行为。

2.1. 2.ChatDeepSeek.py #

2.ChatDeepSeek.py

# uv add  langchain_deepseek
# from langchain_deepseek import ChatDeepSeek
# 从 langchain 导入 ChatDeepSeek 类
from smartchain.chat_models import ChatDeepSeek

# 创建一个 ChatDeepSeek 实例，使用 "deepseek-chat" 模型
llm = ChatDeepSeek(model="deepseek-chat",api_key="sk-bb99cf132b184a169b5e053b346a7c25")

# 调用模型，传入用户消息，返回 AI 的回复结果
result = llm.invoke("你好，你是谁")

# 打印 AI 回复的内容
print(result.content)

2.2. chat_models.py #

smartchain/chat_models.py

# 导入操作系统相关模块
import os

# 导入 openai 模块
import openai

# 从 .messages 模块导入 AIMessage、HumanMessage 和 SystemMessage 类
from .messages import AIMessage, HumanMessage, SystemMessage

# 定义与 OpenAI 聊天模型交互的类
class ChatOpenAI:

    # 初始化方法
    def __init__(self, model: str = "gpt-4o", **kwargs):
        # 初始化 ChatOpenAI 类
        """
        初始化 ChatOpenAI

        Args:
            model: 模型名称，如 "gpt-4o"
            **kwargs: 其他参数（如 temperature, max_tokens 等）
        """
        # 设置模型名称
        self.model = model
        # 获取 api_key，优先从参数获取，否则从环境变量获取
        self.api_key = kwargs.get("api_key") or os.getenv("OPENAI_API_KEY")
        # 如果没有提供 api_key，则抛出异常
        if not self.api_key:
            raise ValueError("需要提供 api_key 或设置 OPENAI_API_KEY 环境变量")
        # 保存除 api_key 之外的其他参数，用于 API 调用
        self.model_kwargs = {k: v for k, v in kwargs.items() if k != "api_key"}
        # 创建 OpenAI 客户端实例
        self.client = openai.OpenAI(api_key=self.api_key)

    # 调用模型生成回复的方法
    def invoke(self, input, **kwargs):
        # 调用模型生成回复
        """
        调用模型生成回复

        Args:
            input: 输入内容，可以是字符串或消息列表
            **kwargs: 额外的 API 参数

        Returns:
            AIMessage: AI 的回复消息
        """
        # 将输入数据转换为消息格式
        messages = self._convert_input(input)
        # 构建 API 请求参数字典
        params = {
            "model": self.model,
            "messages": messages,
            **self.model_kwargs,
            **kwargs
        }
        # 使用 OpenAI 客户端发起 chat.completions.create 调用获取回复
        response = self.client.chat.completions.create(**params)
        # 取出返回结果中的第一个选项
        choice = response.choices[0]
        # 获取消息内容
        content = choice.message.content or ""
        # 返回一个 AIMessage 对象
        return AIMessage(content=content)

    # 流式调用模型生成回复的方法
    def stream(self, input, **kwargs):
        # 流式调用模型生成回复
        """
        流式调用模型生成回复

        Args:
            input: 输入内容，可以是字符串或消息列表
            **kwargs: 额外的 API 参数

        Yields:
            AIMessage: AI 的回复消息块（每次产生部分内容）
        """
        # 将输入数据转换为消息格式
        messages = self._convert_input(input)
        # 构建 API 请求参数字典，启用流式输出
        params = {
            "model": self.model,
            "messages": messages,
            "stream": True,  # 启用流式输出
            **self.model_kwargs,
            **kwargs
        }
        # 使用 OpenAI 客户端发起流式调用
        stream = self.client.chat.completions.create(**params)
        # 迭代流式响应
        for chunk in stream:
            # 检查是否有内容增量
            if chunk.choices and len(chunk.choices) > 0:
                delta = chunk.choices[0].delta
                # 检查 delta 中是否有 content，如果有则发送
                if hasattr(delta, 'content') and delta.content:
                    # 产生包含部分内容的 AIMessage
                    yield AIMessage(content=delta.content)

    # 内部方法，将输入转换为 OpenAI API 需要的消息格式
    def _convert_input(self, input):
        # 将输入转换为 OpenAI API 需要的消息格式
        """
        将输入转换为 OpenAI API 需要的消息格式

        Args:
            input: 字符串、消息列表或 ChatPromptValue

        Returns:
            list[dict]: OpenAI API 格式的消息列表
        """
        # 输入为字符串时，直接封装为用户角色消息
        if isinstance(input, str):
            return [{"role": "user", "content": input}]
        else:
            # 其他输入类型，转为字符串作为 user 消息
            return [{"role": "user", "content": str(input)}]
+
+# 定义与 DeepSeek 聊天模型交互的类
+class ChatDeepSeek:
+   # 初始化方法
+   # model: 模型名称，默认为 "deepseek-chat"
+   # **kwargs: 其他可选参数（如 temperature, max_tokens 等）
+   def __init__(self, model: str = "deepseek-chat", **kwargs):
+       """
+       初始化 ChatDeepSeek
+       
+       Args:
+           model: 模型名称，如 "deepseek-chat"
+           **kwargs: 其他参数（如 temperature, max_tokens 等）
+       """
+       # 设置模型名称
+       self.model = model
+       # 获取 api_key，优先从参数获取，否则从环境变量获取
+       self.api_key = kwargs.get("api_key") or os.getenv("DEEPSEEK_API_KEY")
+       # 如果没有提供 api_key，则抛出异常
+       if not self.api_key:
+           raise ValueError("需要提供 api_key 或设置 DEEPSEEK_API_KEY 环境变量")
+       # 保存除 api_key 之外的其他参数，用于 API 调用
+       self.model_kwargs = {k: v for k, v in kwargs.items() if k != "api_key"}
+       # 获取 DeepSeek 的 base_url，默认为官方地址
+       base_url = kwargs.get("base_url", "https://api.deepseek.com/v1")
+       # 创建 OpenAI 兼容的客户端实例（DeepSeek 使用 OpenAI 兼容的 API）
+       self.client = openai.OpenAI(api_key=self.api_key, base_url=base_url)
+   
+   # 调用模型生成回复的方法
+   # input: 输入内容，可以是字符串或消息列表
+   # **kwargs: 额外的 API 参数
+   def invoke(self, input, **kwargs):
+       """
+       调用模型生成回复
+       
+       Args:
+           input: 输入内容，可以是字符串或消息列表
+           **kwargs: 额外的 API 参数
+       
+       Returns:
+           AIMessage: AI 的回复消息
+       """
+       # 将输入数据转换为消息格式
+       messages = self._convert_input(input)
+       # 构建 API 请求参数字典
+       params = {
+           "model": self.model,
+           "messages": messages,
+           **self.model_kwargs,
+           **kwargs
+       }
+       # 使用 OpenAI 兼容的客户端发起 chat.completions.create 调用获取回复
+       response = self.client.chat.completions.create(**params)
+       # 取出返回结果中的第一个选项
+       choice = response.choices[0]
+       # 获取消息内容
+       content = choice.message.content or ""
+       # 返回一个 AIMessage 对象
+       return AIMessage(content=content)
+   
+   # 内部方法，将输入转换为 API 需要的消息格式
+   # input: 字符串、消息列表或 ChatPromptValue
+   def _convert_input(self, input):
+       """
+       将输入转换为 API 需要的消息格式
+       
+       Args:
+           input: 字符串、消息列表或 ChatPromptValue
+       
+       Returns:
+           list[dict]: API 格式的消息列表
+       """
+       # 如果输入是字符串，直接作为用户消息
+       if isinstance(input, str):
+           return [{"role": "user", "content": input}]    
+       else:
+           # 其他输入类型，转为字符串作为 user 消息
+           return [{"role": "user", "content": str(input)}]

3. ChatTongyi #

• 通过使用 langchain_community 库的ChatTongyi类，可以方便地调用“qwen-max”等通义千问系列模型，实现与模型的多轮对话交互。
• 你只需提供模型名称和 API 密钥，即可完成模型实例化。随后，通过调用 invoke 方法向大模型发送用户消息，便可获得智能回复，整个过程简单高效，适合快速集成和原型开发。

3.1. 3.ChatTongyi.py #

3.ChatTongyi.py

# 安装所需的依赖包：langchain_community 和 dashscope，可以通过终端执行该命令
# uv add langchain_community dashscope

# 导入 ChatTongyi 类，用于与通义千问大模型交互
from langchain import ChatTongyi

# 创建一个 ChatTongyi 实例，指定模型为 "qwen-max"，并传入相应的 API 密钥
llm = ChatTongyi(model="qwen-max", api_key="sk-cc2054c29cf54fec92503bf7016cf383")

# 调用实例的 invoke 方法，向模型发送用户消息 "你是谁"，并获取模型回复的结果
result = llm.invoke("你是谁")

# 打印模型回复的内容到控制台
print(result.content)

3.2. chat_models.py #

smartchain/chat_models.py

# 导入操作系统相关模块
import os

# 导入 openai 模块
import openai

# 从 .messages 模块导入 AIMessage、HumanMessage 和 SystemMessage 类
from .messages import AIMessage, HumanMessage, SystemMessage

# 定义与 OpenAI 聊天模型交互的类
class ChatOpenAI:

    # 初始化方法
    def __init__(self, model: str = "gpt-4o", **kwargs):
        # 初始化 ChatOpenAI 类
        """
        初始化 ChatOpenAI

        Args:
            model: 模型名称，如 "gpt-4o"
            **kwargs: 其他参数（如 temperature, max_tokens 等）
        """
        # 设置模型名称
        self.model = model
        # 获取 api_key，优先从参数获取，否则从环境变量获取
        self.api_key = kwargs.get("api_key") or os.getenv("OPENAI_API_KEY")
        # 如果没有提供 api_key，则抛出异常
        if not self.api_key:
            raise ValueError("需要提供 api_key 或设置 OPENAI_API_KEY 环境变量")
        # 保存除 api_key 之外的其他参数，用于 API 调用
        self.model_kwargs = {k: v for k, v in kwargs.items() if k != "api_key"}
        # 创建 OpenAI 客户端实例
        self.client = openai.OpenAI(api_key=self.api_key)

    # 调用模型生成回复的方法
    def invoke(self, input, **kwargs):
        # 调用模型生成回复
        """
        调用模型生成回复

        Args:
            input: 输入内容，可以是字符串或消息列表
            **kwargs: 额外的 API 参数

        Returns:
            AIMessage: AI 的回复消息
        """
        # 将输入数据转换为消息格式
        messages = self._convert_input(input)
        # 构建 API 请求参数字典
        params = {
            "model": self.model,
            "messages": messages,
            **self.model_kwargs,
            **kwargs
        }
        # 使用 OpenAI 客户端发起 chat.completions.create 调用获取回复
        response = self.client.chat.completions.create(**params)
        # 取出返回结果中的第一个选项
        choice = response.choices[0]
        # 获取消息内容
        content = choice.message.content or ""
        # 返回一个 AIMessage 对象
        return AIMessage(content=content)

    # 流式调用模型生成回复的方法
    def stream(self, input, **kwargs):
        # 流式调用模型生成回复
        """
        流式调用模型生成回复

        Args:
            input: 输入内容，可以是字符串或消息列表
            **kwargs: 额外的 API 参数

        Yields:
            AIMessage: AI 的回复消息块（每次产生部分内容）
        """
        # 将输入数据转换为消息格式
        messages = self._convert_input(input)
        # 构建 API 请求参数字典，启用流式输出
        params = {
            "model": self.model,
            "messages": messages,
            "stream": True,  # 启用流式输出
            **self.model_kwargs,
            **kwargs
        }
        # 使用 OpenAI 客户端发起流式调用
        stream = self.client.chat.completions.create(**params)
        # 迭代流式响应
        for chunk in stream:
            # 检查是否有内容增量
            if chunk.choices and len(chunk.choices) > 0:
                delta = chunk.choices[0].delta
                # 检查 delta 中是否有 content，如果有则发送
                if hasattr(delta, 'content') and delta.content:
                    # 产生包含部分内容的 AIMessage
                    yield AIMessage(content=delta.content)

    # 内部方法，将输入转换为 OpenAI API 需要的消息格式
    def _convert_input(self, input):
        # 将输入转换为 OpenAI API 需要的消息格式
        """
        将输入转换为 OpenAI API 需要的消息格式

        Args:
            input: 字符串、消息列表或 ChatPromptValue

        Returns:
            list[dict]: OpenAI API 格式的消息列表
        """
        # 输入为字符串时，直接封装为用户角色消息
        if isinstance(input, str):
            return [{"role": "user", "content": input}]
        else:
            # 其他输入类型，转为字符串作为 user 消息
            return [{"role": "user", "content": str(input)}]

# 定义与 DeepSeek 聊天模型交互的类
class ChatDeepSeek:
    # 初始化方法
    # model: 模型名称，默认为 "deepseek-chat"
    # **kwargs: 其他可选参数（如 temperature, max_tokens 等）
    def __init__(self, model: str = "deepseek-chat", **kwargs):
        """
        初始化 ChatDeepSeek

        Args:
            model: 模型名称，如 "deepseek-chat"
            **kwargs: 其他参数（如 temperature, max_tokens 等）
        """
        # 设置模型名称
        self.model = model
        # 获取 api_key，优先从参数获取，否则从环境变量获取
        self.api_key = kwargs.get("api_key") or os.getenv("DEEPSEEK_API_KEY")
        # 如果没有提供 api_key，则抛出异常
        if not self.api_key:
            raise ValueError("需要提供 api_key 或设置 DEEPSEEK_API_KEY 环境变量")
        # 保存除 api_key 之外的其他参数，用于 API 调用
        self.model_kwargs = {k: v for k, v in kwargs.items() if k != "api_key"}
        # 获取 DeepSeek 的 base_url，默认为官方地址
        base_url = kwargs.get("base_url", "https://api.deepseek.com/v1")
        # 创建 OpenAI 兼容的客户端实例（DeepSeek 使用 OpenAI 兼容的 API）
        self.client = openai.OpenAI(api_key=self.api_key, base_url=base_url)

    # 调用模型生成回复的方法
    # input: 输入内容，可以是字符串或消息列表
    # **kwargs: 额外的 API 参数
    def invoke(self, input, **kwargs):
        """
        调用模型生成回复

        Args:
            input: 输入内容，可以是字符串或消息列表
            **kwargs: 额外的 API 参数

        Returns:
            AIMessage: AI 的回复消息
        """
        # 将输入数据转换为消息格式
        messages = self._convert_input(input)
        # 构建 API 请求参数字典
        params = {
            "model": self.model,
            "messages": messages,
            **self.model_kwargs,
            **kwargs
        }
        # 使用 OpenAI 兼容的客户端发起 chat.completions.create 调用获取回复
        response = self.client.chat.completions.create(**params)
        # 取出返回结果中的第一个选项
        choice = response.choices[0]
        # 获取消息内容
        content = choice.message.content or ""
        # 返回一个 AIMessage 对象
        return AIMessage(content=content)

    # 内部方法，将输入转换为 API 需要的消息格式
    # input: 字符串、消息列表或 ChatPromptValue
    def _convert_input(self, input):
        """
        将输入转换为 API 需要的消息格式

        Args:
            input: 字符串、消息列表或 ChatPromptValue

        Returns:
            list[dict]: API 格式的消息列表
        """
        # 如果输入是字符串，直接作为用户消息
        if isinstance(input, str):
            return [{"role": "user", "content": input}]    
        else:
            # 其他输入类型，转为字符串作为 user 消息
            return [{"role": "user", "content": str(input)}]
+
+# 定义与通义千问（Tongyi）聊天模型交互的类
+class ChatTongyi:
+   
+   # 初始化方法
+   # 初始化方法，设置模型名称和 API 相关参数
+   def __init__(self, model: str = "qwen-max", **kwargs):
+       """
+       初始化 ChatTongyi
+       
+       Args:
+           model: 模型名称，如 "qwen-max"
+           **kwargs: 其他参数（如 temperature, max_tokens 等）
+       """
+       # 设置模型名称
+       self.model = model
+       # 获取 api_key，优先从参数获取，否则从环境变量获取
+       self.api_key = kwargs.get("api_key") or os.getenv("DASHSCOPE_API_KEY")
+       # 如果没有提供 api_key，则抛出异常
+       if not self.api_key:
+           raise ValueError("需要提供 api_key 或设置 DASHSCOPE_API_KEY 环境变量")
+       # 保存除 api_key 之外的其他参数，用于 API 调用
+       self.model_kwargs = {k: v for k, v in kwargs.items() if k != "api_key"}
+       # 获取通义千问的 API base URL（使用 OpenAI 兼容模式），如果未指定则使用默认值
+       base_url = kwargs.get("base_url", "https://dashscope.aliyuncs.com/compatible-mode/v1")
+       # 创建 OpenAI 兼容的客户端实例（通义千问使用 OpenAI 兼容的 API）
+       self.client = openai.OpenAI(api_key=self.api_key, base_url=base_url)
+   
+   # 调用模型生成回复的方法
+   # 调用模型生成回复，返回 AIMessage 对象
+   def invoke(self, input, **kwargs):
+       """
+       调用模型生成回复
+       
+       Args:
+           input: 输入内容，可以是字符串或消息列表
+           **kwargs: 额外的 API 参数
+       
+       Returns:
+           AIMessage: AI 的回复消息
+       """
+       # 将输入数据转换为消息格式
+       messages = self._convert_input(input)
+       # 构建 API 请求参数字典，包含模型名、消息内容和其他参数
+       params = {
+           "model": self.model,
+           "messages": messages,
+           **self.model_kwargs,
+           **kwargs
+       }
+       # 使用 OpenAI 兼容的客户端发起 chat.completions.create 调用以获取回复
+       response = self.client.chat.completions.create(**params)
+       # 取出返回结果中的第一个回复选项
+       choice = response.choices[0]
+       # 获取回复的消息内容，如果内容不存在则返回空字符串
+       content = choice.message.content or ""
+       # 构建并返回一个 AIMessage 对象
+       return AIMessage(content=content)
+   
+   # 内部方法，将输入转换为 API 需要的消息格式
+   # 支持字符串、消息列表等输入，统一包装为 OpenAI API 格式
+   def _convert_input(self, input):
+       """
+       将输入转换为 API 需要的消息格式
+       
+       Args:
+           input: 字符串、消息列表或 ChatPromptValue
+       
+       Returns:
+           list[dict]: API 格式的消息列表
+       """
+       # 如果输入是字符串，直接包装为“用户”角色的消息
+       if isinstance(input, str):
+           return [{"role": "user", "content": input}]
+       else:
+           # 其他输入类型，转换为字符串作为“用户”消息内容
+           return [{"role": "user", "content": str(input)}]            

4. 多轮对话 #

• 消息

• 通过自定义消息类型（如 HumanMessage, AIMessage, SystemMessage）和自定义 ChatOpenAI 类，实现多轮对话消息的组织与调用。

• 在聊天过程中，通过构建消息列表，可以模拟真实多轮交互，让大模型具备"记忆"上下文的能力。这样模型能够理解当前会话的内容，并基于历史消息做出更合适的回复。
• 如何创建消息对象，并将它们传递给自定义的 ChatOpenAI 实例，让其统一格式化为 OpenAI API 所需的消息格式，实现灵活的上下文管理和消息流的组合。这种用法不仅提高了代码的可读性，也便于扩展自定义消息类型与消息处理逻辑。
• 你可以通过灵活地传递多条不同类型的 message（比如系统指令、人类消息、AI 消息等），实现如上下文记忆、角色扮演、多轮交互等丰富的对话场景。在多轮对话开发、Prompt 工程和复杂业务逻辑中，这种模式极其有用。

4.1. 4.Message.py #

4.Message.py

# 导入 ChatOpenAI 类
#from langchain_openai import ChatOpenAI
# 导入消息类型：AIMessage、HumanMessage、SystemMessage
#from langchain.messages import AIMessage, HumanMessage, SystemMessage


from smartchain.chat_models import ChatOpenAI
from smartchain.messages import AIMessage, HumanMessage, SystemMessage


# 创建一个 ChatOpenAI 实例，指定模型为 "gpt-4o"
llm = ChatOpenAI(model="gpt-4o")

# 构建消息列表，包括系统消息、用户消息和 AI 消息
messages = [
    # 系统消息，设定 AI 助手身份
    SystemMessage(content="你是一个AI助手，请回答用户的问题。"),
    # 用户消息（字符串方式）
    "你好，我叫张三，你是谁？",
    # AI 消息（AIMessage对象方式）
    AIMessage(content="我是GPT-4o，一个AI助手。"),
    # 用户消息（字典方式）
    {"role": "user", "content": "你知道我叫什么吗？"},
    # AI 消息（元组方式）
    ("assistant", "你的名字是张三。"),
]

# 调用 llm.invoke，将消息传入模型，获得 AI 回复
result = llm.invoke(messages)

# 输出 AI 回复内容
print(result,type(result))

4.2. messages.py #

smartchain/messages.py

# 定义基础消息类
class BaseMessage:
    # 基础消息类的文档字符串
    """基础消息类"""

    # 初始化方法，content为消息内容，其余为可选参数
    def __init__(self, content: str, **kwargs):
        # 初始化消息
        # content: 消息内容
        # **kwargs: 其他可选参数
        self.content = content  # 保存消息内容
        self.type = kwargs.get("type", "base")  # 获取消息类型，默认为"base"
        for key, value in kwargs.items():  # 遍历所有其他参数
            if key != "type":  # 排除type参数
                setattr(self, key, value)  # 设置为成员变量

    # 定义当对象被str()或print时的输出内容
    def __str__(self):
        return self.content  # 返回消息内容

    # 定义对象的官方字符串表示，用于debug
    def __repr__(self):
        return f"{self.__class__.__name__}(content={self.content!r})"  # 返回类名和内容

# 定义用户消息类，继承自BaseMessage
class HumanMessage(BaseMessage):
    # 用户消息类的文档字符串
    """用户消息"""

    # 初始化方法，调用父类构造方法，并指定type为"human"
    def __init__(self, content: str, **kwargs):
        # 调用父类构造方法，并固定type为"human"
        super().__init__(content, type="human", **kwargs)

# 定义AI消息类，继承自BaseMessage
class AIMessage(BaseMessage):
    # AI消息类的文档字符串
    """AI 消息"""

    # 初始化方法，调用父类构造方法，并指定type为"ai"
    def __init__(self, content: str, **kwargs):
        # 调用父类构造方法，并固定type为"ai"
        super().__init__(content, type="ai", **kwargs)

+# 定义一个继承自BaseMessage的系统消息类
+class SystemMessage(BaseMessage):
+   # 说明这是系统消息的类
+   """系统消息"""
+   
+   # 初始化方法，content是消息内容，**kwargs为其他可选参数
+   def __init__(self, content: str, **kwargs):
+       # 调用父类BaseMessage的初始化方法，并将type参数固定为"system"
+       super().__init__(content, type="system", **kwargs)

4.3. chat_models.py #

smartchain/chat_models.py

# 导入操作系统相关模块
import os

# 导入 openai 模块
import openai

# 从 .messages 模块导入 AIMessage、HumanMessage 和 SystemMessage 类
from .messages import AIMessage, HumanMessage, SystemMessage

# 定义与 OpenAI 聊天模型交互的类
class ChatOpenAI:

    # 初始化方法
    def __init__(self, model: str = "gpt-4o", **kwargs):
        # 初始化 ChatOpenAI 类
        """
        初始化 ChatOpenAI

        Args:
            model: 模型名称，如 "gpt-4o"
            **kwargs: 其他参数（如 temperature, max_tokens 等）
        """
        # 设置模型名称
        self.model = model
        # 获取 api_key，优先从参数获取，否则从环境变量获取
        self.api_key = kwargs.get("api_key") or os.getenv("OPENAI_API_KEY")
        # 如果没有提供 api_key，则抛出异常
        if not self.api_key:
            raise ValueError("需要提供 api_key 或设置 OPENAI_API_KEY 环境变量")
        # 保存除 api_key 之外的其他参数，用于 API 调用
        self.model_kwargs = {k: v for k, v in kwargs.items() if k != "api_key"}
        # 创建 OpenAI 客户端实例
        self.client = openai.OpenAI(api_key=self.api_key)

    # 调用模型生成回复的方法
    def invoke(self, input, **kwargs):
        # 调用模型生成回复
        """
        调用模型生成回复

        Args:
            input: 输入内容，可以是字符串或消息列表
            **kwargs: 额外的 API 参数

        Returns:
            AIMessage: AI 的回复消息
        """
        # 将输入数据转换为消息格式
        messages = self._convert_input(input)
        # 构建 API 请求参数字典
        params = {
            "model": self.model,
            "messages": messages,
            **self.model_kwargs,
            **kwargs
        }
        # 使用 OpenAI 客户端发起 chat.completions.create 调用获取回复
        response = self.client.chat.completions.create(**params)
        # 取出返回结果中的第一个选项
        choice = response.choices[0]
        # 获取消息内容
        content = choice.message.content or ""
        # 返回一个 AIMessage 对象
        return AIMessage(content=content)

    # 流式调用模型生成回复的方法
    def stream(self, input, **kwargs):
        # 流式调用模型生成回复
        """
        流式调用模型生成回复

        Args:
            input: 输入内容，可以是字符串或消息列表
            **kwargs: 额外的 API 参数

        Yields:
            AIMessage: AI 的回复消息块（每次产生部分内容）
        """
        # 将输入数据转换为消息格式
        messages = self._convert_input(input)
        # 构建 API 请求参数字典，启用流式输出
        params = {
            "model": self.model,
            "messages": messages,
            "stream": True,  # 启用流式输出
            **self.model_kwargs,
            **kwargs
        }
        # 使用 OpenAI 客户端发起流式调用
        stream = self.client.chat.completions.create(**params)
        # 迭代流式响应
        for chunk in stream:
            # 检查是否有内容增量
            if chunk.choices and len(chunk.choices) > 0:
                delta = chunk.choices[0].delta
                # 检查 delta 中是否有 content，如果有则发送
                if hasattr(delta, 'content') and delta.content:
                    # 产生包含部分内容的 AIMessage
                    yield AIMessage(content=delta.content)

    # 内部方法，将输入转换为 OpenAI API 需要的消息格式
    def _convert_input(self, input):
        # 将输入转换为 OpenAI API 需要的消息格式
        """
        将输入转换为 OpenAI API 需要的消息格式

        Args:
            input: 字符串、消息列表或 ChatPromptValue

        Returns:
            list[dict]: OpenAI API 格式的消息列表
        """
        # 输入为字符串时，直接封装为用户角色消息
        if isinstance(input, str):
            return [{"role": "user", "content": input}]
+       # 如果输入是列表类型
+       elif isinstance(input, list):
+           # 新建一个空的消息列表
+           messages = []
+           # 遍历输入列表中的每一个元素
+           for msg in input:
+               # 判断是否为字符串，是则作为用户消息加入
+               if isinstance(msg, str):
+                   messages.append({"role": "user", "content": msg})
+               # 判断是否为 HumanMessage、AIMessage 或 SystemMessage 实例
+               elif isinstance(msg, (HumanMessage, AIMessage, SystemMessage)):
+                   # 如果是 HumanMessage，将角色设为 user
+                   if isinstance(msg, HumanMessage):
+                       role = "user"
+                   # 如果是 AIMessage，将角色设为 assistant
+                   elif isinstance(msg, AIMessage):
+                       role = "assistant"
+                   # 如果是 SystemMessage，将角色设为 system
+                   elif isinstance(msg, SystemMessage):
+                       role = "system"
+                   # 获取消息内容（有 content 属性则取 content，否则转为字符串）
+                   content = msg.content if hasattr(msg, "content") else str(msg)
+                   # 将角色和内容添加到消息列表
+                   messages.append({"role": role, "content": content})
+               # 如果元素本身为字典，直接添加进消息列表
+               elif isinstance(msg, dict):
+                   # 直接添加字典类型的消息
+                   messages.append(msg)
+               # 如果元素为长度为 2 的元组，将其解包为 role 和 content
+               elif isinstance(msg, tuple) and len(msg) == 2:
+                   # 将元组解包为 role 和 content
+                   role, content = msg
+                   # 将角色和内容添加到消息列表
+                   messages.append({"role": role, "content": content})
+           # 返回构建好的消息列表
+           return messages    
        else:
            # 其他输入类型，转为字符串作为 user 消息
            return [{"role": "user", "content": str(input)}]

# 定义与 DeepSeek 聊天模型交互的类
class ChatDeepSeek:
    # 初始化方法
    # model: 模型名称，默认为 "deepseek-chat"
    # **kwargs: 其他可选参数（如 temperature, max_tokens 等）
    def __init__(self, model: str = "deepseek-chat", **kwargs):
        """
        初始化 ChatDeepSeek

        Args:
            model: 模型名称，如 "deepseek-chat"
            **kwargs: 其他参数（如 temperature, max_tokens 等）
        """
        # 设置模型名称
        self.model = model
        # 获取 api_key，优先从参数获取，否则从环境变量获取
        self.api_key = kwargs.get("api_key") or os.getenv("DEEPSEEK_API_KEY")
        # 如果没有提供 api_key，则抛出异常
        if not self.api_key:
            raise ValueError("需要提供 api_key 或设置 DEEPSEEK_API_KEY 环境变量")
        # 保存除 api_key 之外的其他参数，用于 API 调用
        self.model_kwargs = {k: v for k, v in kwargs.items() if k != "api_key"}
        # 获取 DeepSeek 的 base_url，默认为官方地址
        base_url = kwargs.get("base_url", "https://api.deepseek.com/v1")
        # 创建 OpenAI 兼容的客户端实例（DeepSeek 使用 OpenAI 兼容的 API）
        self.client = openai.OpenAI(api_key=self.api_key, base_url=base_url)

    # 调用模型生成回复的方法
    # input: 输入内容，可以是字符串或消息列表
    # **kwargs: 额外的 API 参数
    def invoke(self, input, **kwargs):
        """
        调用模型生成回复

        Args:
            input: 输入内容，可以是字符串或消息列表
            **kwargs: 额外的 API 参数

        Returns:
            AIMessage: AI 的回复消息
        """
        # 将输入数据转换为消息格式
        messages = self._convert_input(input)
        # 构建 API 请求参数字典
        params = {
            "model": self.model,
            "messages": messages,
            **self.model_kwargs,
            **kwargs
        }
        # 使用 OpenAI 兼容的客户端发起 chat.completions.create 调用获取回复
        response = self.client.chat.completions.create(**params)
        # 取出返回结果中的第一个选项
        choice = response.choices[0]
        # 获取消息内容
        content = choice.message.content or ""
        # 返回一个 AIMessage 对象
        return AIMessage(content=content)

    # 内部方法，将输入转换为 API 需要的消息格式
    # input: 字符串、消息列表或 ChatPromptValue
    def _convert_input(self, input):
        """
        将输入转换为 API 需要的消息格式

        Args:
            input: 字符串、消息列表或 ChatPromptValue

        Returns:
            list[dict]: API 格式的消息列表
        """
        # 如果输入是字符串，直接作为用户消息
        if isinstance(input, str):
            return [{"role": "user", "content": input}]    
        else:
            # 其他输入类型，转为字符串作为 user 消息
            return [{"role": "user", "content": str(input)}]

# 定义与通义千问（Tongyi）聊天模型交互的类
class ChatTongyi:

    # 初始化方法
    # 初始化方法，设置模型名称和 API 相关参数
    def __init__(self, model: str = "qwen-max", **kwargs):
        """
        初始化 ChatTongyi

        Args:
            model: 模型名称，如 "qwen-max"
            **kwargs: 其他参数（如 temperature, max_tokens 等）
        """
        # 设置模型名称
        self.model = model
        # 获取 api_key，优先从参数获取，否则从环境变量获取
        self.api_key = kwargs.get("api_key") or os.getenv("DASHSCOPE_API_KEY")
        # 如果没有提供 api_key，则抛出异常
        if not self.api_key:
            raise ValueError("需要提供 api_key 或设置 DASHSCOPE_API_KEY 环境变量")
        # 保存除 api_key 之外的其他参数，用于 API 调用
        self.model_kwargs = {k: v for k, v in kwargs.items() if k != "api_key"}
        # 获取通义千问的 API base URL（使用 OpenAI 兼容模式），如果未指定则使用默认值
        base_url = kwargs.get("base_url", "https://dashscope.aliyuncs.com/compatible-mode/v1")
        # 创建 OpenAI 兼容的客户端实例（通义千问使用 OpenAI 兼容的 API）
        self.client = openai.OpenAI(api_key=self.api_key, base_url=base_url)

    # 调用模型生成回复的方法
    # 调用模型生成回复，返回 AIMessage 对象
    def invoke(self, input, **kwargs):
        """
        调用模型生成回复

        Args:
            input: 输入内容，可以是字符串或消息列表
            **kwargs: 额外的 API 参数

        Returns:
            AIMessage: AI 的回复消息
        """
        # 将输入数据转换为消息格式
        messages = self._convert_input(input)
        # 构建 API 请求参数字典，包含模型名、消息内容和其他参数
        params = {
            "model": self.model,
            "messages": messages,
            **self.model_kwargs,
            **kwargs
        }
        # 使用 OpenAI 兼容的客户端发起 chat.completions.create 调用以获取回复
        response = self.client.chat.completions.create(**params)
        # 取出返回结果中的第一个回复选项
        choice = response.choices[0]
        # 获取回复的消息内容，如果内容不存在则返回空字符串
        content = choice.message.content or ""
        # 构建并返回一个 AIMessage 对象
        return AIMessage(content=content)

    # 内部方法，将输入转换为 API 需要的消息格式
    # 支持字符串、消息列表等输入，统一包装为 OpenAI API 格式
    def _convert_input(self, input):
        """
        将输入转换为 API 需要的消息格式

        Args:
            input: 字符串、消息列表或 ChatPromptValue

        Returns:
            list[dict]: API 格式的消息列表
        """
        # 如果输入是字符串，直接包装为“用户”角色的消息
        if isinstance(input, str):
            return [{"role": "user", "content": input}]
        else:
            # 其他输入类型，转换为字符串作为“用户”消息内容
            return [{"role": "user", "content": str(input)}]            

4.4 类图 #

4.4.1 类图 #

类名	所属模块	主要功能	主要方法/属性
ChatOpenAI	smartchain.chat_models	封装与 OpenAI 聊天模型的交互，用于调用大语言模型生成回复	• __init__(model, **kwargs) - 初始化，指定模型名称 • invoke(input, **kwargs) - 调用模型生成回复，返回 AIMessage • model - 模型名称属性 • api_key - API 密钥属性 • _convert_input(input) - 私有方法，将输入转换为 API 需要的消息格式
SystemMessage	smartchain.messages	系统消息类，表示系统提示消息，用于设定 AI 助手的身份和行为	• __init__(content, **kwargs) - 初始化，type 固定为"system" • content - 消息内容属性 • type - 消息类型属性（值为"system"） • __str__() - 返回消息内容 • __repr__() - 返回对象的字符串表示
AIMessage	smartchain.messages	AI 消息类，表示 AI 助手的回复消息，也可用于多轮对话中的历史消息	• __init__(content, **kwargs) - 初始化，type 固定为"ai" • content - 消息内容属性 • type - 消息类型属性（值为"ai"） • __str__() - 返回消息内容 • __repr__() - 返回对象的字符串表示
BaseMessage	smartchain.messages	基础消息类，所有消息类型的基类	• __init__(content, **kwargs) - 初始化消息 • content - 消息内容属性 • type - 消息类型属性 • __str__() - 返回消息内容 • __repr__() - 返回对象的字符串表示
HumanMessage	smartchain.messages	用户消息类，表示人类用户发送的消息	• __init__(content, **kwargs) - 初始化，type 固定为"human" • 继承自 BaseMessage 的所有属性和方法

4.4.2 类关系图 #

4.4.3 调用关系图 #

4.4.4 数据流转过程 #

1. 初始化阶段

   • 创建 ChatOpenAI 实例，指定模型名称
   • 创建 SystemMessage 和 AIMessage 对象

2. 消息构建阶段

   • 构建消息列表，支持多种格式：
     • SystemMessage 对象
     • 字符串（自动转为用户消息）
     • AIMessage 对象
     • 字典格式 {"role": "user", "content": "..."}
     • 元组格式 ("assistant", "...")

3. 消息转换阶段

   • ChatOpenAI.invoke() 调用 _convert_input() 方法
   • 将各种格式的消息统一转换为 OpenAI API 需要的格式：
     • SystemMessage → {"role": "system", "content": "..."}
     • AIMessage → {"role": "assistant", "content": "..."}
     • 字符串 → {"role": "user", "content": "..."}
     • 字典 → 直接使用
     • 元组 → {"role": role, "content": content}

4. API 调用阶段

   • 调用 OpenAI API，传入格式化后的消息列表
   • 获取 API 响应

5. 结果返回阶段

   • 从 API 响应中提取内容
   • 创建 AIMessage 对象并返回
   • 主程序可以通过 result.content 获取回复内容