如何在另一个类中使用外源方法的内部代码 - python

问：

from dataclasses import dataclass, field 
from typing import Any, Dict, List, Optional, Union

from llama_index.schema import NodeWithScore 
from llama_index.types import TokenGen 
from llama_index.utils import truncate_text

@dataclass 
class StreamingResponse: 
    """StreamingResponse object."""

    response_gen: TokenGen
    source_nodes: List[NodeWithScore] = field(default_factory=list)
    metadata: Optional[Dict[str, Any]] = None
    response_txt: Optional[str] = None
    
    def __str__(self) -> str:
        """Convert to string representation."""
        if self.response_txt is None and self.response_gen is not None:
            response_txt = ""
            for text in self.response_gen:
                response_txt += text
            self.response_txt = response_txt
        return self.response_txt or "None"
    
    def get_response(self) -> Response:
        """Get a standard response object."""
        if self.response_txt is None and self.response_gen is not None:
            response_txt = ""
            for text in self.response_gen:
                response_txt += text
            self.response_txt = response_txt
        return Response(self.response_txt, self.source_nodes, self.metadata)
    
    def print_response_stream(self) -> None:
        """Print the response stream."""
        if self.response_txt is None and self.response_gen is not None:
            response_txt = ""
            for text in self.response_gen:
                print(text, end="", flush=True)
                response_txt += text
            self.response_txt = response_txt
        else:
            print(self.response_txt)

我想使用上面的 def print_response_stream（self） -> None： 方法的内部代码，但以某种方式更改它，以便我可以将其用于 streamlit 框架，在那里我可以用 message_placeholder.markdown（） 替换 print（） 方法，以便它在我的聊天中打印response_txt和文本，而不是控制台。

instance_of_schema = schema.StreamingResponse()

if instance_of_schema.response_txt is None and instance_of_schema.response_gen is not None:
    response_txt = ""
    for text in instance_of_schema.response_gen:
        message_placeholder.markdown(text, end="", flush=True)
        response_txt += text
    instance_of_schema.response_txt = response_txt
else:
    message_placeholder.markdown(instance_of_schema.response_txt)
    full_response = instance_of_schema.response_txt

But then it gives me this Exception:

instance_of_schema = schema.StreamingResponse()
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^ TypeError: StreamingResponse.init() missing 1 required positional argument: 'response_gen'

所以后来我尝试了这种方式：

if self.response_txt is None and self.response_gen is not None:
    response_txt = ""
    for text in self.response_gen:
        message_placeholder.markdown(text, end="", flush=True)
        response_txt += text
        self.response_txt = response_txt
else: 
    message_placeholder.markdown(self.response_txt)
    full_response = self.response_txt 

it gives me this Ecxeption:

if self.response_txt is None and self.response_gen is not None:
   ^^^^

NameError: name 'self' is not defined

我试过这种方式：

if schema.StreamingResponse.response_txt is None and schema.StreamingResponse.response_gen is not None:
    response_txt = ""
    for text in schema.StreamingResponse.response_gen:
        message_placeholder.markdown(text, end="", flush=True)
        response_txt += text
        schema.StreamingResponse.response_txt = response_txt
else:
    message_placeholder.markdown(schema.StreamingResponse.response_txt)
    full_response = schema.StreamingResponse.response_txt

然后它抛出这个异常：

if schema.StreamingResponse.response_txt is None and schema.StreamingResponse.response_gen is not None:
                                                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

AttributeError: type object 'StreamingResponse' has no attribute 'response_gen'

对不起，我认为可能有一个非常简单的答案，但我不明白，我是 Python 的新手，到目前为止我发现的对我没有帮助。