Graphite框架揭秘：如何用它打造可擴展的AI工作流 原創
在當今數字化時代，人工智能（AI）已經滲透到我們生活的方方面面，從日常的語音助手到復雜的企業級應用，AI正在改變我們的工作和生活方式。然而，隨著AI應用的不斷擴展，企業和開發者們面臨著一個共同的挑戰：如何構建符合特定業務需求的AI解決方案？今天，我們要介紹的Graphite框架，正是為解決這一問題而生。
一、為什么我們需要Graphite？
在AI的世界里，已經有許多強大的工具和平臺，比如ChatGPT、Claude等。但這些通用的AI解決方案在處理一些關鍵任務時，往往顯得力不從心。例如，在金融、醫療等對準確性要求極高的領域，即使是微小的錯誤也可能帶來巨大的損失。這就需要一種更加靈活、可控的AI框架，能夠根據具體的業務需求進行定制。Graphite正是在這樣的背景下誕生的。
Graphite是一個開源框架，專門用于構建特定領域的AI助手。它提供了一個高度可擴展的平臺，能夠根據獨特的企業需求進行定制，讓開發者能夠構建出符合具體業務領域的個性化工作流。用一個形象的比喻來說，Graphite就像是AI領域的“樂高積木”，開發者可以像搭積木一樣，自由組合各種組件，打造出自己想要的AI應用。
二、Graphite的核心架構：簡單、強大且可組合
Graphite的架構設計非常巧妙，它由三個概念層組成：助手（Assistants）、節點（Nodes）和工具（Tools）。
- 助手（Assistants）：它們是整個工作流的“指揮官”，負責協調工作流的運行，并管理整個對話的狀態。你可以把助手想象成一個項目經理，它決定著整個項目的進度和方向。
- 節點（Nodes）：節點是工作流中的一個個“小工人”，每個節點都有自己的職責，比如調用一個語言模型或者執行一個函數。它們各司其職，共同完成復雜的工作。
- 工具（Tools）：工具是節點用來完成具體任務的“工具箱”，比如調用一個API或者運行一個Python函數。這些工具為節點提供了實際操作的能力。
Graphite還采用了事件溯源模式，這意味著每一次狀態的變化都會被記錄下來。這就像是給整個系統安裝了一個“黑匣子”，無論何時出現問題，都可以通過這些記錄來追溯原因。
這種架構的好處是顯而易見的。首先，它讓整個系統變得非常模塊化，開發者可以像搭積木一樣，自由地添加、移除節點，甚至可以在不影響其他部分的情況下，對某個節點進行修改。其次，這種架構提高了系統的靈活性和可擴展性，無論業務需求如何變化，都可以輕松地調整工作流。
三、Graphite的四大核心特性：讓AI應用更可靠
（一）可觀察性（Observability）
在復雜的AI系統中，要找出問題的根源往往像大海撈針一樣困難。Graphite通過事件驅動的架構、日志記錄和追蹤功能，讓開發者能夠實時監控系統的運行狀態，快速定位瓶頸或錯誤。這就像是給AI系統安裝了一雙“透視眼”，讓每一個環節都變得透明可測。
（二）冪等性（Idempotency）
在異步工作流中，由于網絡波動或部分失敗等原因，可能需要重復執行某些操作。Graphite的設計強調冪等性操作，確保即使重復調用，也不會出現數據重復或損壞的情況。這就像是給系統加上了一層“保護罩”，避免了因重復操作而引發的混亂。
（三）可審計性（Auditability）
Graphite將事件作為唯一的事實來源，自動記錄每一次狀態變化和決策路徑。這對于那些需要嚴格遵守法規的行業來說，簡直就是“救星”。無論是為了合規性檢查，還是為了調試和追蹤問題，這些詳細的記錄都提供了有力的支持。
（四）可恢復性（Restorability）
在長時間運行的AI任務中，如果中途出現失敗，重新開始可能會浪費大量的時間和資源。Graphite通過檢查點和基于事件的回放功能，讓工作流可以從失敗的那一刻精準恢復，最大限度地減少了停機時間和資源浪費。
四、動手實踐：用Graphite打造一個“了解你的客戶”AI助手
說了這么多，可能你會問：“聽起來不錯，但是實際操作起來難不難？”別擔心，接下來我們就通過一個簡單的例子，來實際感受一下Graphite的強大。
假設我們要為一家健身房打造一個“了解你的客戶”（KYC）AI助手。這個助手的主要任務是收集客戶的全名和電子郵件地址，完成健身房的注冊流程。如果客戶提供的信息不完整，助手會暫停流程，并要求客戶提供更多信息。
（一）搭建工作流
首先，我們需要安裝Graphite框架。在終端中運行以下命令即可完成安裝：
pip install grafi接下來，我們需要定義工作流中的各個組件。根據前面的描述，我們需要創建以下組件：
- 7個主題（Topics）：包括用戶輸入主題、用戶信息提取主題、人工干預主題等。
- 5個節點（Nodes）：包括用戶信息提取節點、動作節點、人工干預節點、注冊用戶節點和響應用戶節點。
以下是部分代碼實現：
from grafi.common.topics.topic import Topic
from grafi.common.topics.human_request_topic import human_request_topic
from grafi.common.topics.output_topic import agent_output_topic
from grafi.nodes.llm_node import LLMNode
from grafi.nodes.llm_function_call_node import LLMFunctionCallNode
from grafi.commands.llm_response_command import LLMResponseCommand
from grafi.commands.function_calling_command import FunctionCallingCommand
from grafi.tools.openai_tool import OpenAITool
from grafi.common.models.message import Message
from grafi.common.decorators.llm_function import llm_function
from grafi.common.models.execution_context import ExecutionContext
import json
import uuid
# 定義主題
user_info_extract_topic = Topic(name="user_info_extract_topic")
hitl_call_topic = Topic(
name="hitl_call_topic",
cnotallow=lambda msgs: msgs[-1].tool_calls[0].function.name != "register_client",
)
register_user_topic = Topic(
name="register_user_topic",
cnotallow=lambda msgs: msgs[-1].tool_calls[0].function.name == "register_client",
)
register_user_respond_topic = Topic(name="register_user_respond")
# 定義用戶信息提取節點
user_info_extract_node = (
LLMNode.Builder()
.name("UserInfoExtractNode")
.subscribe(agent_input_topic)
.command(
LLMResponseCommand.Builder()
.llm(
OpenAITool.Builder()
.name("UserInfoExtractLLM")
.api_key("YOUR_OPENAI_API_KEY")
.model("gpt-3.5-turbo")
.system_message("Extract user's full name and email from the input.")
.build()
)
.build()
)
.publish_to(user_info_extract_topic)
.build()
)
# 定義動作節點
action_node = (
LLMNode.Builder()
.name("ActionNode")
.subscribe(user_info_extract_topic)
.command(
LLMResponseCommand.Builder()
.llm(
OpenAITool.Builder()
.name("ActionLLM")
.api_key("YOUR_OPENAI_API_KEY")
.model("gpt-3.5-turbo")
.system_message("Decide the next action based on the extracted information.")
.build()
)
.build()
)
.publish_to(hitl_call_topic)
.publish_to(register_user_topic)
.build()
)
# 定義人工干預節點
human_request_function_call_node = (
LLMFunctionCallNode.Builder()
.name("HumanRequestNode")
.subscribe(hitl_call_topic)
.command(
FunctionCallingCommand.Builder()
.function_tool(ClientInfo())
.build()
)
.publish_to(human_request_topic)
.build()
)
# 定義注冊用戶節點
register_user_node = (
LLMFunctionCallNode.Builder()
.name("RegisterUserNode")
.subscribe(register_user_topic)
.command(
FunctionCallingCommand.Builder()
.function_tool(RegisterClient())
.build()
)
.publish_to(register_user_respond_topic)
.build()
)
# 定義響應用戶節點
user_reply_node = (
LLMNode.Builder()
.name("UserReplyNode")
.subscribe(register_user_respond_topic)
.command(
LLMResponseCommand.Builder()
.llm(
OpenAITool.Builder()
.name("UserReplyLLM")
.api_key("YOUR_OPENAI_API_KEY")
.model("gpt-3.5-turbo")
.system_message("Generate a response to the user based on the registration result.")
.build()
)
.build()
)
.publish_to(agent_output_topic)
.build()
)（二）測試助手
現在我們已經搭建好了工作流，接下來就可以測試我們的“了解你的客戶”AI助手了。以下是測試代碼：
def test_kyc_assistant():
execution_context = ExecutionContext(
conversation_id="conversation_id",
execution_id=uuid.uuid4().hex,
assistant_request_id=uuid.uuid4().hex,
)
# 初始化助手
assistant = (
KycAssistant.Builder()
.name("KycAssistant")
.api_key("YOUR_OPENAI_API_KEY")
.user_info_extract_system_message(user_info_extract_system_message)
.action_llm_system_message(
"Select the most appropriate tool based on the request."
)
.summary_llm_system_message(
"Response to user with result of registering. You must include 'registered' in the response if succeed."
)
.hitl_request(ClientInfo())
.register_request(RegisterClient())
.build()
)
whileTrue:
# 獲取用戶輸入
user_input = input("User: ")
input_data = [Message(role="user", cnotallow=user_input)]
# 執行助手
output = assistant.execute(execution_context, input_data)
# 處理輸出
responses = []
for message in output:
try:
content_json = json.loads(message.content)
responses.append(content_json["question_description"])
except json.JSONDecodeError:
responses.append(message.content)
respond_to_user = " and ".join(responses)
print("Assistant:", respond_to_user)
# 如果注冊成功，結束循環
if"registered"in output[0].content:
break
if __name__ == "__main__":
test_kyc_assistant()運行這段代碼后，你可以通過終端與我們的AI助手進行交互。例如：
User: Hi, I'd like to sign up for your gym. Could you help me with the process?
Assistant: Please provide your full name and email address to sign up for the gym.
User: My name is John Doe, and my email is john.doe@example.com
Assistant: Congratulations, John! You are now registered at our gym. If you have any questions or need assistance, feel free to ask!五、測試、觀察、調試與改進：讓助手更完美
在實際使用中，我們可能會遇到各種問題。Graphite通過集成OpenTelemetry和Arize的OpenInference，提供了強大的追蹤和觀察功能。我們可以輕松地捕獲助手的行為數據，快速定位問題。
例如，假設我們在測試中發現，當用戶輸入與注冊無關的內容時，助手會報錯。通過追蹤工具，我們可以迅速定位到問題的根源——動作LLM沒有正確選擇工具。于是，我們可以更新動作LLM的系統提示，讓它在用戶輸入無關內容時，調用??request_client_information??工具，禮貌地詢問用戶是否需要注冊幫助。
這種快速迭代和改進的過程，正是Graphite的魅力所在。它不僅幫助我們快速搭建AI應用，還讓我們能夠持續優化，讓助手變得越來越智能。
六、總結：Graphite，為真實世界AI應用而生
Graphite不僅僅是一個框架，它是一種全新的構建AI應用的思維方式。通過簡單而強大的三層執行模型、事件驅動的編排機制以及對可觀察性、冪等性、可審計性和可恢復性的支持，Graphite為開發者提供了一個靈活、可擴展且可靠的平臺。
無論是構建對話式助手，還是自動化工作流，Graphite都能滿足你的需求。它讓我們能夠像搭積木一樣，輕松構建出符合特定業務需求的AI應用。如果你也對AI開發充滿熱情，不妨試試Graphite，或許它就是你一直在尋找的那個“神器”。
本文轉載自公眾號Halo咯咯 作者：基咯咯
