rag_server/transcription/subtask_decompose.py

# -*- codong: utf-8 -*-
"""问题分解与分步处理工具

此脚本用于将复杂问题分解为多个步骤，并对每个步骤进行单独处理，最终汇总成完整答案。
主要通过调用DeepSeek API实现问题分解和分步解答功能。

功能流程：
1. 配置DeepSeek API参数
2. 定义API调用函数
3. 接收原始问题并构建系统提示
4. 调用API分解问题为多个步骤
5. 逐一处理每个步骤并获取回答
6. 汇总所有信息生成完整结果
"""

from openai import OpenAI
from typing import List, Dict, Any, Optional

from cosine_similarity_example import cosine_similarity, cosine_similarity_numpy
# 导入各种可能需要的库（部分未在当前版本使用，但为扩展预留）ilarity_numpy
# 配置信息
# DeepSeek 配置
DEEPSEEK_API_KEY = "sk-cfd2df92c8864ab999d66a615ee812c5"
DEEPSEEK_MODEL = {
    "DeepSeek-R1": "deepseek-reasoner",
    "DeepSeek-V3": "deepseek-chat",
}
def get_deepseek_completion(
    model: str, 
    prompt: str,
    output_type: str = "text",
    tool_calls: bool = False,
    tools: List[Dict] = None
) -> Optional[Dict | List | str]:
    """调用DeepSeek API获取回答"""
    messages = [{"role": "user", "content": prompt}]
    kwargs = {
        "model": DEEPSEEK_MODEL.get(model, "deepseek-chat"),
        "messages": messages,
    }

    # 添加工具调用参数
    if tool_calls and tools:
        kwargs["tools"] = tools
        kwargs["tool_choice"] = "auto"

    # 创建OpenAI客户端连接DeepSeek API
    client = OpenAI(api_key=DEEPSEEK_API_KEY, base_url="https://api.deepseek.com")

    # 设置JSON输出格式
    if output_type == "json":
        kwargs["response_format"] = {"type": "json_object"}

    try:
        response = client.chat.completions.create(**kwargs)
        choice = response.choices[0]

        if output_type == "text":
            return choice.message.content  # 只返回文本
        elif output_type == "json":
            import json
            return json.loads(choice.message.content)
        else:
            raise ValueError(f"Invalid output_type: {output_type}")

    except Exception as e:
        print(f"[ERROR] DeepSeek API调用失败: {e}")
        return None

def main():
    bias_prompt ="请尽量简洁准确地回答："
    orig_query = "如何养一只金刚鹦鹉"
    system_prompt = f"""你是一个智能助手，负责判断问题的复杂度并给出相应处理。
    如果是简单问题，直接组织为json格式的回答返回该回答，例如:
    {{"steps": ["简单问题的回答"]}}。
    如果是复杂问题，将这个问题分解为多个步骤，步骤的组织为json格式，格式为{{
        "steps": [
            "步骤1",
            "步骤2",
            ...
        ]
    }}
    例如：如何写一篇好的博文
    1.确定博文主题和目标受众；
    2.进金刚鹦鹉集和研究；
    3.制定博文大纲；
    ...
    现在问题如下：{orig_query}
    """

    query = system_prompt
    print(f"\n调用DeepSeek回答: {query}")

    answer = get_deepseek_completion(
        model="DeepSeek-V3",  # 使用DeepSeek V3模型
        prompt=query,  # 使用query_text_1作为prompt
        output_type="json"    # 返回文本格式
    )

    ############# ####解析步骤
    if answer:
        print(f"\nDeepSeek回答 分解子问题为:\n{answer}")
    else:
        print("未获取到DeepSeek的回答")

    steps = answer.get('steps', [])
    
    print(f"\n解析出的步骤数量: {len(steps)}")
    for i, step in enumerate(steps, 1):
        print(f"步骤{i}: {step}")

    ##################处理每个步骤
    context = orig_query
    for i, step in enumerate(steps, 1):
        print(f"\n处理步骤{i}: {step}")
        context += f"\n步骤{i}: {step}"
        # 调用DeepSeek处理每个步骤
        '''
        这个步骤过程有选择：
        1.调用使用RAG模型，调用知识库，获取步骤的回答
        2.将context信息作为上下文一并投入prompt，获取步骤的回答
        '''        
        step_answer = get_deepseek_completion(
            model="DeepSeek-V3",  # 使用DeepSeek V3模型
            prompt=bias_prompt + step,  
            output_type="text"    # 返回文本格式
        )
        '''
        这个步骤过程有选择：
        1.调用使用RAG模型，调用知识库，获取步骤的回答
        2.将context信息作为上下文一并投入prompt，获取步骤的回答
        '''        
        if step_answer:
            print(f"步骤{i}的回答: {step_answer}")
            context += f"\n步骤{i}的回答: {step_answer}"
        else:
            print(f"步骤{i}未获取到回答")


    print("结果:", context)

  

if __name__ == "__main__":
    main()