diff --git a/src/config/official_configs.py b/src/config/official_configs.py
index 013b3bd39..81ee5cacd 100644
--- a/src/config/official_configs.py
+++ b/src/config/official_configs.py
@@ -83,6 +83,164 @@ class ChatConfig(ValidatedConfigBase):
     proactive_thinking_interval: int = Field(default=1500, description="主动思考间隔")
     proactive_thinking_prompt_template: str = Field(default="", description="主动思考提示模板")
 
+    def get_current_talk_frequency(self, chat_stream_id: Optional[str] = None) -> float:
+        """
+        根据当前时间和聊天流获取对应的 talk_frequency
+
+        Args:
+            chat_stream_id: 聊天流ID，格式为 "platform:chat_id:type"
+
+        Returns:
+            float: 对应的频率值
+        """
+        if not self.talk_frequency_adjust:
+            return self.talk_frequency
+
+        # 优先检查聊天流特定的配置
+        if chat_stream_id:
+            stream_frequency = self._get_stream_specific_frequency(chat_stream_id)
+            if stream_frequency is not None:
+                return stream_frequency
+
+        # 检查全局时段配置（第一个元素为空字符串的配置）
+        global_frequency = self._get_global_frequency()
+        if global_frequency is not None:
+            return global_frequency
+
+        # 如果都没有匹配，返回默认值
+        return self.talk_frequency
+
+    def _get_time_based_frequency(self, time_freq_list: list[str]) -> Optional[float]:
+        """
+        根据时间配置列表获取当前时段的频率
+
+        Args:
+            time_freq_list: 时间频率配置列表，格式为 ["HH:MM,frequency", ...]
+
+        Returns:
+            float: 频率值，如果没有配置则返回 None
+        """
+        from datetime import datetime
+
+        current_time = datetime.now().strftime("%H:%M")
+        current_hour, current_minute = map(int, current_time.split(":"))
+        current_minutes = current_hour * 60 + current_minute
+
+        # 解析时间频率配置
+        time_freq_pairs = []
+        for time_freq_str in time_freq_list:
+            try:
+                time_str, freq_str = time_freq_str.split(",")
+                hour, minute = map(int, time_str.split(":"))
+                frequency = float(freq_str)
+                minutes = hour * 60 + minute
+                time_freq_pairs.append((minutes, frequency))
+            except (ValueError, IndexError):
+                continue
+
+        if not time_freq_pairs:
+            return None
+
+        # 按时间排序
+        time_freq_pairs.sort(key=lambda x: x[0])
+
+        # 查找当前时间对应的频率
+        current_frequency = None
+        for minutes, frequency in time_freq_pairs:
+            if current_minutes >= minutes:
+                current_frequency = frequency
+            else:
+                break
+
+        # 如果当前时间在所有配置时间之前，使用最后一个时间段的频率（跨天逻辑）
+        if current_frequency is None and time_freq_pairs:
+            current_frequency = time_freq_pairs[-1][1]
+
+        return current_frequency
+
+    def _get_stream_specific_frequency(self, chat_stream_id: str):
+        """
+        获取特定聊天流在当前时间的频率
+
+        Args:
+            chat_stream_id: 聊天流ID（哈希值）
+
+        Returns:
+            float: 频率值，如果没有配置则返回 None
+        """
+        # 查找匹配的聊天流配置
+        for config_item in self.talk_frequency_adjust:
+            if not config_item or len(config_item) < 2:
+                continue
+
+            stream_config_str = config_item[0]  # 例如 "qq:1026294844:group"
+
+            # 解析配置字符串并生成对应的 chat_id
+            config_chat_id = self._parse_stream_config_to_chat_id(stream_config_str)
+            if config_chat_id is None:
+                continue
+
+            # 比较生成的 chat_id
+            if config_chat_id != chat_stream_id:
+                continue
+
+            # 使用通用的时间频率解析方法
+            return self._get_time_based_frequency(config_item[1:])
+
+        return None
+
+    def _parse_stream_config_to_chat_id(self, stream_config_str: str) -> Optional[str]:
+        """
+        解析流配置字符串并生成对应的 chat_id
+
+        Args:
+            stream_config_str: 格式为 "platform:id:type" 的字符串
+
+        Returns:
+            str: 生成的 chat_id，如果解析失败则返回 None
+        """
+        try:
+            parts = stream_config_str.split(":")
+            if len(parts) != 3:
+                return None
+
+            platform = parts[0]
+            id_str = parts[1]
+            stream_type = parts[2]
+
+            # 判断是否为群聊
+            is_group = stream_type == "group"
+
+            # 使用与 ChatStream.get_stream_id 相同的逻辑生成 chat_id
+            import hashlib
+
+            if is_group:
+                components = [platform, str(id_str)]
+            else:
+                components = [platform, str(id_str), "private"]
+            key = "_".join(components)
+            return hashlib.md5(key.encode()).hexdigest()
+
+        except (ValueError, IndexError):
+            return None
+
+    def _get_global_frequency(self) -> Optional[float]:
+        """
+        获取全局默认频率配置
+
+        Returns:
+            float: 频率值，如果没有配置则返回 None
+        """
+        for config_item in self.talk_frequency_adjust:
+            if not config_item or len(config_item) < 2:
+                continue
+
+            # 检查是否为全局默认配置（第一个元素为空字符串）
+            if config_item[0] == "":
+                return self._get_time_based_frequency(config_item[1:])
+
+        return None
+
 
 
 class MessageReceiveConfig(ValidatedConfigBase):
@@ -106,6 +264,132 @@ class ExpressionConfig(ValidatedConfigBase):
     expression_learning: list[list] = Field(default_factory=lambda: [], description="表达学习")
     expression_groups: list[list[str]] = Field(default_factory=list, description="表达组")
 
+    def _parse_stream_config_to_chat_id(self, stream_config_str: str) -> Optional[str]:
+        """
+        解析流配置字符串并生成对应的 chat_id
+
+        Args:
+            stream_config_str: 格式为 "platform:id:type" 的字符串
+
+        Returns:
+            str: 生成的 chat_id，如果解析失败则返回 None
+        """
+        try:
+            parts = stream_config_str.split(":")
+            if len(parts) != 3:
+                return None
+
+            platform = parts[0]
+            id_str = parts[1]
+            stream_type = parts[2]
+
+            # 判断是否为群聊
+            is_group = stream_type == "group"
+
+            # 使用与 ChatStream.get_stream_id 相同的逻辑生成 chat_id
+            import hashlib
+
+            if is_group:
+                components = [platform, str(id_str)]
+            else:
+                components = [platform, str(id_str), "private"]
+            key = "_".join(components)
+            return hashlib.md5(key.encode()).hexdigest()
+
+        except (ValueError, IndexError):
+            return None
+
+    def get_expression_config_for_chat(self, chat_stream_id: Optional[str] = None) -> tuple[bool, bool, float]:
+        """
+        根据聊天流ID获取表达配置
+
+        Args:
+            chat_stream_id: 聊天流ID，格式为哈希值
+
+        Returns:
+            tuple: (是否使用表达, 是否学习表达, 学习间隔)
+        """
+        if not self.expression_learning:
+            # 如果没有配置，使用默认值：启用表达，启用学习，300秒间隔
+            return True, True, 300
+
+        # 优先检查聊天流特定的配置
+        if chat_stream_id:
+            specific_config = self._get_stream_specific_config(chat_stream_id)
+            if specific_config is not None:
+                return specific_config
+
+        # 检查全局配置（第一个元素为空字符串的配置）
+        global_config = self._get_global_config()
+        if global_config is not None:
+            return global_config
+
+        # 如果都没有匹配，返回默认值
+        return True, True, 300
+
+    def _get_stream_specific_config(self, chat_stream_id: str) -> Optional[tuple[bool, bool, float]]:
+        """
+        获取特定聊天流的表达配置
+
+        Args:
+            chat_stream_id: 聊天流ID（哈希值）
+
+        Returns:
+            tuple: (是否使用表达, 是否学习表达, 学习间隔)，如果没有配置则返回 None
+        """
+        for config_item in self.expression_learning:
+            if not config_item or len(config_item) < 4:
+                continue
+
+            stream_config_str = config_item[0]  # 例如 "qq:1026294844:group"
+
+            # 如果是空字符串，跳过（这是全局配置）
+            if stream_config_str == "":
+                continue
+
+            # 解析配置字符串并生成对应的 chat_id
+            config_chat_id = self._parse_stream_config_to_chat_id(stream_config_str)
+            if config_chat_id is None:
+                continue
+
+            # 比较生成的 chat_id
+            if config_chat_id != chat_stream_id:
+                continue
+
+            # 解析配置
+            try:
+                use_expression = config_item[1].lower() == "enable"
+                enable_learning = config_item[2].lower() == "enable"
+                learning_intensity = float(config_item[3])
+                return use_expression, enable_learning, learning_intensity
+            except (ValueError, IndexError):
+                continue
+
+        return None
+
+    def _get_global_config(self) -> Optional[tuple[bool, bool, float]]:
+        """
+        获取全局表达配置
+
+        Returns:
+            tuple: (是否使用表达, 是否学习表达, 学习间隔)，如果没有配置则返回 None
+        """
+        for config_item in self.expression_learning:
+            if not config_item or len(config_item) < 4:
+                continue
+
+            # 检查是否为全局配置（第一个元素为空字符串）
+            if config_item[0] == "":
+                try:
+                    use_expression = config_item[1].lower() == "enable"
+                    enable_learning = config_item[2].lower() == "enable"
+                    learning_intensity = float(config_item[3])
+                    return use_expression, enable_learning, learning_intensity
+                except (ValueError, IndexError):
+                    continue
+
+        return None
+
 
 
 class ToolConfig(ValidatedConfigBase):