Situation

• 我們正在嘗試用 llama-index 的 workflow 自行架構 SubQuestionQueryEngine
• 應用情境為： 給定一題單選題考題
  • step1: query
    • 考題會先被切分成一系列的子問題
  • step2: sub_question
    • 對每一個子問題，我們讓 ReActAgent 用 tavily 去搜參考資料找出答案，回傳 answer
  • step3: combine_answers
    • 彙整所有的子問題以及答案來進行作答
• 在 Day11: SubQuestionQueryEngine(上): SubQuestion 與 Workflow 我們已經確定了整體 workflow 的架構
  • 測試了 SubQuestion 的生成但是覺得不滿意
• 在 Day12: SubQuestionQueryEngine(中): Streaming events 與 ReActAgent 釐清中間用來回答子問題的 ReActAgent
  • 但是留了自訂的 prompt 沒有改

Task

• 我們今天直接 End2End 的把整個 SubQuestionQueryEngine 跑起來
  • 包含 step3: combine_answers
  • 還有把預設的 prompt 改一改

Action

1. 首先是必要的 import

   import os
   import json
   
   from dotenv import find_dotenv, load_dotenv
   _ = load_dotenv(find_dotenv())
   TAVILY_API_KEY = os.getenv("TAVILY_API_KEY")
   
   from llama_index.core.workflow import (
       step,
       Context,
       Workflow,
       Event,
       StartEvent,
       StopEvent,
   )
   from llama_index.core import PromptTemplate
   from llama_index.llms.openai import OpenAI
   from llama_index.core.agent.workflow import ReActAgent
   from llama_index.tools.tavily_research.base import TavilyToolSpec
   from llama_index.core.tools import FunctionTool
   from llama_index.utils.workflow import draw_all_possible_flows
   

2. input

   qset = {
     "id": "113-1-1-med-surg",
     "year": "113",
     "time": "1",
     "qid": "1",
     "discipline": "內外科護理學",
     "ans": "C",
     "question": "有關多發性硬化症之診斷檢查，下列何者錯誤？",
     "options": {
      "A": "腦脊髓液分析可發現IgG抗體上升",
      "B": "視覺誘發電位可觀察到受損的神經在傳導過程出現延遲和中斷",
      "C": "超音波檢查可發現中樞神經系統髓鞘脫失",
      "D": "核磁共振影像可用來確認多發性硬化症之斑塊"
     },
     "discipline_slug": "med-surg"
   }
   exam_question = f"題幹: {qset['question']}\n選項: \nA: {qset['options']['A']}; B: {qset['options']['B']}; C: {qset['options']['C']}; D: {qset['options']['D']}."
   print(exam_question)
   

   • 我們的測試輸入是一題單選題，原始答案是 C，答案沒有餵給 LLM
   • 看起來像這樣
     

3. Prompt
   包含: SUB_QUESTION_PROMPT, AGENT_PROMPT, COMBINE_ANSWER_PROMPT

• SUB_QUESTION_PROMPT

    SUB_QUESTION_PROMPT = PromptTemplate("""你是一個考題拆解助手。請根據以下單選題，產生一系列子問題。規則如下：

    1. 子問題需為單一問句，避免複合句。
    2. 每個子問題必須包含完整上下文，不可依賴原始題目才能理解。
    3. 子問題的集合在合併答案後，應能完整回答此單選題。
    4. 回應必須是**純 JSON 格式**，不得包含任何額外文字或 Markdown。

    ### 範例輸出：
    {{
      "sub_questions": [
        "舊金山的人口是多少？",
        "舊金山的年度預算是多少？",
        "舊金山的 GDP 是多少？"
      ]
    }}

    以下是單選題：
    {exam_question}
    """
    )

    print(SUB_QUESTION_PROMPT.format(exam_question=exam_question))

• 好讀版
  

• 主要就是把 Day11: SubQuestionQueryEngine(上): SubQuestion 與 Workflow 發現的幾個問題講給 chatgpt 聽， 然後請他改一版出來

• AGENT_PROMPT

  • 因為太長了，所以用貼的 這裡
  • 原始英文的版本也附在下面作為對照
    • 可以用 agent.get_prompts() 取
  • 試改這個主要不是認定說用中文的會比較好，而是哪天預設的不 work 的時候至少我們還有招

• COMBINE_ANSWER_PROMPT

  COMBINE_ANSWER_PROMPT = PromptTemplate("""你是一個考題作答助手。以下是一題單選題，已經被拆解成數個子問題，並且每個子問題都已有答案。
  請將所有子問題的答案整合，產生一個完整且連貫的最終解答，以回答原始單選題。
  
  以下是單選題：
  {exam_question}
  
  子問題與答案：
  {sub_qa}
  """)
  

4. Event

   class QueryEvent(Event):
       question: str
   
   
   class AnswerEvent(Event):
       question: str
       answer: str
   

• 我們現在的 workflow 還算單純
  • 就 subquestion 切了之後會用 QueryEvent 發出去
  • Agent 答完 subquestion 之後會用 AnswerEvent 發出去

5. run_agent_with_stream

• 我們這邊先訂一個 helper function 用來 run ReActAgent
• 這個會在 workflow 裡面呼叫
• 主要就是會打印中間步驟的過程，不然他跑很久中間都沒有回應很可怕
  • 用的是 Day12: SubQuestionQueryEngine(中): Streaming events 與 ReActAgent 講的技巧

  from llama_index.core.agent.workflow import AgentInput, AgentOutput, ToolCall, ToolCallResult
  
  async def run_agent_with_stream(agent, query):
      handler = agent.run(query)
      results = []
  
      async for ev in handler.stream_events(expose_internal=False):
          name = ev.__class__.__name__
          print(f"-----stream event: {name}")
          results.append((name, ev))
  
          if isinstance(ev, AgentInput):
              print(f"len of chat message: {len(ev.input)}")
          elif isinstance(ev, AgentOutput):
              print(ev.response.blocks[0].text)
          elif isinstance(ev, ToolCall):
              print(f"{ev.tool_name}: {ev.tool_kwargs}")
          elif isinstance(ev, ToolCallResult):
              num_rv = len(ev.tool_output.blocks)
              print(f"num_result: {num_rv}")
  
      # 最終 response
      response = await handler
      return results, response
  

6. workflow

• 主要的架構可以參考 Day11: SubQuestionQueryEngine(上): SubQuestion 與 Workflow
  • 這邊只是把原本直接給值的地方換成 llm 呼叫
• 再來是新增的 combine_answers 的部分，主要就是兩步
  • 一個是把中間回傳的子問題以及子答案串成單一個 string 叫 subqa
    • 這步可能可以要求他要有 citation
  • 然後就是 prompt llm 叫他回答原始考題
• 三個 initial 的部分都應該可以初始化一次之後就讓他從 ctx 拿，這個我們就下次改進
• 我們把昨天自製的 tavily tool 改回 預設的版本
• ReActAgent 可以用 .update_prompts 把原本的 system prompt 改掉

  class SubQuestionQueryEngine(Workflow):
      @step
      async def query(self, ctx: Context, ev: StartEvent) -> QueryEvent:
          # initial
          llm = OpenAI(model="gpt-5-mini", temperature=0, json_mode=True)
          # subquestions gen
          print('sub question gen...')
          response = llm.complete(SUB_QUESTION_PROMPT.format(exam_question=ev.question))
          print('sub question gen complete')
          sub_questions = json.loads(response.text)['sub_questions']
          # get num_question
          num_question = len(sub_questions)
          await ctx.store.set("num_question", len(sub_questions))
          await ctx.store.set("exam_question", exam_question)
          for idx, q in enumerate(sub_questions):
              print(f"send Q{idx+1}: {q}")
              ctx.send_event(QueryEvent(question=q))
          return None
  
      @step
      async def sub_question(self, ctx: Context, ev: QueryEvent) -> AnswerEvent:
          # initial
          tavily_tool = TavilyToolSpec(
              api_key=TAVILY_API_KEY,
          )
          tavily_tool_list = tavily_tool.to_tool_list()
          llm = OpenAI(model="gpt-5-mini", temperature=0, is_streaming=False)  # streaming False for non-verified organisations
          agent = ReActAgent(tools=tavily_tool_list, llm=llm, streaming=False, verbose=False)
          agent.update_prompts({"react_header": AGENT_PROMPT})
          # call
          results, response = await run_agent_with_stream(agent, ev.question)
          answer = response.response.blocks[0].text
          return AnswerEvent(question=ev.question, answer=answer)
  
      @step
      async def combine_answers(
          self, ctx: Context, ev: AnswerEvent
      ) -> StopEvent | None:
          num_question = await ctx.store.get("num_question")
          exam_question = await ctx.store.get("exam_question")
          # wait until we receive all events
          result = ctx.collect_events(ev, [AnswerEvent] * num_question)
          if result is None:
              print('combine_answers waite None')
              return None
          # combine sub_question and sub_answer
          sub_qa = "\n\n".join([
              f"Question: {qa.question}: \n Answer: {qa.answer}"
              for qa in result
          ])
  
          llm = OpenAI(model="gpt-5-mini", temperature=0, is_streaming=False)
          response = llm.complete(COMBINE_ANSWER_PROMPT.format(sub_qa=sub_qa, exam_question=exam_question))
          return StopEvent(result=response.text)
  

7. workflow visualize

   draw_all_possible_flows(
       SubQuestionQueryEngine, filename="day13_sub_question_query_engine.html"
   )
   

• 結果如下：
  

8. run

• code

  w = SubQuestionQueryEngine(timeout=600, verbose=False)
  
  handler = w.run(start_event=StartEvent(question=exam_question))
  response = await handler
  

9. Results

• 首先是整體題目的 response，可以看到答案是正確的
  

• 接著是更新 prompt 後的 subquestion，可以與前天對照，在 不需要上下文就能看懂，以及 不要一個子問題裡又問了好幾個問題 的方面確實有改善
  

修改前:

• 最後是更新 prompt 後的 ReAct Agent，這邊只有擷其中一個子問題
  
  修改前:
  
• 觀察如下：
  • 改成中文 prompt 之後比較明顯的是 檢索的關鍵字 變成用中文了
  • Thought 後面的文字也是中文，感覺起來很像他變成在用中文思考
• 在其他 case 下還是有看到沒有查資料直接回答的情況，不過沒關係，我們後面會強迫他要引經據典

Summary

• 我們今天補上了最後 combine_answers 的部分

• 並且嘗試修改了 子問題 生成部分，以及 ReActAgent 部分的 prompt

• 最後搭建出了完整的 SubQuestionQueryEngine as Workflow

  • 至少在單一測資下看起來還有模有樣

• 熬了三天我們終於結束 SubQuestionQueryEngine 的部分，明天要來把 CitationQueryEngine 也整進來，造出我們主題 1-2 的 Dataset

  • 造出來之後我們還可以做個簡易的驗證看看在有查網路跟沒有查網路的情況下，他實際到底是有沒有差