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）：包括用户信息提取节点、动作节点、人工干预节点、注册用户节点和响应用户节点。

以下是部分代码实现：

fromgrafi.common.topics.topicimportTopic
fromgrafi.common.topics.human_request_topicimporthuman_request_topic
fromgrafi.common.topics.output_topicimportAgent_output_topic
fromgrafi.nodes.llm_nodeimportLLMNode
fromgrafi.nodes.llm_function_call_nodeimportLLMFunctionCallNode
fromgrafi.commands.llm_response_commandimportLLMResponseCommand
fromgrafi.commands.function_calling_commandimportFunctionCallingCommand
fromgrafi.tools.openai_toolimportOpenAITool
fromgrafi.common.models.messageimportMessage
fromgrafi.common.decorators.llm_functionimportllm_function
fromgrafi.common.models.execution_contextimportExecutionContext
importjson
importuuid

# 定义主题
user_info_extract_topic = Topic(name="user_info_extract_topic")
hitl_call_topic = Topic(
  name="hitl_call_topic",
  condition=lambdamsgs: msgs[-1].tool_calls[0].function.name !="register_client",
)
register_user_topic = Topic(
  name="register_user_topic",
  condition=lambdamsgs: 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助手了。以下是测试代码：

deftest_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 = []
   formessageinoutput:
     try:
        content_json = json.loads(message.content)
        responses.append(content_json["question_description"])
     exceptjson.JSONDecodeError:
        responses.append(message.content)

    respond_to_user =" and ".join(responses)
    print("Assistant:", respond_to_user)

   # 如果注册成功，结束循环
   if"registered"inoutput[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，或许它就是你一直在寻找的那个“神器”。