Plugins developing problems

[eurekaHm]

Self Checks

• I have searched for existing issues search for existing issues, including closed ones.
• I confirm that I am using English to submit this report (我已阅读并同意 Language Policy).
• [FOR CHINESE USERS] 请务必使用英文提交 Issue，否则会被关闭。谢谢！:)
• Please do not modify this template :) and fill in all the required fields.

1. Is this request related to a challenge you're experiencing? Tell me about your story.

hello, I am developing my "file_process" plugins, which is used to split my document by titles and appendixs. The question is my documet can be processed suessfully but it would return something worng like 'Exception: Connection is closed'

2. Additional context or comments

my program：
from collections.abc import Generator
from typing import Any, List
from dify_plugin import Tool
from dify_plugin.entities.tool import ToolInvokeMessage
from io import BytesIO

from dify_plugin.file.file import File
from docx import Document
from docx.oxml.text.paragraph import CT_P
from docx.oxml.table import CT_Tbl
from docx.text.paragraph import Paragraph
from docx.table import Table
import tiktoken

class DocSplitterTool(Tool):

def _invoke(self, tool_parameters: dict[str, Any]) -> Generator[ToolInvokeMessage]:

    # yield self.create_text_message("Hello, this is a test message.")

    try:
        # 获取并验证文件参数
        doc_file = tool_parameters.get("doc_file")
        if not doc_file:
            yield self.create_text_message("请上传Word文档")
            return

        if doc_file.mime_type != "application/vnd.openxmlformats-officedocument.wordprocessingml.document":
            yield self.create_text_message("仅支持有效的.docx文件")
            return

        # 使用上下文管理器处理字节流
        with BytesIO(doc_file.blob) as doc_stream:
            sections = self.split_by_heading(doc_stream)

            print("OUTPUT:"+sections)
            print()
            # 返回结构化结果
            message = self.create_text_message(sections)

            print(f"MESSAGE: {message}")

            yield self.create_text_message("Hello, this is a test message.")
            return

    except Exception as e:
        print(f"捕获异常: {str(e)}")
        yield self.create_text_message(f"处理失败: {str(e)}")

def count_tokens(self, text: str) -> int:
    """计算文本的 tokens 数量"""
    encoding = tiktoken.get_encoding("cl100k_base")
    return len(encoding.encode(text))

def split_content_by_tokens(self, content: str, max_tokens: int = 4000) -> List[str]:
    """按 tokens 数量分割内容"""
    if self.count_tokens(content) <= max_tokens:
        return [content]

    parts = []
    current_part = []
    current_tokens = 0

    paragraphs = content.split('\n')
    for para in paragraphs:
        para_tokens = self.count_tokens(para)

        if para_tokens > max_tokens:
            if current_part:
                parts.append('\n'.join(current_part))
                current_part = []
                current_tokens = 0

            sentences = para.split('。')
            current_sentence = []
            for sent in sentences:
                sent_tokens = self.count_tokens(sent)
                if current_tokens + sent_tokens > max_tokens:
                    if current_sentence:
                        parts.append('。'.join(current_sentence) + '。')
                        current_sentence = []
                        current_tokens = 0
                current_sentence.append(sent)
                current_tokens += sent_tokens

            if current_sentence:
                parts.append('。'.join(current_sentence) + '。')
        else:
            if current_tokens + para_tokens > max_tokens:
                parts.append('\n'.join(current_part))
                current_part = [para]
                current_tokens = para_tokens
            else:
                current_part.append(para)
                current_tokens += para_tokens

    if current_part:
        parts.append('\n'.join(current_part))

    return parts

def split_by_heading(self, doc_stream: BytesIO) -> str:
    """核心处理逻辑：按标题分割文档"""
    doc = Document(doc_stream)
    sections = []
    current_title = None
    last_valid_title = None
    content = []
    titles = []

    for element in doc._element.body.iterchildren():
        if isinstance(element, CT_P):
            para = Paragraph(element, doc)
            if para.style.name.startswith(('Heading', '附录')):
                if current_title is not None:
                    self._finalize_section(sections, content, current_title, titles)
                    content = []

                new_title = para.text.strip()
                if new_title:
                    current_title = new_title
                    last_valid_title = new_title
                else:
                    current_title = last_valid_title
            else:
                if current_title is not None:
                    content.append(para.text)
        elif isinstance(element, CT_Tbl):
            table = self._process_table(element, doc)
            if current_title is not None and table:
                content.append(table)

    if current_title is not None:
        self._finalize_section(sections, content, current_title, titles)

    result_str = "\""
    for section in sections:
        # 将字典转换为字符串表示
        section_str = str(section)
        # 替换单引号为双引号以符合示例格式
        # section_str = section_str.replace("'", '"')
        result_str += section_str

    result_str += "\""
    return result_str

    # return sections

def _process_table(self, element, doc) -> str:
    """处理表格为 Markdown 格式"""
    table = Table(element, doc)
    md_rows = []
    for i, row in enumerate(table.rows):
        cells = [cell.text.strip() for cell in row.cells]
        if not any(cells):
            continue
        md_row = "| " + " | ".join(cells) + " |"
        md_rows.append(md_row)

        if i == 0 and md_rows:
            separator = "| " + " | ".join(["---"] * len(cells)) + " |"
            md_rows.append(separator)

    return "\n".join(md_rows) if md_rows else ""

def _finalize_section(self, sections, content, current_title, titles):
    """完成当前章节的处理"""
    section_content = "\n".join(content)
    if section_content.strip() and self.count_tokens(section_content) >= 50:
        content_parts = self.split_content_by_tokens(section_content)
        if len(content_parts) > 1:
            for i, part in enumerate(content_parts, 1):
                new_title = f"{current_title}{i}"
                sections.append({"title": new_title, "content": part})
                titles.append(new_title)
        else:
            sections.append({"title": current_title, "content": section_content})
            titles.append(current_title)

OUTPUT

[eurekaHm]

My program gets stuck at this step: yield self.create_text_message() and returns an error message at the end, it is strange that the message can be generated successfully