DeepSeek-R1 x Agentic RAG：构建带"深度思考"开关的知识研究助理｜深度长文

......
#RAG查询引擎
query_engine = vector_index.as_query_engine(similarity_top_k=5,llm=llm_openai)
query_tool = QueryEngineTool.from_defaults(query_engine=query_engine,name=f'query_tool_{name}',
                                               description=f'用来查询DeepSeek-R1的技术细节')

...
#搜索工具
search_tool = FunctionTool.from_defaults(fn=web_search,name = 'search_tool',description='用来搜索网络以获取信息')

#RAG agent
rag_agent = FunctionCallingAgent.from_tools(
[query_tool,search_tool],
llm=llm_openai,
system_prompt="""
你是一个问题回答专家。你必须使用工具查询技术细节或搜索网络信息，并始终用中文回答问题。
* 尽量保留工具返回中的Markdown图片信息。* 如果一个工具无法回答问题，请尝试另一个工具。
不要基于先验知识或编造。""")

agent_response = rag_agent.query("How is DeepSeek-R1-Zero AIME accuracy during training")
display(Markdown(str(agent_response)))

...
    # 根据大纲提出一些问题
    @step()
    async def plan_questions(
        self, ctx: Context, ev: OutlineEvent
    ) -> QuestionEvent:

        #事件中的大纲信息，并保存到上下文，后面步骤中要用
        outline = ev.outline
        await ctx.set("outline", outline)

        #上下文中获取原始问题
        query = await ctx.get("original_query")

        #Prompt提示
        prompt = f"""你需要根据输入的问题和回答大纲制定一系列问题，这些问题将帮助获得完成大纲所需的事实：
        * 你不能假设任何现有的知识
        * 问题需要涵盖大纲中的所有主要要点
        * 每一个问题必须自洽完整，不能依赖于上下文
        * 你的输出应该是一个问题列表，每个问题在新的一行
        * 问题最多不超过5个
        ----------------
        输入问题：{query}
        ----------------
        回答大纲：{outline}"""

        #生成问题（调用LLM）
        response = await llm_openai.acomplete(prompt)

        #取出<think>部分，如果你使用deepseek-r1才需要
        response = re.sub(r"<think>.*?</think>", "", str(response), flags=re.DOTALL)
        questions = response.split("\n")
        questions = [x for x in questions if x]

        #问题数量
        await ctx.set("num_questions", len(questions))

        #触发下一个step，并传递问题
        #这里会触发并行处理
        for question in questions:
           ctx.send_event(QuestionEvent(question=question))

...
    @step()
    async def answer_question(
        self, ctx: Context, ev: QuestionEvent
    ) -> AnswerEvent:

        #取出问题
        question = ev.question

        #调用上面创建的agentic rag获得答案
        response = await rag_agent.aquery(question)

        #如果打开“深度思考”，则触发下一步；否则就结束！
        if (await ctx.get("deepresearch")):
            return AnswerEvent(question=question, answer=str(response))
        else:
            return StopEvent(result=str(response))

async def main():

    #workflow实例
    rsearch_agent = DeepResearchAgent(timeout=600, verbose=True)

    #调用，deepresearch为控制开关
    handler = rsearch_agent.run(
        query ="DeepSeek-R1大模型用了什么技术？国际上评价如何",
        deepresearch = False,
        rag_agent = rag_agent
    )
    final_result = await handler
    print(final_result)