在当今数字化时代，人工智能（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",
condition=lambda msgs: msgs[-1].tool_calls[0].function.name != "register_client",
)
register_user_topic = Topic(
name="register_user_topic",
condition=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", content=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
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，或许它就是你一直在寻找的那个“神器”。
