Github 上 Star 数最多的大模型应用基础服务 Dify 深度解读（一）

# 构造 prompt

GENERATOR_QA_PROMPT = (
    '<Task> The user will send a long text. Generate a Question and Answer pairs only using the knowledge in the long text. Please think step by step.'
    'Step 1: Understand and summarize the main content of this text.\n'
    'Step 2: What key information or concepts are mentioned in this text?\n'
    'Step 3: Decompose or combine multiple pieces of information and concepts.\n'
    'Step 4: Generate questions and answers based on these key information and concepts.\n'
    '<Constraints> The questions should be clear and detailed, and the answers should be detailed and complete. '
    'You must answer in {language}, in a style that is clear and detailed in {language}. No language other than {language} should be used. \n'
    '<Format> Use the following format: Q1:\nA1:\nQ2:\nA2:...\n'
    '<QA Pairs>'
)

def generate_qa_document(cls, tenant_id: str, query, document_language: str):
    prompt = GENERATOR_QA_PROMPT.format(language=document_language)

    model_manager = ModelManager()
    model_instance = model_manager.get_default_model_instance(
        tenant_id=tenant_id,
        model_type=ModelType.LLM,
    )

    # 拼接出完整的调用 prompt

    prompt_messages = [
        SystemPromptMessage(content=prompt),
        UserPromptMessage(content=query)
    ]

    # 调用大模型直接生成问答对

    response = model_instance.invoke_llm(
        prompt_messages=prompt_messages,
        model_parameters={
            'temperature': 0.01,
            "max_tokens": 2000
        },
        stream=False
    )

    answer = response.message.content
    return answer.strip()

tools = []
# 根据用户输入的意图，使用知识库的描述构造 prompt

for dataset in available_datasets:
    description = dataset.description
    if not description:
        description = 'useful for when you want to answer queries about the ' + dataset.name

    description = description.replace('\n', '').replace('\r', '')
    message_tool = PromptMessageTool(
        name=dataset.id,
        description=description,
        parameters={
            "type": "object",
            "properties": {},
            "required": [],
        }
    )
    tools.append(message_tool)

# 支持 ReAct 模式

if planning_strategy == PlanningStrategy.REACT_ROUTER:
    react_multi_dataset_router = ReactMultiDatasetRouter()
    dataset_id = react_multi_dataset_router.invoke(query, tools, model_config, model_instance,
                                                    user_id, tenant_id)
# 支持 Function Call 模式

elif planning_strategy == PlanningStrategy.ROUTER:
    function_call_router = FunctionCallMultiDatasetRouter()
    dataset_id = function_call_router.invoke(query, tools, model_config, model_instance)

# 关键词检索

if retrival_method == 'keyword_search':
    keyword_thread = threading.Thread(target=RetrievalService.keyword_search, kwargs={
        'flask_app': current_app._get_current_object(),
        'dataset_id': dataset_id,
        'query': query,
        'top_k': top_k,
        'all_documents': all_documents,
        'exceptions': exceptions,
    })
    threads.append(keyword_thread)
    keyword_thread.start()

# 向量检索（混合检索中也会调用）

if RetrievalMethod.is_support_semantic_search(retrival_method):
    embedding_thread = threading.Thread(target=RetrievalService.embedding_search, kwargs={
        'flask_app': current_app._get_current_object(),
        'dataset_id': dataset_id,
        'query': query,
        'top_k': top_k,
        'score_threshold': score_threshold,
        'reranking_model': reranking_model,
        'all_documents': all_documents,
        'retrival_method': retrival_method,
        'exceptions': exceptions,
    })
    threads.append(embedding_thread)
    embedding_thread.start()

# 文本检索（混合检索中也会调用）

if RetrievalMethod.is_support_fulltext_search(retrival_method):
    full_text_index_thread = threading.Thread(target=RetrievalService.full_text_index_search, kwargs={
        'flask_app': current_app._get_current_object(),
        'dataset_id': dataset_id,
        'query': query,
        'retrival_method': retrival_method,
        'score_threshold': score_threshold,
        'top_k': top_k,
        'reranking_model': reranking_model,
        'all_documents': all_documents,
        'exceptions': exceptions,
    })
    threads.append(full_text_index_thread)
    full_text_index_thread.start()

for thread in threads:
    thread.join()

# 混合检索之后会执行向量和文本检索结果合并后的重排序

if retrival_method == RetrievalMethod.HYBRID_SEARCH:
    data_post_processor = DataPostProcessor(str(dataset.tenant_id), reranking_model, False)
    all_documents = data_post_processor.invoke(
        query=query,
        documents=all_documents,
        score_threshold=score_threshold,
        top_n=top_k
    )

def invoke_rerank(self, query: str, docs: list[str], score_threshold: Optional[float] = None,
                    top_n: Optional[int] = None,
                    user: Optional[str] = None) \
        -> RerankResult:
    self.model_type_instance = cast(RerankModel, self.model_type_instance)

    # 轮询调用重排序模型，获得重排序得分，并基于得分进行排序

    return self._round_robin_invoke(
        function=self.model_type_instance.invoke,
        model=self.model,
        credentials=self.credentials,
        query=query,
        docs=docs,
        score_threshold=score_threshold,
        top_n=top_n,
        user=user
    )