----------
操作系统：Ubuntu 22.04.5
Python 版本：3.11.10
PyTorch 版本：2.5.1
----------

from modelscope import snapshot_download
model_dir = snapshot_download('deepseek-ai/DeepSeek-R1-Distill-Qwen-7B', cache_dir='/home/dubbo/model', revision='master')

# the path of a model. It could be one of the following options:
# 1. A local directory path of a turbomind model
# 2. The model_id of a lmdeploy-quantized model
# 3. The model_id of a model hosted inside a model repository
model = 'deepseek-ai/DeepSeek-R1-Distill-Qwen-7B'

backend_config = TurbomindEngineConfig(
    cache_max_entry_count=0.2,
    max_context_token_num=20544,
    session_len=20544
)

gen_config = GenerationConfig(
    top_p=0.95,
    temperature=0.6,
    max_new_tokens=8192,
    stop_token_ids=[151329, 151336, 151338],
    do_sample=True  # enable sampling
)

class DeepSeekAiServicer:

    def __init__(self, model: str, backend_config: TurbomindEngineConfig, gen_config: GenerationConfig):
        self.llm = pipeline(model, backend_config=backend_config)
        self.gen_config = gen_config

    def chat(self, content):

        # According to DeepSeek's official recommendations,
        # Each prompt needs to end with <think>\n
        # If it is a mathematical reasoning content, it is recommended to include (in both Chinese and English):
        # Please reason step by step, and put your final answer within \boxed{}.
        prompts = [{
            "role": "user",
            "content": "What is the meaning of life?<think>\n"
            }]

        # Response Example:
        # "<think> The meaning of life is to be happy. </think> I think the meaning of life is to be happy."
        response = self.llm(prompts, gen_config=self.gen_config)
        return response

"""
The code related to the configuration is the same as above.
"""

class DeepSeekAiServicer:

    def __init__(self, model: str, backend_config: TurbomindEngineConfig, gen_config: GenerationConfig):
        self.llm = pipeline(model, backend_config=backend_config)
        self.gen_config = gen_config

    def chat(self, stream):
        # read request from stream
        request = stream.read()
        print(f"Received request: {request}")
        # prepare prompts
        prompts = [{
            "role": request.role,
            "content": request.content + "<think>\n"
        }]

        is_think = False

        # perform streaming inference
        for item in self.llm.stream_infer(prompts, gen_config=gen_config):
            # update think status
            if item.text == "<think>":
                is_think = True
                continue
            elif item.text == "</think>":
                is_think = False
                continue
            # According to the state of thought, decide the content of the reply.
            if is_think:
                # send thought
                stream.write(chat_pb2.ChatReply(think=item.text, answer=""))
            else:
                # send answer
                stream.write(chat_pb2.ChatReply(think="", answer=item.text))

        stream.done_writing()

def build_server_handler():
    # build a method handler
    deepseek_ai_servicer = DeepSeekAiServicer(model, backend_config, gen_config)
    method_handler = RpcMethodHandler.server_stream(
        deepseek_ai_servicer.chat,
        method_name="chat",
        request_deserializer=chat_pb2.ChatRequest.FromString,
        response_serializer=chat_pb2.ChatReply.SerializeToString,
    )
    # build a service handler
    service_handler = RpcServiceHandler(
        service_name="org.apache.dubbo.samples.llm.api.DeepSeekAiService",
        method_handlers=[method_handler],
    )
    return service_handler

if __name__ == '__main__':
    # build a service handler
    service_handler = build_server_handler()
    service_config = ServiceConfig(service_handler=service_handler)

    # Configure the Zookeeper registry
    registry_config = RegistryConfig.from_url("zookeeper://zookeeper:2181")
    bootstrap = Dubbo(registry_config=registry_config)

    # Create and start the server
    bootstrap.create_server(service_config).start()

    # 30days
    sleep(30 * 24 * 60 * 60)

@Component
public class Consumer implements CommandLineRunner {

    @DubboReference
    private DeepSeekAiService deepSeekAiService;

    @Override
    public void run(String... args) throws Exception {
        ChatRequest request = ChatRequest.newBuilder()
                .setRole("user")
                .setContent("你好，你是谁？你能帮我制定一个dubbo学习计划嘛？")
                .build();

        deepSeekAiService.chat(request, new StreamObserver<ChatReply>() {

            private boolean isThinkPrinted = false;
            private boolean isAnswerPrinted = false;

            @Override
            public void onNext(ChatReply value) {
                printSection("Think", value.getThink(), isThinkPrinted);
                printSection("Answer", value.getAnswer(), isAnswerPrinted);
            }

            @Override
            public void onError(Throwable t) {
                System.err.println("Error received: " + t.getMessage());
                t.printStackTrace();
            }

            @Override
            public void onCompleted() {
                System.out.println("------------ Chat Completed ------------");
            }

            private void printSection(String label, String content, boolean isPrinted) {
                if (!content.isEmpty()) {
                    if (!isPrinted) {
                        System.out.println("------------- " + label + " -------------");
                        if (Objects.equals("Think", label)){
                            isThinkPrinted = true;
                        } else {
                            isAnswerPrinted = true;
                        }
                    }
                    System.out.print(content);
                }
            }
        });
    }
}