生产级校园AI Agent架构设计：基于LangGraph的多步骤任务拆解与状态管理深度实践
引言：校园场景是AI Agent最真实的“修罗场”
当用户说出“帮我规划明天的校园生活”时，系统真正面对的并不是一个简单问答任务，而是一条包含查询、推理、推荐、预约和异常处理的复杂业务链路。
例如：
“我明天上午第二节没课，想找一个空研讨间做小组展示排练，排练结束后再推荐一个附近人少、味道不错的食堂，帮我把时间安排好。”
这句话至少包含以下任务：
- 查询课程表，确认明天空闲时间。
- 计算可用时间区间，并预留步行或通勤时间。
- 查询研讨间资源及其开放状态。
- 调用预约接口完成资源预订。
- 检索食堂高峰时段和用户偏好。
- 校验排练、移动和用餐时间是否冲突。
- 在接口失败时进行重试、降级或转人工处理。
如果仍然采用“一次Prompt加一次LLM调用”的方式，系统很容易出现以下问题：
- 模型生成不存在的空闲时间；
- 推荐地点距离当前场所过远；
- 预约接口失败后流程中断；
- 多轮对话中丢失用户的时间和偏好；
- 用户无法知道任务执行到了哪一步。
LLM擅长理解自然语言和进行概率性推理，但它并不适合直接承担鉴权、事务执行、业务校验和异常恢复等职责。
生产级校园AI Agent的核心思想是：
让大模型负责理解和规划，让确定性代码负责执行、验证、持久化和恢复。
LangGraph提供了一个非常适合这种场景的基础：它将Agent组织为可持久化、可循环、可观测的状态图，使复杂任务不再是一次性调用，而是一个可以暂停、恢复、重试和人工接管的执行流程。
一、重新认识Agent State：它不只是消息列表
很多早期Agent只维护如下状态：
{
"messages": []
}
这种设计适合简单聊天，却不足以支撑多步骤业务。对于校园服务来说，系统至少需要知道：
- 用户是谁；
- 当前会话属于哪个任务；
- 原始需求是什么；
- 任务已经拆解成了哪些步骤；
- 当前执行到了哪一步；
- 哪些工具已经调用过；
- 工具返回了什么；
- 哪些数据可以复用；
- 当前是否处于重试或降级状态；
- 最终结果是否已经生成。
1.1 建议的状态分层
一个可落地的State可以划分为六个部分：
| 状态域 | 主要内容 |
|---|---|
| 会话域 | 用户ID、会话ID、消息历史 |
| 规划域 | 原始需求、任务列表、当前任务游标 |
| 执行域 | 工具调用记录、调用参数、返回结果 |
| 缓存域 | 课程表、资源查询、知识库检索结果 |
| 可靠性域 | 重试次数、错误信息、降级等级 |
| 产出域 | 最终回答、终止状态 |
示例定义如下：
from typing import Annotated, Any, Literal, Optional
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
class ToolExecutionRecord(TypedDict, total=False):
tool_name: str
input_params: dict[str, Any]
output_payload: Any
start_time: str
end_time: str
status: Literal["pending", "success", "failed", "timeout"]
error_trace: Optional[str]
retry_count: int
class CampusAgentState(TypedDict, total=False):
# 会话域
user_id: str
session_id: str
messages: Annotated[list, add_messages]
# 规划域
original_user_query: str
decomposed_tasks: list[dict[str, Any]]
current_task_index: int
# 执行域
execution_trace: list[ToolExecutionRecord]
intermediate_cache: dict[str, Any]
# 可靠性域
retry_count: int
last_error: Optional[str]
degradation_level: Literal[
"normal",
"fallback",
"manual_transfer"
]
# 产出域
final_answer: str
is_terminated: bool
这里有几个关键设计点。
第一，messages最好使用LangGraph提供的消息Reducer，而不是简单覆盖列表。这样每个节点返回的新消息可以自动追加到历史中。
第二，工具执行记录应该具有独立结构。它不仅用于调试，也可以被日志系统、审计系统和监控系统消费。
第三，时间字段建议使用字符串或标准化时间格式，而不是直接依赖Python对象。这样更方便进行Checkpoint持久化、序列化和跨服务传输。
第四，重试次数最好同时支持全局统计和任务级统计。一个失败的食堂检索任务，不应该影响其他已经成功的课程查询任务。
二、图拓扑设计：用状态机组织复杂任务
一个典型的校园Agent可以采用如下拓扑：
START
|
Guardrail
|
+-- 直接回答 --------> END
|
Planner
|
Executor
|
Validator
|
+-- 继续执行 --------> Executor
|
+-- 需要重试 --------> Recover
|
+-- 需要澄清 --------> Human/Clarification
|
+-- 全部完成 --------> Finalizer
|
END
其中每个节点都有明确职责：
| 节点 | 类型 | 职责 |
|---|---|---|
| Guardrail | 确定性 | 鉴权、敏感内容拦截、静态问答 |
| Planner | 概率性 | 将自然语言转换为结构化任务计划 |
| Executor | 确定性 | 根据任务类型调用工具或业务API |
| Validator | 确定性 | 验证返回结果是否满足业务规则 |
| Recover | 确定性 | 重试、换方案、降级或人工接管 |
| Finalizer | 确定性 | 汇总结果并生成最终响应 |
LangGraph的价值不只是“把函数连起来”，更重要的是每个节点之间传递的是明确的状态。流程可以在任意节点暂停，也可以从Checkpoint恢复。
三、Guardrail：在进入LLM之前解决确定性问题
校园服务中有大量请求不需要经过大模型，例如校历、校车、考试安排和服务时间查询。
同时，涉及代考、改分、伪造证明等内容的请求，也应该由规则系统直接处理，而不是交给模型自由发挥。
async def guardrail_node(state: CampusAgentState) -> dict:
query = state["original_user_query"]
sensitive_keywords = ["代考", "改分", "伪造成绩"]
if any(keyword in query for keyword in sensitive_keywords):
return {
"final_answer": "该问题涉及校纪校规，请联系教务处或相关管理部门咨询。",
"degradation_level": "manual_transfer",
"is_terminated": True,
}
static_answers = {
"校历": "本学期校历请以教务系统公布的正式版本为准。",
"校车": "校车运行时间请以校园交通服务平台的实时信息为准。",
}
for keyword, answer in static_answers.items():
if keyword in query:
return {
"final_answer": answer,
"is_terminated": True,
}
return {
"degradation_level": "normal",
"is_terminated": False,
}
Guardrail通常还应该承担以下职责：
- 校验用户身份和访问权限；
- 检查接口调用频率；
- 判断用户是否具备预约资格；
- 拦截明显的提示词注入；
- 过滤不允许写入系统的数据；
- 判断当前服务是否处于维护时间。
原则很简单：
能通过规则确定的事情，就不要消耗模型调用。
四、Planner：让LLM负责规划，但不允许它自由发明能力
Planner的作用不是直接执行任务，而是将自然语言转成结构化计划。
例如，用户的需求可以被转换为：
{
"tasks": [
{
"id": "task_1",
"type": "query_schedule",
"description": "查询明日上午的课程安排",
"depends_on": []
},
{
"id": "task_2",
"type": "find_room",
"description": "查找符合时间要求的空研讨间",
"depends_on": ["task_1"]
},
{
"id": "task_3",
"type": "book_room",
"description": "预约选定的研讨间",
"depends_on": ["task_2"]
},
{
"id": "task_4",
"type": "recommend_canteen",
"description": "推荐附近且低拥挤度的食堂",
"depends_on": ["task_3"]
}
]
}
相比简单的字符串任务列表，这种结构能够表达：
- 任务类型；
- 任务依赖；
- 工具名称；
- 输入参数；
- 是否需要用户确认；
- 失败后的备用策略。
可以使用Pydantic约束Planner输出：
from pydantic import BaseModel, Field
from typing import Literal
class PlannedTask(BaseModel):
task_id: str
task_type: Literal[
"query_schedule",
"find_room",
"book_room",
"recommend_canteen",
"clarify"
]
description: str
depends_on: list[str] = Field(default_factory=list)
requires_confirmation: bool = False
class TaskPlan(BaseModel):
tasks: list[PlannedTask]
clarification_question: str | None = None
Planner节点：
async def planner_node(state: CampusAgentState) -> dict:
llm = get_planner_model()
structured_llm = llm.with_structured_output(TaskPlan)
prompt = f"""
你是校园生活服务规划器。
只允许使用以下任务类型：
- query_schedule
- find_room
- book_room
- recommend_canteen
- clarify
不要创造新的工具，不要直接执行预约。
如果信息不足，生成clarify任务。
用户需求：
{state["original_user_query"]}
"""
plan: TaskPlan = await structured_llm.ainvoke(prompt)
return {
"decomposed_tasks": [
task.model_dump()
for task in plan.tasks
],
"current_task_index": 0,
"intermediate_cache": {
"clarification_question": plan.clarification_question
},
"last_error": None,
}
生产环境中，还应在Planner之后增加计划校验器，检查：
- 任务类型是否在白名单内；
- 任务依赖是否形成循环；
- 预约类任务是否包含必要参数；
- 是否存在未经授权的写操作；
- 任务数量是否超过上限；
- 是否需要用户二次确认。
五、规则优先：对抗规划漂移
大模型即使输出符合JSON格式，也不代表计划一定正确。格式正确与业务正确是两件事。
对于高频且稳定的意图，可以先使用规则引擎：
async def robust_planner(state: CampusAgentState) -> dict:
query = state["original_user_query"]
if "课表" in query and "空闲" in query:
tasks = [
{
"task_id": "schedule",
"task_type": "query_schedule",
"description": "查询指定日期的课程安排",
"depends_on": [],
},
{
"task_id": "free_time",
"task_type": "find_room",
"description": "基于空闲时间查找可用研讨间",
"depends_on": ["schedule"],
},
]
return {
"decomposed_tasks": tasks,
"current_task_index": 0,
}
return await planner_node(state)
实际系统可以采用“快慢双轨”策略：
- 规则引擎处理高频、明确和低风险意图。
- 小模型处理常见但需要轻度理解的请求。
- 大模型处理复杂、多步骤和模糊需求。
- 所有计划最终都必须经过Schema和业务规则校验。
这套策略可以降低延迟和Token成本，也能减少模型在常见任务上的不必要发挥。
六、Executor：用工具注册表代替字符串判断
简单示例中常见这样的写法：
tool_name = "query_schedule" if "课表" in task_desc else "book_resource"
这种逻辑很脆弱。只要任务描述稍有变化，就可能路由到错误工具。
更稳妥的方式是使用结构化任务类型和工具注册表：
TOOL_REGISTRY = {
"query_schedule": query_schedule,
"find_room": find_available_rooms,
"book_room": book_room,
"recommend_canteen": recommend_canteen,
}
Executor只根据task_type选择工具：
async def executor_node(state: CampusAgentState) -> dict:
index = state["current_task_index"]
tasks = state["decomposed_tasks"]
if index >= len(tasks):
return {"is_terminated": True}
task = tasks[index]
task_type = task["task_type"]
tool = TOOL_REGISTRY.get(task_type)
if tool is None:
return {
"last_error": f"未注册的任务类型：{task_type}",
"degradation_level": "manual_transfer",
}
try:
result = await call_with_timeout(
tool,
task,
timeout_seconds=3
)
return {
"intermediate_cache": {
**state.get("intermediate_cache", {}),
task["task_id"]: result,
},
"last_error": None,
}
except TimeoutError:
return {
"last_error": "工具调用超时",
}
except Exception as exc:
return {
"last_error": str(exc),
}
这里需要注意一个重要问题：工具调用与状态更新最好具有清晰的事务边界。
对于查询类操作，可以安全重试；对于预约、取消预约、缴费等写操作，则必须考虑幂等性。否则，第一次请求可能已经成功，只是响应丢失，第二次重试就可能产生重复预约。
解决方式包括：
- 为每次写操作生成幂等键；
- 在调用前检查是否已经存在相同业务请求；
- 对接口返回结果进行最终一致性查询；
- 区分“请求未发出”“请求已发出但超时”和“明确业务失败”。
七、Validator：HTTP成功不等于业务成功
很多系统只判断HTTP状态码：
if response.status_code == 200:
return "success"
但校园系统经常出现以下情况：
{
"code": 0,
"message": "success",
"data": null
}
这在协议层面可能是成功，在业务层面却完全不可用。
因此，Validator必须验证返回数据是否满足业务约束。
def validate_tool_result(
task_type: str,
result: dict
) -> tuple[bool, str | None]:
if result.get("code") != 0:
return False, result.get("message", "接口返回失败")
data = result.get("data")
if task_type == "query_schedule":
if not data or not isinstance(data.get("courses"), list):
return False, "课程表数据为空或格式错误"
if task_type == "find_room":
rooms = data.get("rooms", []) if data else []
if not isinstance(rooms, list):
return False, "研讨间列表格式错误"
if task_type == "book_room":
if not data or not data.get("reservation_id"):
return False, "预约成功但缺少预约编号"
return True, None
Validator还可以检查现实世界中的约束：
- 预约开始时间必须早于结束时间；
- 资源容量必须满足小组人数；
- 房间必须处于开放状态；
- 食堂距离不能超过可接受范围；
- 推荐时间不能与课程或预约冲突；
- 返回的预约时间不能落在系统维护窗口内。
这一步是杜绝“表面成功”的关键。
八、条件路由：重试、恢复和降级
LangGraph允许根据当前State选择下一条边。一个简单的路由函数如下：
from typing import Literal
def route_after_validation(
state: CampusAgentState
) -> Literal["executor", "recover", "finalizer"]:
if state.get("last_error"):
retry_count = state.get("retry_count", 0)
if retry_count < 2:
return "recover"
return "recover"
index = state["current_task_index"]
total = len(state["decomposed_tasks"])
if index + 1 >= total:
return "finalizer"
return "executor"
恢复节点不应该只做一件事：重复调用同一个接口。
更合理的恢复策略包括：
- 使用指数退避重试；
- 缩小查询范围；
- 切换备用接口；
- 使用缓存结果；
- 跳过非关键任务；
- 请求用户补充信息；
- 转人工处理。
例如：
async def recover_node(state: CampusAgentState) -> dict:
retry_count = state.get("retry_count", 0) + 1
error = state.get("last_error", "")
if "超时" in error and retry_count <= 2:
return {
"retry_count": retry_count,
"last_error": None,
}
if "没有可用研讨间" in error:
return {
"degradation_level": "fallback",
"last_error": None,
"intermediate_cache": {
**state.get("intermediate_cache", {}),
"fallback_message": "当前没有符合条件的研讨间，可尝试调整时间或地点。",
},
}
return {
"degradation_level": "manual_transfer",
"final_answer": "当前预约服务暂时不可用，请稍后重试或联系服务台。",
"is_terminated": True,
}
需要特别注意，重试计数不能只使用一个全局变量。更好的做法是：
{
"retry_count": 1,
"task_retry_count": {
"query_schedule": 0,
"book_room": 1
}
}
这样可以避免某个失败任务消耗掉整个流程的重试额度。
九、构建LangGraph工作流
将上述节点连接起来：
from langgraph.graph import StateGraph, START, END
builder = StateGraph(CampusAgentState)
builder.add_node("guardrail", guardrail_node)
builder.add_node("planner", robust_planner)
builder.add_node("executor", executor_node)
builder.add_node("validator", validator_node)
builder.add_node("recover", recover_node)
builder.add_node("finalizer", finalizer_node)
builder.add_edge(START, "guardrail")
builder.add_conditional_edges(
"guardrail",
route_after_guardrail,
{
"planner": "planner",
"end": END,
}
)
builder.add_edge("planner", "executor")
builder.add_edge("executor", "validator")
builder.add_conditional_edges(
"validator",
route_after_validation,
{
"executor": "executor",
"recover": "recover",
"finalizer": "finalizer",
}
)
builder.add_conditional_edges(
"recover",
route_after_recover,
{
"executor": "executor",
"finalizer": "finalizer",
"end": END,
}
)
builder.add_edge("finalizer", END)
graph = builder.compile()
真正上线时，还应配置Checkpoint存储，使流程能够在以下情况下恢复：
- 服务进程重启；
- 工具调用需要人工确认；
- 用户暂时离开；
- 预约操作等待异步回调；
- 多步骤任务执行时间较长。
Checkpoint的本质，是把Agent从“一次函数调用”升级为“可恢复业务流程”。
十、记忆管理：解决长对话中的上下文丢失
校园助手往往需要处理多轮对话：
“帮我找明天下午的研讨间。”
“人数是六个人。”
“不要太远，最好靠近东区。”
“预约前先告诉我具体时间。”
如果每轮都携带完整历史，Token成本会持续增加；如果简单截断，又容易丢失关键条件。
可以采用分层记忆：
短期记忆
保存当前任务直接相关的消息，例如：
- 当前日期；
- 人数；
- 地点偏好；
- 时间限制；
- 用户最近一次确认。
摘要记忆
将较早的对话压缩成结构化摘要：
{
"user_preferences": {
"preferred_zone": "东区",
"max_walking_minutes": 10,
"group_size": 6
},
"confirmed_constraints": [
"预约前必须先展示时间和地点",
"不接受跨校区资源"
]
}
长期记忆
经过用户授权后，保存稳定偏好，例如常用校区、饮食偏好和常用资源。
记忆系统必须明确区分：
- 用户本轮临时要求；
- 已确认的长期偏好；
- 推测出来但尚未确认的信息。
未经用户确认的推测，不应直接作为预约参数。
十一、事务性操作：预约前必须确认，写入后必须核验
查询和推荐通常是低风险操作，预约、取消、缴费等则属于高风险写操作。
对于这类操作，建议采用如下流程：
生成预约方案
|
展示房间、时间、地点和费用
|
请求用户确认
|
生成幂等键
|
调用预约接口
|
查询最终预约状态
|
返回预约编号
不能仅凭接口超时就认为预约失败，也不能仅凭HTTP 200就认为预约成功。
正确的判断应当包含：
- 是否生成业务订单号；
- 是否返回资源编号；
- 是否查询到最终状态；
- 是否与用户确认的参数一致；
- 是否在重复请求中保持幂等。
这也是Agent系统区别于普通聊天机器人的重要地方：它必须对现实世界中的状态变化负责。
十二、全链路可观测性：让每一次失败都能被解释
没有观测数据，就无法知道Agent为什么失败。
建议为每次任务生成唯一的：
request_idsession_idtrace_idtask_idtool_call_id
并记录：
- 节点开始和结束时间；
- 模型耗时和Token消耗；
- 工具名称和参数摘要；
- 外部接口耗时；
- 重试次数；
- 校验失败原因；
- 降级等级；
- 最终用户可见结果。
OpenTelemetry示例：
from opentelemetry import trace
tracer = trace.get_tracer("campus.agent")
async def traced_executor(state: CampusAgentState) -> dict:
task = state["decomposed_tasks"][
state["current_task_index"]
]
with tracer.start_as_current_span("agent.executor") as span:
span.set_attribute("task.type", task["task_type"])
span.set_attribute("task.id", task["task_id"])
try:
result = await executor_node(state)
span.set_attribute("result.status", "success")
return result
except Exception as exc:
span.record_exception(exc)
span.set_attribute("result.status", "failed")
raise
监控指标可以包括：
- 任务整体成功率；
- 各工具成功率；
- 平均执行时长；
- P95和P99延迟；
- Planner计划校验失败率；
- Validator业务失败率；
- 重试率；
- 人工接管率；
- 单次任务平均成本。
例如，如果发现预约接口在每天21点后的失败率明显升高，就可以增加服务时间前置校验，而不是让用户反复等待重试。
十三、安全边界：状态管理也必须遵循最小权限
校园Agent通常会接触课程表、预约记录、联系方式甚至缴费信息，因此不能只关注流程正确性，还必须关注数据安全。
建议遵循以下原则：
最小权限
不同工具使用不同权限：
- 查询课程表只能读取本人数据；
- 查询房间可以读取公共资源；
- 预约资源需要额外写权限；
- 取消预约需要二次确认。
敏感数据脱敏
日志中不应直接记录：
- 身份证号码；
- 完整手机号；
- 访问令牌；
- 详细个人隐私；
- 未经授权的课程信息。
工具参数白名单
不允许模型直接拼接任意URL、SQL或系统命令。所有工具都应由服务端注册，并对参数进行类型和范围校验。
用户确认
涉及以下操作时，必须让用户明确确认：
- 预约；
- 取消；
- 缴费；
- 提交材料；
- 修改个人信息；
- 向第三方发送信息。
十四、生产落地检查清单
在上线前，可以从以下几个方面进行验收：
状态设计
- State是否可以完整描述任务进度？
- 状态是否可以序列化和恢复？
- 是否区分临时数据、缓存数据和长期记忆？
- 是否记录了任务级执行轨迹？
规划可靠性
- Planner是否使用结构化输出？
- 工具和任务类型是否有白名单？
- 是否校验任务依赖和参数完整性？
- 是否有规则引擎处理高频意图？
工具执行
- 查询和写操作是否区分？
- 写操作是否具备幂等能力？
- 是否有超时、重试和备用接口？
- 是否对业务返回结果进行校验？
异常恢复
- 是否支持任务级重试？
- 是否能够识别可恢复和不可恢复错误？
- 是否提供明确的降级结果？
- 是否支持人工接管？
可观测性
- 是否可以根据Trace还原完整流程？
- 是否记录了模型、工具和业务校验耗时？
- 是否能够定位失败节点？
- 是否建立了成功率、延迟和成本指标？
用户体验
- 是否会在高风险写操作前请求确认？
- 是否能告诉用户当前执行进度？
- 是否能解释失败原因？
- 是否避免把内部异常堆栈直接展示给用户？
结语：Agent的核心竞争力是确定性兜底
校园AI Agent的难点，从来不只是如何调用一个更强的大模型，而是如何让模型参与复杂业务时仍然保持可控、可解释和可恢复。
一套可靠的校园Agent应当遵循三条原则：
- 能通过规则解决的问题，不交给LLM。
- 能通过确定性校验确认的结果，不相信模型的自我判断。
- 任何可能改变现实状态的操作，都必须具备权限控制、幂等机制和结果核验。
LangGraph的价值，正在于它把这些工程原则落实为可执行的状态图：Guardrail负责拦截，Planner负责规划，Executor负责执行，Validator负责验证，Recover负责恢复，Checkpoint负责持久化，Trace负责解释全过程。
在这套架构中，大模型不再是一个不可控的黑盒，而是被放置在明确的边界之内。它负责理解复杂需求、生成候选计划和组织自然语言；而真正涉及权限、数据、事务和现实世界状态的部分，则由确定性的工程代码接管。
这才是校园AI Agent走向生产环境的关键：
用图结构管理复杂流程，用强类型State承载上下文，用业务规则约束模型，用可观测性保障结果，用确定性工程驾驭概率性智能。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
14
0- 0
已为社区贡献3条内容
所有评论(0)