使用GraphBit构建生产级智能体工作流

原创

用户11764306

发布于 2026-01-15 19:31:19

1120

本文通过完整示例演示了如何使用GraphBit框架构建面向生产的智能体工作流程。我们将创建端到端的客户支持工单处理系统，展示图结构执行、工具调用和可选的LLM驱动智能体如何在一个系统中协同工作。

首先安装所需依赖并导入核心库：

!pip -q install graphbit rich pydantic numpy

import os
import time
import json
import random
from dataclasses import dataclass
from typing import Dict, Any, List, Optional
import numpy as np
from rich import print as rprint
from rich.panel import Panel
from rich.table import Table

初始化GraphBit运行时并配置执行参数：

from graphbit import init, shutdown, configure_runtime, get_system_info, health_check, version
from graphbit import Workflow, Node, Executor, LlmConfig
from graphbit import tool, ToolExecutor, ExecutorConfig
from graphbit import get_tool_registry, clear_tools

configure_runtime(worker_threads=4, max_blocking_threads=8, thread_stack_size_mb=2)
init(log_level="warn", enable_tracing=False, debug=False)

定义支持工单的数据模型并生成模拟数据集：

@dataclass
class Ticket:
    ticket_id: str
    user_id: str
    text: str
    created_at: float

def make_tickets(n: int = 10) -> List[Ticket]:
    seeds = [
        "我的卡片支付失败两次，该怎么办？",
        "我想立即取消订阅。",
        "打开仪表板时应用崩溃。",
        "请更新账户上的电子邮件地址。",
        "7天后仍未收到退款。",
        "我的交付延迟且跟踪信息卡住。",
        "我怀疑账户上有欺诈活动。",
        "如何更改我的账单周期日期？",
        "网站非常慢且超时。",
        "我忘记密码无法登录。",
        "请提供退单流程详情。",
        "需要上个月付款的发票。"
    ]
    random.shuffle(seeds)
    out = []
    for i in range(n):
        out.append(
            Ticket(
                ticket_id=f"T-{1000+i}",
                user_id=f"U-{random.randint(100,999)}",
                text=seeds[i % len(seeds)],
                created_at=time.time() - random.randint(0, 7 * 24 * 3600),
            )
        )
    return out

注册确定性业务工具，实现工单分类、路由和响应草稿生成：

clear_tools()

@tool(_description="将支持工单分类到粗略类别中。")
def classify_ticket(text: str) -> Dict[str, Any]:
    t = text.lower()
    if "fraud" in t or "fraudulent" in t:
        return {"category": "fraud", "priority": "p0"}
    if "cancel" in t:
        return {"category": "cancellation", "priority": "p1"}
    if "refund" in t or "chargeback" in t:
        return {"category": "refunds", "priority": "p1"}
    if "password" in t or "login" in t:
        return {"category": "account_access", "priority": "p2"}
    if "crash" in t or "slow" in t or "timeout" in t:
        return {"category": "bug", "priority": "p2"}
    if "payment" in t or "billing" in t or "invoice" in t:
        return {"category": "billing", "priority": "p2"}
    if "delivery" in t or "tracking" in t:
        return {"category": "delivery", "priority": "p3"}
    return {"category": "general", "priority": "p3"}

@tool(_description="将工单路由到队列（返回队列ID和SLA小时数）。")
def route_ticket(category: str, priority: str) -> Dict[str, Any]:
    queue_map = {
        "fraud": ("risk_ops", 2),
        "cancellation": ("retention", 8),
        "refunds": ("payments_ops", 12),
        "account_access": ("identity", 12),
        "bug": ("engineering_support", 24),
        "billing": ("billing_support", 24),
        "delivery": ("logistics_support", 48),
        "general": ("support_general", 48),
    }
    q, sla = queue_map.get(category, ("support_general", 48))
    if priority == "p0":
        sla = min(sla, 2)
    elif priority == "p1":
        sla = min(sla, 8)
    return {"queue": q, "sla_hours": sla}

@tool(_description="根据类别和优先级生成剧本响应。")
def draft_response(category: str, priority: str, ticket_text: str) -> Dict[str, Any]:
    templates = {
        "fraud": "我们已临时保护您的账户。请确认最后3笔交易并重置凭据。",
        "cancellation": "我们可以帮助取消您的订阅。请确认您的计划和期望的生效日期。",
        "refunds": "我们正在检查退款状态。请分享订单/付款参考号和日期。",
        "account_access": "让我们帮您重新登录。请使用密码重置链接；如果被阻止，我们将验证身份。",
        "bug": "感谢您的报告。请分享设备/浏览器信息和截图；我们将尝试复现。",
        "billing": "我们可以帮助处理账单问题。请确认最后4位数字和所需发票期间。",
        "delivery": "我们正在检查运输状态。请分享您的跟踪ID和递送地址PIN/ZIP。",
        "general": "感谢您与我们联系。"
    }
    base = templates.get(category, templates["general"])
    tone = "urgent" if priority == "p0" else ("fast" if priority == "p1" else "standard")
    return {
        "tone": tone,
        "message": f"{base}\n\n我们收到的上下文：'{ticket_text}'",
        "next_steps": ["request_missing_info", "log_case", "route_to_queue"]
    }

将注册的工具组合成离线执行流水线：

tool_exec_cfg = ExecutorConfig(
    max_execution_time_ms=10_000,
    max_tool_calls=50,
    continue_on_error=False,
    store_results=True,
    enable_logging=False
)
tool_executor = ToolExecutor(config=tool_exec_cfg)

def offline_triage(ticket: Ticket) -> Dict[str, Any]:
    c = classify_ticket(ticket.text)
    rt = route_ticket(c["category"], c["priority"])
    dr = draft_response(c["category"], c["priority"], ticket.text)
    return {
        "ticket_id": ticket.ticket_id,
        "user_id": ticket.user_id,
        "category": c["category"],
        "priority": c["priority"],
        "queue": rt["queue"],
        "sla_hours": rt["sla_hours"],
        "draft": dr["message"],
        "tone": dr["tone"],
        "steps": [
            ("classify_ticket", c),
            ("route_ticket", rt),
            ("draft_response", dr),
        ]
    }

构建有向GraphBit工作流程，包含多个具有明确定义职责的智能体节点：

SYSTEM_POLICY = "您是一个可靠的支持运营智能体。仅返回严格的JSON。"

workflow = Workflow("工单分类工作流程（GraphBit）")

summarizer = Node.agent(
    name="Summarizer",
    agent_id="summarizer",
    system_prompt=SYSTEM_POLICY,
    prompt="用1-2行总结此工单。返回JSON：{\"summary\":\"...\"}\n工单：{input}",
    temperature=0.2,
    max_tokens=200
)

router_agent = Node.agent(
    name="RouterAgent",
    agent_id="router",
    system_prompt=SYSTEM_POLICY,
    prompt=(
        "您必须使用工具。\n"
        "调用classify_ticket(text)、route_ticket(category, priority)、draft_response(category, priority, ticket_text)。\n"
        "返回包含字段的JSON：category、priority、queue、sla_hours、message。\n"
        "工单：{input}"
    ),
    tools=[classify_ticket, route_ticket, draft_response],
    temperature=0.1,
    max_tokens=700
)

formatter = Node.agent(
    name="FinalFormatter",
    agent_id="final_formatter",
    system_prompt=SYSTEM_POLICY,
    prompt=(
        "验证JSON并仅输出严格JSON：\n"
        "{\"ticket_id\":\"...\",\"category\":\"...\",\"priority\":\"...\",\"queue\":\"...\",\"sla_hours\":0,\"customer_message\":\"...\"}\n"
        "输入：{input}"
    ),
    temperature=0.0,
    max_tokens=500
)

添加可选的LLM配置和执行逻辑，使同一工作流程在提供程序密钥可用时能够自主运行：

def pick_llm_config() -> Optional[Any]:
    if os.getenv("OPENAI_API_KEY"):
        return LlmConfig.openai(os.getenv("OPENAI_API_KEY"), "gpt-4o-mini")
    if os.getenv("ANTHROPIC_API_KEY"):
        return LlmConfig.anthropic(os.getenv("ANTHROPIC_API_KEY"), "claude-sonnet-4-20250514")
    if os.getenv("DEEPSEEK_API_KEY"):
        return LlmConfig.deepseek(os.getenv("DEEPSEEK_API_KEY"), "deepseek-chat")
    if os.getenv("MISTRALAI_API_KEY"):
        return LlmConfig.mistralai(os.getenv("MISTRALAI_API_KEY"), "mistral-large-latest")
    return None

def run_agent_flow_once(ticket_text: str) -> Dict[str, Any]:
    llm_cfg = pick_llm_config()
    if llm_cfg is None:
        return {
            "mode": "offline",
            "note": "设置OPENAI_API_KEY / ANTHROPIC_API_KEY / DEEPSEEK_API_KEY / MISTRALAI_API_KEY以启用执行。",
            "input": ticket_text
        }
    executor = Executor(llm_cfg, lightweight_mode=True, timeout_seconds=90, debug=False)
    wf = Workflow("Single Ticket Run")
    # 节点定义与之前类似，略去重复代码
    # ...
    result = executor.execute(wf)
    return result

在单个工单上执行工作流程并捕获执行状态和输出，这最后一步说明了系统如何从离线确定性无缝过渡到完全智能体执行。

总之，我们实现了一个完整的GraphBit工作流程，涵盖运行时配置、工具注册、离线确定性执行、指标聚合，以及可选的基于智能体的外部LLM提供程序编排。我们展示了相同的业务逻辑如何通过工具手动执行，以及通过验证图中的智能体节点自动执行，突出了GraphBit作为执行底层而非仅仅是LLM包装器的优势。我们证明了复杂的智能体系统可以设计为优雅失败、在没有外部依赖的情况下运行，并在启用LLM时仍然扩展到完全自主的工作流程。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

图计算服务