摘要

传统在线 OJ 系统通常只返回 AC / WA / TLE / MLE / RE / CE 这类判题状态。这个结果对系统来说是明确的，但对学生来说还不够友好：他可能知道“错了”，却不知道为什么错、哪里有风险、应该复习什么知识点。
我的 OJ 项目原本已经具备 Java 代码提交、Docker 沙箱判题、RabbitMQ 异步判题、结果持久化等能力。后续我在不改动原 Java 判题主链路的前提下，增量接入了一个 Python FastAPI + LangGraph AI 判题增强服务。
这套 AI 服务的核心原则是：
AI 不替代 OJ 判题，AI 只增强分析、解释和学习反馈。
也就是说：
Java OJ + Docker 沙箱仍然是唯一可信的代码执行和判题来源；
Python AI 服务不执行用户代码；
LangGraph 只负责编排多个 Agent；
AI 负责静态扫描、安全风控、RAG 检索、用户记忆、代码点评和实时状态推送；
LLM 不可用时，系统仍能通过规则化解释降级返回。
本文会从技术细节出发，总结 LangGraph 多 Agent AI 判题增强链路的设计。

为什么 OJ 需要 AI 增强？

1. 传统 OJ 的判题结果通常是这样的：

• AC 通过
• WA 答案错误
• TLE 超时
• MLE 内存超限
• RE 运行时错误
• CE 编译错误
  这些状态对平台是够用的，但对用户学习并不够。
  比如用户提交一段 Java 代码后收到 WA，他可能还想知道：
  是边界条件错了，还是输出格式错了？
  复杂度有没有问题？
  有没有使用高风险 API？
  如果经常写出类似错误，应该重点补什么？
  有没有相关知识点或题解方向可以参考？
  这正是 AI 增强层适合做的事情。
  但这里有一个重要边界：AI 不能替代判题。
  OJ 的判题结果必须是确定性的，必须由真实测试用例、真实代码执行、真实 Docker 沙箱资源限制决定。LLM 可能幻觉，RAG 可能召回不准，Prompt 也可能不稳定，所以 AI 只能做辅助解释，不能决定最终 AC / WA。
  因此我采用的设计是：
  Java OJ：负责确定性判题
  Python AI：负责增强解释、风控、RAG、记忆和实时反馈
  整体架构
  AI 判题增强服务是一个独立 Python 微服务，通过 HTTP 接入原有 Spring Boot OJ。

原 Java OJ 继续负责：
用户、题目、考试、提交等业务逻辑；
Java 代码动态编译执行；
Docker 沙箱隔离；
CPU / 内存 / 超时限制；
MySQL / Redis 数据存储；
普通判题链路的稳定运行。
Python AI 服务负责：
LangGraph 多 Agent 编排；
静态扫描和安全风控前置；
调用 Java OJ 原生判题接口；
解析 Java OJ 判题结果；
RAG 题解和知识点检索；
用户历史做题记忆；
AI 代码点评；
SSE 实时推送 Agent 状态；
Redis + SQLite 保存完整 AI 判题结果。
这个架构最大的好处是：AI 能力作为旁路增强独立演进，不会破坏原来的稳定判题链路。

为什么选择 LangGraph

如果只是“调用一次大模型生成点评”，其实一个普通 FastAPI 接口就够了。
但 OJ AI 判题增强不是单次问答，它天然包含多个步骤：
标准化用户提交上下文；
静态扫描危险 API；
读取用户历史记忆；
检索题解和知识点；
汇合并行结果；
做安全风控决策；
必要时调用 Java OJ；
解析判题结果；
生成 AI 解释；
更新用户记忆；
推送每个阶段的实时状态。
这些步骤有串行依赖，也有并行分支，还有条件路由。用普通函数硬写也能做，但很容易变成一大坨流程代码。
LangGraph 更适合这种场景：
用 StateGraph 表达工作流；
用统一 JudgeState 传递上下文；
每个 Agent 只负责一个职责；
支持条件边和并行分支；
支持 RetryPolicy；
配合 SSE 可以把节点状态实时推给前端。

全局状态体 JudgeState

整个工作流围绕 JudgeState 传递数据。它既包含原始提交信息，也包含各个 Agent 的输出。
简化后的状态结构如下：

class JudgeState(TypedDict, total=False):
    trace_id: str
    user_id: int
    question_id: int
    exam_id: int | None
    program_type: int
    language: str
    user_code: str
    difficulty: int
    time_limit: int
    space_limit: int
    input_list: list[str]
    output_list: list[str]

    stage: JudgeStage
    error_message: str | None

    context: ContextInfo
    static_scan: StaticScanResult
    security_risk: SecurityRiskResult
    java_judge_payload: JavaOjJudgePayload
    java_oj_result: JavaOjJudgeResult
    result_parse: ResultParseResult
    ai_explain: AiExplainResult
    user_memory: UserMemory
    rag_context: RagContext

    should_reject: bool
    should_call_java_oj: bool

    agent_logs: Annotated[list[AgentLog], operator.add]
    metrics: Annotated[list[AgentMetric], operator.add]
    parallel_completed: Annotated[list[str], operator.add]'

这里有几个设计点：
trace_id：贯穿整条 AI 判题链路，方便日志、SSE、结果查询。
stage：标记当前执行阶段，例如 STATIC_SCANNED、REJECTED、DONE。
static_scan / security_risk / rag_context / user_memory：每个 Agent 只写自己负责的字段。
agent_logs / metrics：使用 operator.add 追加，保留完整链路轨迹。
should_reject / should_call_java_oj：安全风控后用于条件路由。
这样做的好处是，每个 Agent 都像一个小型纯业务模块：输入是 JudgeState，输出是状态增量。
LangGraph 工作流实现
核心编排代码在 workflow.py：

def build_judge_graph():
    workflow = StateGraph(JudgeState)

    retry_policy = RetryPolicy(
        max_attempts=settings.agent_max_retry + 1,
        initial_interval=0.2,
        backoff_factor=2.0,
        max_interval=2.0,
        jitter=True,
    )

    workflow.add_node("context", _with_timeout(context_agent), retry_policy=retry_policy)
    workflow.add_node("static_scan", _with_timeout(static_scan_agent), retry_policy=retry_policy)
    workflow.add_node("memory", _with_timeout(memory_agent), retry_policy=retry_policy)
    workflow.add_node("rag", _with_timeout(rag_agent), retry_policy=retry_policy)
    workflow.add_node("parallel_join", _with_timeout(parallel_join_agent), retry_policy=retry_policy)
    workflow.add_node("security_risk", _with_timeout(security_risk_agent), retry_policy=retry_policy)
    workflow.add_node("java_oj_judge", _with_timeout(java_oj_judge_agent), retry_policy=retry_policy)
    workflow.add_node("result_parse", _with_timeout(result_parse_agent), retry_policy=retry_policy)
    workflow.add_node("ai_explain", _with_timeout(ai_explain_agent), retry_policy=retry_policy)
    workflow.add_node("persist_memory", _with_timeout(persist_memory_agent), retry_policy=retry_policy)

    workflow.add_edge(START, "context")
    workflow.add_conditional_edges("context", _fanout_after_context, ["static_scan", "memory", "rag"])
    workflow.add_edge(["static_scan", "memory", "rag"], "parallel_join")
    workflow.add_edge("parallel_join", "security_risk")

    workflow.add_conditional_edges(
        "security_risk",
        _route_after_security_risk,
        {
            "reject": "result_parse",
            "judge": "java_oj_judge",
        },
    )

    workflow.add_edge("java_oj_judge", "result_parse")
    workflow.add_edge("result_parse", "ai_explain")
    workflow.add_edge("ai_explain", "persist_memory")
    workflow.add_edge("persist_memory", END)

    return workflow.compile()

可以看到，整个流程不是简单串行，而是串并行混合：
节点编排：
START
-> ContextAgent
-> 并行分支
-> StaticScanAgent
-> MemoryAgent
-> RagAgent
-> ParallelJoinAgent
-> SecurityRiskAgent
-> 条件分支
-> Reject: ResultParseAgent
-> Allow: JavaOjJudgeAgent -> ResultParseAgent
-> AiExplainAgent
-> MemoryPersistAgent
-> END
并行分支的代码：

def _fanout_after_context(state: JudgeState) -> list[Send]:
    return [
        Send("static_scan", state),
        Send("memory", state),
        Send("rag", state),
    ]

安全风控后的条件路由：

def _route_after_security_risk(state: JudgeState) -> Literal["reject", "judge"]:
    if state.get("should_reject"):
        return "reject"

    if state.get("should_call_java_oj", True):
        return "judge"

    return "reject"

这就是 LangGraph 在项目里的核心价值：用图结构清晰表达 Agent 的执行顺序、并行关系和条件分支。
Agent 超时、重试和实时事件
每个 Agent 都被 _with_timeout 包了一层：

async def guarded(state: JudgeState) -> dict[str, Any]:
    trace_id = state.get("trace_id", "")
    agent_name = _agent_display_name(agent)

    await event_hub.publish(
        trace_id,
        "agent_started",
        {
            "agent_name": agent_name,
            "stage": state.get("stage", "INIT"),
        },
    )

    try:
        result = agent(state)
        if inspect.isawaitable(result):
            output = await asyncio.wait_for(result, timeout=settings.agent_timeout_seconds)
        else:
            output = result

        await event_hub.publish(
            trace_id,
            "agent_completed",
            {
                "agent_name": agent_name,
                "stage": output.get("stage", state.get("stage")),
                "agent_logs": output.get("agent_logs", []),
                "metrics": output.get("metrics", []),
            },
        )
        return output
    except asyncio.TimeoutError:
        await event_hub.publish(
            trace_id,
            "agent_failed",
            {
                "agent_name": agent_name,
                "stage": state.get("stage", "INIT"),
                "error": f"Agent 执行超过 {settings.agent_timeout_seconds}s",
                "timeout": True,
            },
        )
        raise

这层包装做了三件很工程化的事情：
每个 Agent 开始时推送 agent_started；
执行完成时推送 agent_completed；
超时或异常时推送 agent_failed。
同时，LangGraph 节点上还配置了 RetryPolicy，遇到异常时可以自动重试。
这比“串行调用几个函数”更适合做可观测的 AI 工作流。
ContextAgent：统一上下文和 Java OJ 请求体
ContextAgent 是整个流程的入口节点，它不做判题、不做扫描，只做上下文标准化。

def context_agent(state: JudgeState) -> dict[str, Any]:
    trace_id = state.get("trace_id") or str(uuid4())
    user_code = state.get("user_code") or ""
    language = state.get("language") or _resolve_language(state.get("program_type"))

    context = {
        "user_id": state["user_id"],
        "question_id": state["question_id"],
        "exam_id": state.get("exam_id"),
        "program_type": state["program_type"],
        "language": language,
        "code_length": len(user_code),
        "request_source": "fastapi-langgraph",
    }

    java_judge_payload = {
        "userId": state["user_id"],
        "examId": state.get("exam_id"),
        "programType": state["program_type"],
        "questionId": state["question_id"],
        "difficulty": state.get("difficulty"),
        "timeLimit": state.get("time_limit"),
        "spaceLimit": state.get("space_limit"),
        "userCode": user_code,
        "inputList": state.get("input_list") or [],
        "outputList": state.get("output_list") or [],
    }

    return {
        "trace_id": trace_id,
        "language": language,
        "context": context,
        "java_judge_payload": java_judge_payload,
        "stage": "CONTEXT_READY",
    }

它的作用是把原始请求变成稳定的内部状态，后续 Agent 不需要直接关心 FastAPI 请求结构。
StaticScanAgent：不调用 LLM 的轻量静态扫描
静态扫描 Agent 的设计很克制：它不执行用户代码，也不调用 LLM，只用正则规则识别风险和可疑模式。
风险规则示例：

RISK_PATTERNS = {
    r"\bRuntime\.getRuntime\s*\(": "Java Runtime 执行风险",
    r"\bProcessBuilder\s*\(": "Java 进程创建风险",
    r"\bSystem\.exit\s*\(": "强制退出进程风险",
    r"\bThread\s*\(": "自建线程风险",
    r"\bSocket\s*\(": "网络访问风险",
    r"\bFileInputStream\s*\(": "文件读取风险",
    r"\bFileOutputStream\s*\(": "文件写入风险",
    r"\bjava\.io\.File\b": "文件系统访问风险",
    r"\bClass\.forName\s*\(": "反射加载风险",
    r"\bsetAccessible\s*\(": "反射越权风险",
}

可疑模式示例：

SUSPICIOUS_PATTERNS = {
    r"while\s*\(\s*true\s*\)": "疑似无限循环 while(true)",
    r"for\s*\(\s*;\s*;\s*\)": "疑似无限循环 for(;;)",
    r"\bScanner\s*\(": "Scanner 在大输入下可能较慢",
    r"\bSystem\.out\.println\s*\([^;]*\+\s*": "循环中频繁字符串拼接输出可能较慢",
}

它输出的是结构化扫描结果：

static_scan = {
    "line_count": len(lines),
    "estimated_complexity": _estimate_complexity(user_code),
    "risk_keywords": risk_keywords,
    "suspicious_patterns": suspicious_patterns,
    "suggestions": suggestions,
    "pass_scan": len(risk_keywords) == 0,
}

这里需要注意：StaticScanAgent 不直接拒绝请求，它只是提供信号。真正是否拒绝，由后面的 SecurityRiskAgent 决定。
MemoryAgent：读取用户历史做题画像
MemoryAgent 负责读取用户历史做题记忆：

async def memory_agent(state: JudgeState) -> dict[str, Any]:
    memory = await MemoryStore().get_user_memory(state["user_id"])
    return {
        "user_memory": memory,
        "parallel_completed": [AGENT_NAME],
    }

记忆字段包括：
常见错误类型；
代码风格；
薄弱知识点；
历史提交摘要；
模型偏好。
存储门面是 MemoryStore：
class MemoryStore:
“”"
用户做题记忆门面。

优先使用 Redis；Redis 关闭或异常时自动降级到本地内存，
保证 AI 判题主链路不被记忆系统拖垮。
"""

async def get_user_memory(self, user_id: int) -> UserMemory:
    memory = None
    if self._redis:
        try:
            memory = await self._redis.get(user_id)
        except Exception:
            memory = None

    if memory is None:
        memory = await self._local.get(user_id)

    return memory or _default_memory(user_id)

这个设计重点不是“记忆多高级”，而是“记忆系统不能拖垮主流程”。Redis 异常时自动降级到本地内存，保证 AI 判题链路继续可用。
RagAgent：检索题解和知识点，但不泄露标准答案
RagAgent 负责从知识库里检索相关题解和知识点：

documents = await RagStore().retrieve(
    question_id=state["question_id"],
    language=state.get("language", "java"),
    user_code=state.get("user_code", ""),
    result_type=result_parse.get("result_type"),
    weak_points=user_memory.get("weak_points") or [],
    top_k=settings.rag_top_k,
)

然后写入 rag_context：

rag_context = {
    "enabled": settings.rag_enabled,
    "query": _build_query_summary(state),
    "retrieved_documents": documents,
    "knowledge_points": _extract_by_type(documents, "knowledge"),
    "solution_hints": _extract_by_type(documents, "solution"),
    "personalized_focus": _build_personalized_focus(state, documents),
}

当前版本的 RAG 检索很轻量：本地 JSON + 可解释打分。

def _score_document(self, document, query_tokens, question_id, language) -> float:
    document_text = " ".join(
        [
            document.get("title", ""),
            document.get("content", ""),
            " ".join(document.get("tags") or []),
        ]
    )
    doc_tokens = set(self._tokenize(document_text))
    overlap = len(query_tokens & doc_tokens)
    score = overlap / math.sqrt(len(doc_tokens))

    if question_id in (document.get("question_ids") or []):
        score += 5.0

    tags = {tag.lower() for tag in (document.get("tags") or [])}
    if language.lower() in tags:
        score += 0.5

    return score

这里的取舍是：
不一开始引入复杂向量数据库，降低部署成本；
通过题目 ID、语言、标签、用户代码 token、薄弱点做可解释召回；
后续可以把底层替换为 Chroma / FAISS / Milvus / Pinecone；
Agent 层只依赖 RagDocument，不感知底层存储实现。
另外，Prompt 中明确要求：
结合 RAG 检索到的题解和知识点，但不要直接泄露完整标准答案。

这点对 OJ 很重要，因为平台应该帮助学习，而不是直接给答案。
SecurityRiskAgent：高危代码拒绝进入 Java OJ
SecurityRiskAgent 会根据静态扫描结果做安全策略裁决。
高危规则：

HIGH_RISK_RULES = {
    "Java Runtime 执行风险",
    "Java 进程创建风险",
    "文件写入风险",
    "反射越权风险",
}

中危规则：

MEDIUM_RISK_RULES = {
    "强制退出进程风险",
    "自建线程风险",
    "网络访问风险",
    "文件读取风险",
    "文件系统访问风险",
    "反射加载风险",
}

核心逻辑：

if high_hits:
    security_risk = {
        "allow_judge": False,
        "risk_level": "HIGH",
        "reject_reason": "代码命中高危 API，已在 AI 判题编排层拒绝进入沙箱。",
        "hit_rules": high_hits,
    }
elif medium_hits:
    security_risk = {
        "allow_judge": True,
        "risk_level": "MEDIUM",
        "reject_reason": None,
        "hit_rules": medium_hits,
    }
else:
    security_risk = {
        "allow_judge": True,
        "risk_level": "LOW",
        "reject_reason": None,
        "hit_rules": [],
    }

return {
    "security_risk": security_risk,
    "stage": "SECURITY_CHECKED" if allow_judge else "REJECTED",
    "should_reject": not allow_judge,
    "should_call_java_oj": allow_judge,
}

这一步的意义是：明显危险的代码不用再打到 Java OJ 和 Docker 沙箱，先在 AI 编排层拒绝，降低攻击面和沙箱压力。
但这并不意味着 AI 取代沙箱。Docker 沙箱仍然是最终安全执行边界，AI 风控只是前置过滤。
JavaOjJudgeAgent：Python 不执行代码，只调用 Java OJ
如果安全风控通过，JavaOjJudgeAgent 会调用原 Java OJ 判题接口。

async def java_oj_judge_agent(state: JudgeState) -> dict[str, Any]:
    if state.get("should_reject"):
        return _skip_result(...)

    payload = state.get("java_judge_payload") or {}
    client = JavaOjClient()
    raw_result = await client.judge(payload)
    java_oj_result = _normalize_java_oj_result(raw_result)

    return {
        "java_oj_result": java_oj_result,
        "stage": "JAVA_OJ_JUDGED",
    }

这个 Agent 是跨语言边界：
Python 不编译代码；
Python 不运行用户代码；
Python 不判断最终 AC/WA；
Java OJ 仍负责 Docker 沙箱、测试用例执行、资源限制和原始判题结果。
返回结果会被转换成 Python 状态里的 java_oj_result：

return {
    "pass_flag": data.get("pass"),
    "exe_message": data.get("exeMessage"),
    "score": data.get("score"),
    "case_results": data.get("userExeResultList") or [],
}

这就是“AI 不替代判题”的代码级体现。
AiExplainAgent：LLM 可用就调用，不可用就规则降级
AiExplainAgent 负责最终解释，它会汇总：
静态扫描结果；
安全风控结果；
Java OJ 判题结果；
结果解析；
用户历史记忆；
RAG 检索上下文。
核心逻辑：

async def ai_explain_agent(state: JudgeState) -> dict[str, Any]:
    if settings.llm_enabled:
        try:
            ai_explain = await _build_llm_explain(state)
            llm_message = "LLM 代码点评生成完成。"
        except Exception as exc:
            ai_explain = _build_rule_based_explain(state)
            llm_message = f"LLM 调用失败，已降级为规则解释：{exc}"
    else:
        ai_explain = _build_rule_based_explain(state)
        llm_message = "LLM 未开启，使用规则解释降级。"

    return {
        "ai_explain": ai_explain,
        "stage": "AI_EXPLAINED",
    }

Prompt 中有几个关键约束：

1. 不要替代判题结果，判题结果以 Docker 沙箱执行为准。
2. 结合用户历史记忆，指出重复出现的问题和可执行的改进建议。
3. 结合 RAG 检索到的题解和知识点，但不要直接泄露完整标准答案。
4. 语言简洁，适合在线判题平台直接展示。
   如果 LLM 没开，或者模型调用失败，会走规则化解释：

return {
    "summary": _build_summary(result_type),
    "code_review": _build_code_review(static_scan, security_risk, user_memory, rag_context),
    "optimization_advice": _build_optimization_advice(result_type, complexity_hint),
    "complexity_analysis": complexity_hint,
    "learning_suggestions": _build_learning_suggestions(...),
    "interview_highlights": _build_interview_highlights(result_type),
}

这点是 AI 工程化里非常重要的设计：模型能力可以增强体验，但不能成为系统稳定性的单点依赖。
SSE 实时推送：让前端看到 Agent 执行过程
如果一次 AI 判题链路包含多个 Agent，用户只看到一个长时间 loading 会很不友好。因此项目里提供了 SSE 实时状态流。
事件总线实现：

class SseEventHub:
    def __init__(self, history_limit: int = 200) -> None:
        self._subscribers = defaultdict(set)
        self._history = defaultdict(lambda: deque(maxlen=history_limit))
        self._lock = asyncio.Lock()

    async def publish(self, trace_id: str, event_type: str, payload: dict | None = None) -> None:
        event = {
            "trace_id": trace_id,
            "event_type": event_type,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "payload": payload or {},
        }

        async with self._lock:
            self._history[trace_id].append(event)
            subscribers = list(self._subscribers.get(trace_id, set()))

        for queue in subscribers:
            await queue.put(event)

编码成标准 SSE 文本帧：

def encode_sse(event: dict[str, Any]) -> str:
    event_type = event.get("event_type", "message")
    payload = json.dumps(event, ensure_ascii=False, default=str)
    return f"event: {event_type}\ndata: {payload}\n\n"

FastAPI 暴露三个相关接口：

@app.post("/api/v1/ai-judge/submit/async")
async def submit_ai_judge_async(request: JudgeRequest) -> dict:
    return await service.submit_async(request)

@app.post("/api/v1/ai-judge/submit/stream")
async def submit_ai_judge_stream(request: JudgeRequest) -> StreamingResponse:
    return StreamingResponse(
        service.submit_stream(request),
        media_type="text/event-stream",
    )

@app.get("/api/v1/ai-judge/events/{trace_id}")
async def subscribe_ai_judge_events(trace_id: str) -> StreamingResponse:
    return StreamingResponse(
        service.event_stream(trace_id),
        media_type="text/event-stream",
    )

前端可以看到类似事件：

event: workflow_started
event: agent_started
event: agent_completed
event: workflow_completed
event: heartbeat

为什么这里用 SSE，而不是 WebSocket？
因为这个场景主要是服务端向客户端单向推送判题状态，SSE 足够轻量，浏览器原生支持，重连也方便。如果后续要做双向交互式 Agent，再考虑 WebSocket。
结果持久化：保存完整 JudgeState 方便复盘
AI 判题结果会按 trace_id 保存完整状态。
存储策略：
SQLite：默认兜底，不依赖 Java MySQL；
Redis：可选开启，方便高频查询；
保存完整 JudgeState，包括 Agent 日志、指标、RAG 命中和 AI 解释。
核心代码：

class ResultStore:
    async def save(self, state: dict[str, Any]) -> None:
        await self._local.save(state)
        if self._redis:
            try:
                await self._redis.save(state)
            except Exception:
                pass

SQLite 表结构：

CREATE TABLE IF NOT EXISTS ai_judge_results (
    trace_id TEXT PRIMARY KEY,
    user_id INTEGER,
    question_id INTEGER,
    stage TEXT,
    success INTEGER,
    result_type TEXT,
    total_cost_ms INTEGER,
    state_json TEXT NOT NULL,
    created_at TEXT DEFAULT (datetime('now')),
    updated_at TEXT DEFAULT (datetime('now'))
)

保存完整状态的好处是：
可以按 trace_id 回放一次 AI 判题链路；
可以分析每个 Agent 耗时；
可以定位是 RAG、LLM、Java OJ 还是 Memory 出了问题；
面试讲项目时可以展示完整工程闭环。
一个高危代码拒绝测试
测试请求：

curl -X POST http://127.0.0.1:8000/api/v1/ai-judge/submit \
  -H "Content-Type: application/json" \
  -d '{
    "userId": 1,
    "questionId": 1001,
    "examId": null,
    "programType": 0,
    "language": "java",
    "userCode": "public class Main { public static void main(String[] args) throws Exception { Runtime.getRuntime().exec(null); } }",
    "difficulty": 1,
    "timeLimit": 1000,
    "spaceLimit": 262144,
    "inputList": ["1 2"],
    "outputList": ["3"]
  }'

预期流程：
ContextAgent
-> StaticScanAgent 命中 Java Runtime 执行风险
-> MemoryAgent 读取用户记忆
-> RagAgent 检索知识点
-> ParallelJoinAgent 汇合
-> SecurityRiskAgent 判定 HIGH 风险
-> 跳过 JavaOjJudgeAgent
-> ResultParseAgent
-> AiExplainAgent
-> MemoryPersistAgent
这个测试可以证明：
高危代码不会进入 Java OJ Docker 沙箱；
AI 编排层能独立完成风险拒绝；
即使拒绝，也能返回解释和学习建议；
结果可持久化，可通过 trace_id 查询。

工程设计取舍

1. 为什么不直接改 Java OJ？
   因为原 Java OJ 已经承担核心判题职责，稳定性优先。AI 能力不应该侵入主链路。独立 Python 服务可以快速迭代，也方便后续替换 Agent 框架、RAG 存储或 LLM Provider。
2. 为什么 AI 不参与最终裁决？
   判题结果必须确定、可复现、可解释。LLM 可能幻觉，不能决定 AC / WA。AI 只负责解释和建议，最终结果仍由 Java OJ + Docker 沙箱决定。
3. 为什么 StaticScan / Memory / RAG 并行？
   三者互不依赖：
   静态扫描只依赖用户代码；
   Memory 只依赖 user_id；
   RAG 依赖题目和代码上下文。
   并行执行可以减少整体链路耗时。
4. 为什么 LLM 失败还能返回结果？
   因为 AiExplainAgent 有规则化降级。模型增强的是体验，不是系统可用性的唯一来源。
5. 为什么结果存 SQLite，而不是直接写 Java MySQL？
   AI 服务是增量能力，不应该一开始就修改 Java 主库表结构。SQLite 作为兜底存储足够支撑本地演示和链路复盘，后续需要生产化再迁移到统一数据库。

项目亮点

我没有让 AI 直接判题，而是在原有 Java OJ + Docker 沙箱判题链路之外，增量接入了一个 Python FastAPI + LangGraph 的 AI 判题增强服务。Java OJ 仍然负责确定性的代码编译、沙箱执行、资源限制和 AC/WA/TLE/MLE 等结果裁决；Python 服务只做多 Agent 编排和解释增强。
具体流程是：ContextAgent 先标准化提交上下文，然后并行执行 StaticScanAgent、MemoryAgent 和 RagAgent，分别做静态风险扫描、读取用户历史做题画像、检索题解知识点。三条分支汇合后，SecurityRiskAgent 根据扫描结果决定是否允许进入 Java OJ。如果命中 Runtime.exec、ProcessBuilder 等高危规则，就在 AI 编排层拒绝进入沙箱；否则由 JavaOjJudgeAgent 调用原 Java OJ 判题接口。最后 ResultParseAgent 解析结果，AiExplainAgent 结合静态扫描、RAG、Memory 和 LLM 生成个性化反馈，MemoryPersistAgent 更新用户画像。
工程上我还做了 Agent 超时、RetryPolicy、SSE 实时状态推送、Redis/SQLite 结果持久化，以及 LLM 不可用时的规则化降级，保证 AI 能力不会影响核心判题稳定性。

这个回答里有几个关键词很适合引导追问：
AI 不替代判题；
FastAPI 独立微服务；
LangGraph StateGraph；
JudgeState 全局状态；
StaticScan / Memory / RAG 并行；
SecurityRisk 条件路由；
JavaOjJudgeAgent 跨语言调用；
SSE 实时推送；
LLM 降级；
Redis / SQLite 持久化。

总结

这套 LangGraph 多 Agent AI 判题增强服务，本质上不是“给 OJ 套一层大模型”，而是一次 AI 工程化增量改造。
核心设计可以总结为：
用 FastAPI 独立部署 AI 服务，不侵入 Java OJ 主链路；
用 LangGraph StateGraph 编排多个 Agent；
用 JudgeState 贯穿上下文、日志、指标和结果；
用 StaticScanAgent 做轻量静态扫描；
用 MemoryAgent 引入用户历史做题画像；
用 RagAgent 检索题解和知识点，但不泄露完整答案；
用 SecurityRiskAgent 决定是否允许进入 Java OJ；
用 JavaOjJudgeAgent 调用原 Java OJ，Python 不执行用户代码；
用 AiExplainAgent 生成解释，并支持 LLM 失败时规则降级；
用 SSE 推送实时状态；
用 Redis / SQLite 保存完整链路结果。
最重要的一点是：AI 始终是增强能力，不是最终裁判。
对于 OJ 这种对确定性和安全性要求很高的系统，这个边界比“模型效果看起来很智能”更重要。