Llamaindex推出workflow應對復雜LLM應用構建，以及技術實現從圖（Graph）轉向事件驅動（EDA）原因解析
同一天，LLM應用開發另一個代表產品Llamaindex也發布了其在此領域的新功能——workflow，進一步提升應用編排的能力。早在去年，Llamaindex在這方面已經有了動作，推出了Query Pipeline（詳見：???應用編排的未來是Pipeline，LlamaIndex開發預覽版推出Query Pipeline，提升應用開發的靈活性???），它是一個聲明式設計，可以自定義整個查詢流程為一個DAG（有向無環圖）流程，支持從簡單到復雜的不同服務流程。
對于一般的RAG類的流程來講，DAG是可以應付的，并且簡單直觀，這也是大量LLM workflow設計的一致選擇，然而對于Agent流程來講，標準的DAG的結構有一定的缺陷，比如無法處理循環邏輯（有環），然而經典的ReAct的流程就是一個循環迭代的過程。除此之外，Llamaindex官方還提出了這一結構的一些其他問題：
1）不易排錯
2）模糊了組件和模塊執行邏輯
3）Pipeline執行器實現越來越復雜，必須處理大量不同的邊（edge）情況
4） 復雜的Pipeline，難以閱讀。
一旦我們在查詢Pipeline中添加了環，這些圍繞圖的開發應用的用戶體驗問題就會被放大。以下是一些常見麻煩：
1）很多核心編排邏輯（如 if-else 語句和 while 循環）都被定義到圖的邊（edge）上。定義這些邊（edge）會變得繁瑣冗長。
2）處理可選值和默認值的邊的情況變得很困難。作為一個框架，很難確定參數是否會從上游節點傳遞。
3）對于構建Agent的開發人員來說，用有環的圖來定義并不總那么自然。Agent封裝了一個由 LLM 驅動的通用實體，它可以接收觀察結果并生成響應。在這里，圖的形式強制要求 "Agent"節點明確定義傳入邊和傳出邊，迫使用戶定義與其他節點的冗長通信模式。
這一些問題，迫使Llamaindex官方團隊重新審視這種設計的合理性。實際上，筆者在設計Flowengine時也遇到這樣的問題，順著dag圖來設計編排執行器雖然很直覺，但是并不是最佳做法，理由兩點：
一，它迫使開發者需要從宏觀解析圖中邊（edge）和節點（node）的關系，整個邏輯非常復雜，特別是對于復雜的流程節點的處理以及失敗情況恢復來講，都涉及到大量的狀態管理，這都使得圖很復雜，特別是對邊的處理，進而導致編排器實現復雜。
二，違反依賴倒置原則，選擇應用編排的方式，很大程度上是希望圖上的組件是可以復用，可插拔的，不應該考慮它到底處于一個什么樣的圖中，畢竟先有組件，再有具體的業務流程Pipeline。而前面的做法，就使得組件節點需要適配圖的結構，這顯然不利于組件沉淀復用，也導致了組件開發的復雜性。
對于此，Llamaindex推出了新的實現方法——事件驅動的模式來協調組件流程執行。顯然事件驅動的模式，將圖流程的調度變成了組件如何訂閱和處理事件上來。這樣很多原來邊上的處理邏輯就變成了組件自己的行為，極大的降低了復雜度和依賴，并且這樣的設計可以很容易的實現重試，失敗，超時，循環，甚至是human-in-loop等原本直接解析圖而產生的復雜邏輯。對于pipeline執行器來講，提供消息分發和Context維持，以及根據訂閱情況喚醒執行相關組件即可，也簡化了整個實現的復雜度。
我們來看看Llamaindex的workflow是如何編寫的：
from llama_index.core.workflow import (
StartEvent,
StopEvent,
Workflow,
step,
)
from llama_index.llms.openai import OpenAI
class OpenAIGenerator(Workflow):
@step()
async def generate(self, ev: StartEvent) -> StopEvent:
query = ev.get("query")
llm = OpenAI()
response = await llm.acomplete(query)
return StopEvent(result=str(response))
w = OpenAIGenerator(timeout=10, verbose=False)
result = await w.run(query="What's LlamaIndex?")
print(result)上面例子定義了一個workflow類OpenAIGenerator，其中generate函數使用@step裝飾器標記為這是一個workflow步驟，方法簽名定義了其接收什么樣的事件消息以及返回值定義該步驟執行后發布什么樣的消息。
Llamaindex同時給出了這種方式下循環的實現方法：
class ExtractionDone(Event):
output: str
passage: str
class ValidationErrorEvent(Event):
error: str
wrong_output: str
passage: str
class ReflectionWorkflow(Workflow):
@step()
async def extract(
self, ev: StartEvent | ValidationErrorEvent
) -> StopEvent | ExtractionDone:
if isinstance(ev, StartEvent):
passage = ev.get("passage")
if not passage:
return StopEvent(result="Please provide some text in input")
reflection_prompt = ""
elif isinstance(ev, ValidationErrorEvent):
passage = ev.passage
reflection_prompt = REFLECTION_PROMPT.format(
wrong_answer=ev.wrong_output, error=ev.error
)
llm = Ollama(model="llama3", request_timeout=30)
prompt = EXTRACTION_PROMPT.format(
passage=passage, schema=CarCollection.schema_json()
)
if reflection_prompt:
prompt += reflection_prompt
output = await llm.acomplete(prompt)
return ExtractionDone(output=str(output), passage=passage)
@step()
async def validate(
self, ev: ExtractionDone
) -> StopEvent | ValidationErrorEvent:
try:
json.loads(ev.output)
except Exception as e:
print("Validation failed, retrying...")
return ValidationErrorEvent(
error=str(e), wrong_output=ev.output, passage=ev.passage
)
return StopEvent(result=ev.output)
w = ReflectionWorkflow(timeout=60, verbose=True)
result = await w.run(
passage="There are two cars available: a Fiat Panda with 45Hp and a Honda Civic with 330Hp."
)
print(result)在這個例子中，validate步驟接收試驗性模式提取的結果作為事件，并且它可以通過返回ValidationErrorEvent來決定再次嘗試，該ValidationErrorEvent最終將被傳遞到extract步驟，該extract步驟將執行下一次嘗試。這樣就實現了循環迭代的邏輯。
由于編程本身的問題，復雜的業務流程讀代碼是件痛苦的事情，Llamaindex提供了類似LangGraph Studio的能力，對執行流程可視化，方便開發者進行調試。
可以看出，Llamaindex在應對復雜的LLM應用時，采用了與Langchain相似的策略，即高代碼+可視化輔助調試的思路。這其中，事件驅動的流程編排是一個獨特的設計。但筆者認為，事件驅動本身是可以和聲明式、低代碼Pipeline開發相融合的，用戶可以采用直觀的拖拉拽編排整個流程，而編排器實現可以采用事件驅動的方式而非解析圖的方式，這樣豈不是更好？甚至可以提供兩種模式編程和低代碼可視化，兩者還可以實現互操作，更大層面覆蓋了不同背景的開發者。事實上，FlowEngine便是采用了這樣的設計，更多細節可以加入群了解。
更多llamaxindex的workflow細節可以參看官方博文：??https://www.llamaindex.ai/blog/introducing-workflows-beta-a-new-way-to-create-complex-ai-applications-with-llamaindex??
