利用LangGraph和Ollama实现多智能体工单分类处理

凡是负责处理支持邮件箱的人都知道早晨的常规工作流程：有人会阅读每一封新收到的工单，将密码重置和账单相关的问题分类整理，同时找出那些真正紧急的消息。这完全是机械化的操作，正是由一小队协同工作的客服人员可以承担的任务。

接下来将通过 LangGraph 来组建这样的团队。一名客服人员负责对每封工单进行分类，路由器决定后续处理方式，另一名客服人员则负责起草回复。紧急案件会跳过自动化流程，直接交由人工处理。该图结构只需编写一次，即可在 Ollama 上本地运行，也可通过云 API 运行，而不会对客服人员或路由机制造成任何改动。

为何需要多名客服人员

单个智能体就像从不与任何人交流的工人，在真正的公司中，这样的架构很快就会停止发展。一旦增加更多任务，单个指令就会变得冗长、脆弱且难以调试，因为需要同一个模型同时完成分类、决策、写作和风险评估等工作。将任务拆分后，每部分都会保持较小规模，便于测试和替换。多智能体系统则类似一支由各司其职的专家组成的团队，共同协调下一步行动。

工单分拣很符合这一理念，因为相关工作天然会分化，而如果对每条消息都采用相同的处理方式，至少在以下情况中会出问题：

• 常规的账单问题可以自动回答。
• 对于模糊的功能请求，可以先生成初步回复，再由人工进行完善。

像“我的应用程序已停机”这样的消息根本不应由自动化系统处理，而应立即转交给人工处理。

LangGraph正是处理这类需要根据当前状态决定下一步行动的场景。该团队共有三名成员：

1. 分类代理，负责读取工单并为其分配类别和优先级。
2. 回复代理，为那些可以自动处理的工单起草回复内容。
3. 人工审核节点，负责将紧急工单转交给人手处理，而非自行回复。

在它们之间是路由逻辑，相当于一个监督者，它会根据优先级选择处理路径。

LangGraph的思维方式

想象一下白板上的流程图。LangGraph由四个部分组成：

• 状态是在各个节点之间传递的剪贴板数据，其中包含了目前关于该任务的所有信息：原始文本、类别、优先级、已拟定的回复内容（如有）以及当前状态。
• 节点即某个处理站，它接收状态数据，执行相应操作后返回更新后的状态。
• 边则是从一个节点指向另一个节点的固定箭头。
• 条件边是在运行时动态选择的箭头，此时会由一个小型路由函数读取当前状态并决定下一步的走向，而不是一直沿用固定的路径。

状态在节点间流动，边将节点连接起来，而条件边则允许路径根据当前状态的不同而分支。对于初学者来说，状态其实就是一个具有固定结构的 Python 字典，TypedDict 用于指定该字典应包含的键。

技术架构

LangGraph 是调度层，负责定义节点、状态以及各组件之间的路由方式。

Ollama with Llama 3 可让所有功能在您的本地设备上免费运行。由于支持工单通常包含用户的个人数据，因此在企业环境中，将这类数据保存在自有基础设施上而非发送至外部 API，具有显著优势。

API 模型，即这里的 gpt-4o-mini，适用于没有多余硬件或需要比小型本地模型更高性能的场景。由于图结构只需编写一次，因此可以先在本地运行，之后再切换到 API，而无需重新编写逻辑。

逐步构建

1. 设置

需要安装三个软件包：用于处理图结构的 langgraph，作为与 Ollama 通信的桥梁的 langchain-ollama，以及作为与 API 模型通信的桥梁的 langchain-openai。

# Install LangGraph and the two model integrations
pip install langgraph langchain-ollama langchain-openai

一个全新的虚拟环境可以确保该项目的软件包不会与您的其他项目混淆。在 Windows 系统上，使用 python -m venv yourname 创建虚拟环境，再通过其中的 Scripts 文件夹激活它。在 Mac 或 Linux 系统上，则先执行 python3 -m venv venv，然后运行 source venv/bin/activate。一旦虚拟环境被激活，上述安装命令就只会影响当前项目。

如果尚未安装 Ollama，可从 ollama.com/download 获取。之后通过终端拉取本地模型：

# Download the Llama 3 model so we can run it locally
ollama pull llama3

2. 代理程序

从现在开始，所有内容都会被放入一个名为 triage.py 的文件中，并按照以下顺序从上到下添加。

模型工厂。首先创建一个能让两种后端都正常工作的简单工厂函数 get_llm，它根据某个设置返回相应的模型。工厂的唯一职责就是返回一个已准备好的对象。可以把它想象成一台咖啡机，上面有一个标有“本地”字样的按钮，另一个标有“API”字样：按下其中一个按钮就能得到对应的咖啡，无需了解其内部工作原理。

该函数接受一个名为 backend 的参数，其默认值为 "local"，因此不带参数调用时就会使用本地模型。

# A factory that returns the model we want, so the rest of the code
# never has to care which backend we are using.
def get_llm(backend: str = "local"):
    if backend == "local":
        from langchain_ollama import ChatOllama
        # temperature=0 keeps classification stable and repeatable
        return ChatOllama(model="llama3", temperature=0)
    elif backend == "api":
        from langchain_openai import ChatOpenAI
        return ChatOpenAI(model="gpt-4o-mini", temperature=0)
    else:
        raise ValueError("backend must be 'local' or 'api'")


# Choose your backend here: "local" or "api"
BACKEND = "local"
llm = get_llm(BACKEND)

if/elif/else 语句链会检查参数，然后返回 llama3、gpt-4o-mini 或错误信息。这些导入语句特意放在各个分支中——仅在本地工作的用户永远不需要 langchain-openai，反之亦然。

temperature=0 用于控制模型回答的随机性。数值越高，回答的多样性越大；而设置为 0 则能使回复尽可能稳定且可预测。这对于分类任务来说非常合适，因为相同的工单应该始终被归入同一类别。BACKEND = "local" 用于指定后端类型，下一行代码则会调用相应的生成函数。

分类器：该组件负责读取工单并返回对应的类别及优先级。为了能够以编程方式处理结果，模型会被要求以 JSON 格式输出答案。

parse_json辅助函数会将回复转换为真正的Python字典。它首先使用find来定位开头和结尾的大括号，因为本地模型并不总是能返回格式完整的JSON，它们往往会加上诸如“好的，这是JSON内容：……希望这对你有帮助。”之类的冗余文字。

classify_ticket函数就是代理本身。它接收当前状态并返回更新结果。其提示词明确规定了模型作为客服分诊助手的角色，要求答案限定在四类、三种优先级范围内，并且必须为纯JSON格式，不得包含任何额外注释。由于提示词设计得相当严格，因此返回的回复更为整洁，parse_json需要处理的任务也就更少了。调用llm.invoke(prompt)即可将提示词发送给模型。

注意返回的内容：只有该代理填写的两个字段，即类别和优先级。这是 LangGraph 的核心特性。节点不会返回全部状态，而只返回它想要修改的部分，LangGraph 会将这些部分合并到共享状态中。这样能使节点保持简洁，便于单独测试，同时也能避免某个代理意外覆盖属于另一个代理的字段。

import json

def parse_json(text: str) -> dict:
    # Local models sometimes wrap their JSON in extra text,
    # so we grab everything between the first { and the last }.
    start = text.find("{")
    end = text.rfind("}") + 1
    return json.loads(text[start:end])

def classify_ticket(state: dict) -> dict:
    prompt = f"""You are a support triage assistant.
Classify the following support ticket.
Respond with ONLY a JSON object with two keys:
"category": one of ["billing", "technical", "account", "general"]
"priority": one of ["low", "medium", "high"]

Ticket:
{state['ticket']}"""

    response = llm.invoke(prompt)
    result = parse_json(response.content)

    # We return only the keys we want to update in the shared state
    return {"category": result["category"], "priority": result["priority"]}

回复者。第二位智能体负责为那些适合自动化处理的工单起草回复。draft_reply函数会获取当前状态，构建一个提示词，将模型设定为友善且简洁的客服人员，并设置相应限制：不得做出无法兑现的承诺。该函数会返回两个字段，即draft_reply中的草拟文本以及“自动起草”的状态，这样你就能知道该工单是经过自动化处理而非转交人工处理的。这个提示词的设计刻意保持谦逊——由模型撰写的客服回复应当简短且诚实，至于退款时间或功能上线日期之类的承诺，则应由人工来决定。

def draft_reply(state: dict) -> dict:
    prompt = f"""You are a friendly and concise support agent.
Write a short, helpful reply to this {state['category']} ticket.
Do not make promises you cannot keep.

Ticket:
{state['ticket']}"""

    response = llm.invoke(prompt)

    return {"draft_reply": response.content, "status": "auto-drafted"}

人工处理环节。第三类代理从不调用模型，它仅将工单标记为已升级，以便由人工处理。它会返回一个空的draft_reply，因为不应存在自动回复，同时状态会显示为“已升级至人工审核”。这样就能明确表明该工单由人工负责。在图中，节点无需使用大语言模型才能被视为代理——任何能够获取状态、执行任务并返回更新内容的实体都可以存在于同一框架中，无论它是调用模型、访问数据库还是完全不采取任何操作。

def escalate_to_human(state: dict) -> dict:
    # No auto-reply for urgent tickets. We flag them for a person.
    return {
        "draft_reply": "",
        "status": "escalated to human review",
    }

3. 图表的连接方式

这三个智能体各自都有任务。现在需要将它们连接起来，而第一步就是描述共享状态的结构。TicketState类就是之前提到的剪贴板，用于声明存在的字段及其类型。

这五个字段还记录了工单在系统中的处理路径。ticket保存原始文本，分类器负责填写category和priority，回复者则填写draft_reply，或在升级处理后将该字段留空。status则记录工单最终的处理结果。

from typing import TypedDict


class TicketState(TypedDict):
    ticket: str        # the original ticket text
    category: str      # filled in by the classifier
    priority: str      # filled in by the classifier
    draft_reply: str   # filled in by the responder (or cleared on escalation)
    status: str        # where the ticket ended up

接下来是路由器，也就是监管者。route_by_priority会读取分类器所设定的优先级，然后返回一个字符串告知LangGraph应选择哪条路径。如果优先级为“高”，工单就会交给人工处理；否则该函数会返回“responder”，工单将获得自动回复。

def route_by_priority(state: TicketState) -> str:
    # Urgent tickets skip the automatic reply and go to a human.
    if state["priority"] == "high":
        return "human_review"
    return "responder"

现在来构建图表：

1. 使用 StateGraph(TicketState) 创建一个空图表，并传入状态结构，这样图表就能知道剪贴板上包含哪些字段。
2. 通过 add_node 注册三个智能体。每次调用都会将一个简短名称与一个函数关联起来。“classifier”这一名称对应 classify_ticket 函数，“responder”对应 draft_reply 函数，而“human_review”对应 escalate_to_human 函数。后续的所有步骤都会用这些名称来引用节点，因此请谨慎选择并保持一致性。
3. 添加 add_edge(START, “classifier”)，这样每个工单都会从分类器开始处理。START 是一个内置的起始点。

使用 add_conditional_edges 函数添加条件边，该函数需要分支发生的节点（“classifier”）、路由函数（route_by_priority）以及一个查找表。

将两条路径都发送到 END，因为无论工单是否得到处理或升级，流程都已结束。

调用 graph.compile() 函数，将定义转换为可执行的形式。其结果会被存储在 app 中，而正是该部分负责处理工单。

from langgraph.graph import StateGraph, START, END

# Create the graph and tell it the shape of our state
graph = StateGraph(TicketState)

# Register each agent as a node
graph.add_node("classifier", classify_ticket)
graph.add_node("responder", draft_reply)
graph.add_node("human_review", escalate_to_human)

# Every ticket starts at the classifier
graph.add_edge(START, "classifier")

# After classifying, the router decides where to go next
graph.add_conditional_edges(
    "classifier",
    route_by_priority,
    {
        "responder": "responder",       # normal tickets get an auto-draft
        "human_review": "human_review",  # urgent tickets get escalated
    },
)

# Both paths finish the run
graph.add_edge("responder", END)
graph.add_edge("human_review", END)

# Compile turns our definition into something runnable
app = graph.compile()

传递给 add_conditional_edges 的字典即为查找表。其键是路由器可以返回的字符串，值则是这些字符串所指向的节点。当路由器返回“human_review”时，LangGraph会跳转到 human_review 节点。这种映射关系正是图形分支机制的核心。日后若要添加新的处理类别，比如账单处理专员，只需新增一个节点、表中的一个新键以及路由器中的一条分支即可。

4. 运行方式：本地或 API

首先通过该图形结构发送紧急工单：

# A clearly urgent billing issue
urgent_ticket = (
    "I was charged twice for my subscription this month "
    "and I need a refund today. This is unacceptable."
)

result = app.invoke({"ticket": urgent_ticket})

print("Category:", result["category"])
print("Priority:", result["priority"])
print("Status:  ", result["status"])
print("Reply:   ", result["draft_reply"])

使用 python triage.py 运行该脚本。虽然每次运行以及不同模型生成的回复文字可能略有差异，但类别、优先级和状态应保持一致，这正是设置 temperature=0 的原因。由于该工单的优先级较高，路由器会将其转交给人类处理人员。系统返回的状态为“已升级至人工审核”，且不会生成自动回复。

再试一个普通的工单：

# A routine, low-stakes question
routine_ticket = "How do I change the email address on my account?"

result = app.invoke({"ticket": routine_ticket})

print("Category:", result["category"])
print("Priority:", result["priority"])
print("Status:  ", result["status"])
print("Reply:   ", result["draft_reply"])

该流程会将任务传递给响应器，返回一份草拟的回复，供人工在发送前审核。同时运行这两个任务可以直观地展示分支差异：图表相同、代码相同，仅因分类器所设定的优先级不同而选择两条不同的处理路径。

切换到 API 模型

由于有了工厂函数，只需一个 API 密钥以及仅修改的一行代码，即可通过 API 而非本地方式执行相同功能。

登录 OpenAI API 平台，创建一个新的密钥并立即复制下来，因为该密钥仅显示一次。为确保安全，请勿将密钥直接粘贴到脚本中，而应将其作为环境变量存储，让代码从该位置读取。在终端中运行适用于您系统的命令，将 your-key-here 替换为实际密钥即可：

# On Windows (PowerShell)
setx OPENAI_API_KEY "your-key-here"

# On Mac/Linux
export OPENAI_API_KEY="your-key-here"

请在终端中运行这些命令，而非脚本中，最好是在虚拟环境已激活的同一终端里。该变量是在终端会话层面设置的，而非在venv内部。之后，这个键值会存储在代码之外的某个位置，不会出现在脚本中，也不会意外被上传到GitHub上。

接着修改脚本中的一行，再运行python triage.py即可：

# Switch from local to API without touching the graph or the agents
BACKEND = "api"
llm = get_llm(BACKEND)

展示内容及后续操作方向

该示例模拟了真实的日常工作流程。其目的并非取代支持团队，而是承担起常规分类任务，而重要且紧急的案例则由人工处理。由于该系统中的图表、代理及路由功能既支持 Ollama 也支持 API，因此可以先用 Ollama 进行原型开发，当需要更高性能时再切换到 API 模型。

该系统的扩展空间很大。代理可以拥有内存功能，这样同一客户的后续工单就能保留上下文信息。还可以引入更多专业人员，比如能够访问真实账户数据的账单处理人员。该系统的图表可通过 API 与实际的工单系统相连，从而使脚本成为完整工作流程的一部分。此外，还可以设置适当的人工审核环节，将升级后的工单放入已标注分类信息的审核队列中。

如需进一步了解，可参阅 https://docs.langchain.com/oss/python/langgraph/overview 上的 LangGraph 和 LangChain 文档，以及 https://docs.ollama.com/quickstart 上的 Ollama 快速入门指南，其中介绍了此处使用的工具。