Situation

• 先前我們在 Day6: pdf2txt 使用 llama-parse 與 mistral-ocr 時，我們已經把包含一系列單選題的 pdf 檔轉為 txt 檔案，後續我們會拿裡面的單選題來當成測資
• 由於整理 context 資料這件事有點卡住了(QQ)，我們這次先從"前面"的模塊往後接看能不能給我們什麼靈感
• 理想的劇本是：
  • 給定一題單選題考題
  • 這些單選題考題會被切成一系列的子問題(SubQuestion)發送出去
  • 收到子問題的 agent
    • 自己去查網路得到一系列的 context
    • 自己彙整一下這些查到的 context 得到自己的觀點
    • 逐句引用自己觀點來回答子問題
  • 彙整所有子問題的答案後，再逐句引用子問題的答案來對原始的單選題進行推理作答
  • 由於單選題可以對答案，所以我們可以比較各種情況來評估模塊是否進一步修改
  • 此外我們可以得到 (subquestion, context, answer) 這樣的 dataset
• 然後現在我們來體驗現實到底有多骨感

Task

• 我們這次要來探索 llama-index 的 SubQuestionQueryEngine (workflow 簡化版)
  • 這個是使用 llama_index 的 workflow 構建 SubQuestionQueryEngine (簡化版)的官方範例:
    • Sub Question Query Engine as a workflow
      • 主要參考這個
  • 這個是 llama_index 使用 SubQuestionQueryEngine 的範例
    Sub Question Query Engine
• 整體會分成兩部分：
  • SubQuestionQueryEngine(上)，也就是今天
    • 我們要先學一下 workflow 怎麼發送多個事件(subquestion)出去，以及等到多個事件都處理完了才開始下一步行動
      • 參考 Concurrent execution of workflows
    • 然後我們感受一下把單選題切成多個子問題的部分有沒有需要進一步修改
      • QQ
  • SubQuestionQueryEngine(下)
    • 主要工作會放在回答子問題的部份我們換成 agent
    • 以及有了所有子問題的答案後我們怎麼填答

Action

part1: workflow

1. import and setup

   # pip install llama-index-utils-workflow
   # OPENAI_API_KEY
   
   import os
   from dotenv import find_dotenv, load_dotenv
   _ = load_dotenv(find_dotenv())
   
   from llama_index.core.workflow import Workflow
   from llama_index.core.workflow import step
   from llama_index.core.workflow import Event
   from llama_index.core.workflow import StartEvent
   from llama_index.core.workflow import StopEvent
   from llama_index.core.workflow import Context
   from llama_index.utils.workflow import draw_all_possible_flows
   

   • 首先是需要安裝 workflow 的 utils: pip install llama-index-utils-workflow，用來可視化定義出來的 workflow
   • 然後我們需要 .env 下有 OPENAI_API_KEY，雖然這個 part2 才會用到
   • 接著我們 import 一系列 workflow 要用到的模塊

2. workflow 概覽

• 我們這邊分成三個 steps:

  • query

    • 這步主要是在給定問題的情況下，我們會 prompt llm 去產一個 list of question，就是我們的 subquestion
      • 所以這時我們會拿到 num_question
    • 然後我們會用一個 for 逐個把 subquestion 發出去
    • 因為 subquestion 前面已經發出去了，所以這步的回傳會是 None

  • sub_question

    • 這步會收單獨一個子問題，然後想辦法具體的給出這個子問題的答案
    • 中間會涉及查網路、總結等等有的沒的，所以預計是放一個 Agent 在這邊
    • 但我們今天可以先不管他

  • combine_answers

    • 這步如果有了一個 sub_question 答完，就會被觸發一次，然後我們會先不管他
    • 要等到所有的 sub_question 都答完了，才真的開始做事
    • 做事就是總結完然後回傳我們的答案

• 我們就三步

  • 基本上就是個跟把大象放冰箱一樣的事

• 如果你覺得這幾步命名很混淆的話

  • 我也有同感
  • 我們是照著範例 Sub Question Query Engine as a workflow 命的，為了有個比較清楚的對照，這個沒關係後面我們會改

3. Event

• code

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

• 我們的 Event 有 QueryEvent 跟 AnswerEvent
  • QueryEvent 放的就是 subquestion
  • AnswerEvent 放的就是 subquestion 還有對應的 answer
• 如果你對什麼是 Event 不熟悉，你可以跟著昨天 Day10: CitationQueryEngine 與 Workflow 跑一次，我們架了我們的第一個 workflow

4. workflow 本 low

• 這邊只是確認整體會按照我們預期的順序執行，所以內容我們都先用假的

  class SubQuestionQueryEngine(Workflow):
      @step
      async def query(self, ctx: Context, ev: StartEvent) -> QueryEvent:
          # Fake subquestions gen
          FAKE_NUM_SUB_QUESTION = 5
          sub_questions = [f'q{i}' for i in range(FAKE_NUM_SUB_QUESTION)]
  
          # get num_questions
          num_question = len(sub_questions)
          await ctx.store.set("num_question", len(sub_questions))
  
          for q in sub_questions:
              #self.send_event(QueryEvent(question=question))
              print(f"send: {q}")
              ctx.send_event(QueryEvent(question=q))
          return None
  
      @step
      async def sub_question(self, ctx: Context, ev: QueryEvent) -> AnswerEvent:
          print(f"Sub-question is {ev.question}")
          # get fake answer
          answer = f"answer of: {ev.question}"
          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")
          # wait until we receive all events
          result = ctx.collect_events(ev, [AnswerEvent] * num_question)
          if result is None:
              print('combine_answers output None')
              return None
  
          # do something with all {num_question} together
          print(result)
          return StopEvent(result="Done")
  

• query 的部分:
  • 我們先取得所有的 sub_questions
  • 把總共有多少 sub_questions 存到 Context combine_answers 的時候要用
  • 用 ctx.send_event 把 sub_question 發出去
    • 這邊官網範例是用 self.send_event，不確定是不是版本問題，總之我用 ctx.send 才不會報錯
• sub_question 就是拿到 question 的話我們就回他 'answer of {question}'
• combine_answers
  • 先取得 num_question
  • 用 ctx.collect_events(ev, [AnswerEvent] * num_question) 來取 answer
    • 這個如果沒有全部回答完就會是 None，然後我們就不理他
  • 真的拿到所有 sub_question 的答案我們才要做事

5. 可視化一下我們的 workflow

• code

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

• result
  • 

6. 執行看看

• code

  w = SubQuestionQueryEngine(timeout=10, verbose=False)
  result = await w.run()
  print('---')
  print(result)
  

• 結果

  send: q0
  send: q1
  send: q2
  send: q3
  send: q4
  Sub-question is q0
  Sub-question is q1
  Sub-question is q2
  Sub-question is q3
  Sub-question is q4
  combine_answers output None
  combine_answers output None
  combine_answers output None
  combine_answers output None
  [AnswerEvent(question='q0', answer='answer of: q0'), AnswerEvent(question='q1', answer='answer of: q1'), AnswerEvent(question='q2', answer='answer of: q2'), AnswerEvent(question='q3', answer='answer of: q3'), AnswerEvent(question='q4', answer='answer of: q4')]
  ---
  Done
  

• 先把 verbose 關掉不然很吵
• combine_answers 確實會被執行 num_question 次，但只有最後一次才真的會做事
• 整體來說這樣的流程確實符合我們的預期

part2: subquestions

7. 首先是我們的範例考題

   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"
   }
   

8. 接著是我們的 prompt

   question = f"以下為單選題，請從各選項中選擇最符合題意的答案：\n題幹: {qset['question']}.\n選項: \nA: {qset['options']['A']}; B: {qset['options']['B']}; C: {qset['options']['C']}; D: {qset['options']['D']}."
   print(question)
   prompt = f"""給定一個使用者的問題，請輸出一系列相關的子問題，讓所有子問題的答案合在一起後，
   能完整回答該問題。
   請只用純 JSON 格式回應，不要包含任何 Markdown，例如：
   {{
       "sub_questions": [
           "舊金山的人口是多少？",
           "舊金山的預算是多少？",
           "舊金山的 GDP 是多少？"
       ]
   }}
   以下是使用者的問題：{question}
   """
   

• 好讀版：
  • 
• 主要是參考原始範例 Sub Question Query Engine as a workflow 然後給 chatgpt 翻譯成中文
  • 原始的 prompt 有 tool 這邊就刪去了，主要是我覺得他範例這邊也是直接 prompt llm 就出子問題，沒有實際用到 tool
  • 然後要求用 json 回傳

9. 呼叫與回傳

• code

  import json
  from llama_index.llms.openai import OpenAI
  llm = OpenAI(
      model="gpt-5-mini",
      temperature=0,
      json_mode=True
  )
  response = llm.complete(prompt)
  print(json.loads(response.text))
  

• result:

  • 

• 首先他確實是回傳了 json 給我們，這個後續我們格式確定可以加上更嚴格的限制

• 以回答的品質來說：

  • 還是希望每個子問題就是一個直接的問題
  • 而且子問題不應該需要有任何上下文才能看懂
  • 問題看起來不夠通用，這樣很有可能在錯誤的敘述上鑽牛角尖導致最後選不到對的答案

• 看來我們後面還要看看是不是有更靠譜的方法來幫我們生成子問題

  • 比如說先有了第一個子問題的答案，然後才來產生二個子問題

Summary

• 我們今天試著把 SubQuestion as Workflow 的骨架做出來
• 學會了怎麼發多個任務，以及怎麼等到所有發出去的任務都做完了才做事
• 測試了範例的子問題生成步驟(就是query)，結果不盡如人意，但沒關係我們的庫存裡還有其他範例
• 我們明天先來把整個流程跑通，再來看看後面怎麼改
• 好想趕快進展到把所有事情放在一起完成我們的 End to End 流程

Reference

• Sub Question Query Engine

  • SubQuestionQueryEngine 的使用範例

• Sub Question Query Engine as a workflow

  • 把 SubQuestionQueryEngine 做成 workflow 的官方範例

• Concurrent execution of workflows

  • workflow 如何 Emitting multiple events 以及 Collecting events 的官方文檔