Llamaindex推出workflow應(yīng)對(duì)復(fù)雜LLM應(yīng)用構(gòu)建，以及技術(shù)實(shí)現(xiàn)從圖（Graph）轉(zhuǎn)向事件驅(qū)動(dòng)（EDA）原因解析
同一天，LLM應(yīng)用開發(fā)另一個(gè)代表產(chǎn)品Llamaindex也發(fā)布了其在此領(lǐng)域的新功能——workflow，進(jìn)一步提升應(yīng)用編排的能力。早在去年，Llamaindex在這方面已經(jīng)有了動(dòng)作，推出了Query Pipeline（詳見：???應(yīng)用編排的未來是Pipeline，LlamaIndex開發(fā)預(yù)覽版推出Query Pipeline，提升應(yīng)用開發(fā)的靈活性???），它是一個(gè)聲明式設(shè)計(jì)，可以自定義整個(gè)查詢流程為一個(gè)DAG（有向無環(huán)圖）流程，支持從簡(jiǎn)單到復(fù)雜的不同服務(wù)流程。
對(duì)于一般的RAG類的流程來講，DAG是可以應(yīng)付的，并且簡(jiǎn)單直觀，這也是大量LLM workflow設(shè)計(jì)的一致選擇，然而對(duì)于Agent流程來講，標(biāo)準(zhǔn)的DAG的結(jié)構(gòu)有一定的缺陷，比如無法處理循環(huán)邏輯（有環(huán)），然而經(jīng)典的ReAct的流程就是一個(gè)循環(huán)迭代的過程。除此之外，Llamaindex官方還提出了這一結(jié)構(gòu)的一些其他問題：
1）不易排錯(cuò)
2）模糊了組件和模塊執(zhí)行邏輯
3）Pipeline執(zhí)行器實(shí)現(xiàn)越來越復(fù)雜，必須處理大量不同的邊（edge）情況
4） 復(fù)雜的Pipeline，難以閱讀。
一旦我們?cè)诓樵働ipeline中添加了環(huán)，這些圍繞圖的開發(fā)應(yīng)用的用戶體驗(yàn)問題就會(huì)被放大。以下是一些常見麻煩：
1）很多核心編排邏輯（如 if-else 語句和 while 循環(huán)）都被定義到圖的邊（edge）上。定義這些邊（edge）會(huì)變得繁瑣冗長。
2）處理可選值和默認(rèn)值的邊的情況變得很困難。作為一個(gè)框架，很難確定參數(shù)是否會(huì)從上游節(jié)點(diǎn)傳遞。
3）對(duì)于構(gòu)建Agent的開發(fā)人員來說，用有環(huán)的圖來定義并不總那么自然。Agent封裝了一個(gè)由 LLM 驅(qū)動(dòng)的通用實(shí)體，它可以接收觀察結(jié)果并生成響應(yīng)。在這里，圖的形式強(qiáng)制要求 "Agent"節(jié)點(diǎn)明確定義傳入邊和傳出邊，迫使用戶定義與其他節(jié)點(diǎn)的冗長通信模式。
這一些問題，迫使Llamaindex官方團(tuán)隊(duì)重新審視這種設(shè)計(jì)的合理性。實(shí)際上，筆者在設(shè)計(jì)Flowengine時(shí)也遇到這樣的問題，順著dag圖來設(shè)計(jì)編排執(zhí)行器雖然很直覺，但是并不是最佳做法，理由兩點(diǎn)：
一，它迫使開發(fā)者需要從宏觀解析圖中邊（edge）和節(jié)點(diǎn)（node）的關(guān)系，整個(gè)邏輯非常復(fù)雜，特別是對(duì)于復(fù)雜的流程節(jié)點(diǎn)的處理以及失敗情況恢復(fù)來講，都涉及到大量的狀態(tài)管理，這都使得圖很復(fù)雜，特別是對(duì)邊的處理，進(jìn)而導(dǎo)致編排器實(shí)現(xiàn)復(fù)雜。
二，違反依賴倒置原則，選擇應(yīng)用編排的方式，很大程度上是希望圖上的組件是可以復(fù)用，可插拔的，不應(yīng)該考慮它到底處于一個(gè)什么樣的圖中，畢竟先有組件，再有具體的業(yè)務(wù)流程Pipeline。而前面的做法，就使得組件節(jié)點(diǎn)需要適配圖的結(jié)構(gòu)，這顯然不利于組件沉淀復(fù)用，也導(dǎo)致了組件開發(fā)的復(fù)雜性。
對(duì)于此，Llamaindex推出了新的實(shí)現(xiàn)方法——事件驅(qū)動(dòng)的模式來協(xié)調(diào)組件流程執(zhí)行。顯然事件驅(qū)動(dòng)的模式，將圖流程的調(diào)度變成了組件如何訂閱和處理事件上來。這樣很多原來邊上的處理邏輯就變成了組件自己的行為，極大的降低了復(fù)雜度和依賴，并且這樣的設(shè)計(jì)可以很容易的實(shí)現(xiàn)重試，失敗，超時(shí)，循環(huán)，甚至是human-in-loop等原本直接解析圖而產(chǎn)生的復(fù)雜邏輯。對(duì)于pipeline執(zhí)行器來講，提供消息分發(fā)和Context維持，以及根據(jù)訂閱情況喚醒執(zhí)行相關(guān)組件即可，也簡(jiǎn)化了整個(gè)實(shí)現(xiàn)的復(fù)雜度。
我們來看看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)上面例子定義了一個(gè)workflow類OpenAIGenerator，其中g(shù)enerate函數(shù)使用@step裝飾器標(biāo)記為這是一個(gè)workflow步驟，方法簽名定義了其接收什么樣的事件消息以及返回值定義該步驟執(zhí)行后發(fā)布什么樣的消息。
Llamaindex同時(shí)給出了這種方式下循環(huán)的實(shí)現(xiàn)方法：
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)在這個(gè)例子中，validate步驟接收試驗(yàn)性模式提取的結(jié)果作為事件，并且它可以通過返回ValidationErrorEvent來決定再次嘗試，該ValidationErrorEvent最終將被傳遞到extract步驟，該extract步驟將執(zhí)行下一次嘗試。這樣就實(shí)現(xiàn)了循環(huán)迭代的邏輯。
由于編程本身的問題，復(fù)雜的業(yè)務(wù)流程讀代碼是件痛苦的事情，Llamaindex提供了類似LangGraph Studio的能力，對(duì)執(zhí)行流程可視化，方便開發(fā)者進(jìn)行調(diào)試。
可以看出，Llamaindex在應(yīng)對(duì)復(fù)雜的LLM應(yīng)用時(shí)，采用了與Langchain相似的策略，即高代碼+可視化輔助調(diào)試的思路。這其中，事件驅(qū)動(dòng)的流程編排是一個(gè)獨(dú)特的設(shè)計(jì)。但筆者認(rèn)為，事件驅(qū)動(dòng)本身是可以和聲明式、低代碼Pipeline開發(fā)相融合的，用戶可以采用直觀的拖拉拽編排整個(gè)流程，而編排器實(shí)現(xiàn)可以采用事件驅(qū)動(dòng)的方式而非解析圖的方式，這樣豈不是更好？甚至可以提供兩種模式編程和低代碼可視化，兩者還可以實(shí)現(xiàn)互操作，更大層面覆蓋了不同背景的開發(fā)者。事實(shí)上，F(xiàn)lowEngine便是采用了這樣的設(shè)計(jì)，更多細(xì)節(jié)可以加入群了解。
更多l(xiāng)lamaxindex的workflow細(xì)節(jié)可以參看官方博文：??https://www.llamaindex.ai/blog/introducing-workflows-beta-a-new-way-to-create-complex-ai-applications-with-llamaindex??
