Quiz Agent

测验代理

使用LangGraph构建的最小发布问答代理，可以在终端中运行，并可作为iOS问答应用程序的后端引擎重用。

分配上下文

本项目符合AI开发人员课程作业要求：

框架：LangGraph（状态图模式） 工具：Tavily MCP服务器（模型上下文协议） LLM：OpenAI GPT-4o-mini

为什么选择MCP？ 根据任务的建议，使用MCP而不是框架特定的工具，此实现连接到Tavily的托管MCP服务器(langchain-mcp-adapters)用于网络搜索功能。这为工具集成提供了一种标准化的、基于协议的方法。

符合分配标准:

• ✅ 带有工具集成的代理框架（LangGraph）
• ✅ MCP协议使用（Tavily MCP服务器）
• ✅ 基于LLM的查询响应（OpenAI）
• ✅ 外部工具集成（网络搜索答案验证）
• ✅ 完整的源代码和文档

概述

该代理充当一个简洁的酒吧问答“测验大师”，即：

• 使用LLM生成琐事问题
• 通过细致的评分来评估用户的答案
• 通过Tavily MCP提供源链接
• 跟踪分数和难度

建筑

现有代码库分析

测验代理是按照课程中的模式构建的 WebOperator 例子：

状态图模式 （从 11-web-operator/4_Web_operator/agent2_correct/graph.py):

• 州定义为 TypedDict 随着 Annotated 类型
• add_messages 消息历史记录缩减器
• 基于状态的路由条件边

工具集成 （从 8_langgraph/5_agent/0_react_manual/main.py):

• MultiServerMCPClient 用于Tavily MCP连接
• 用于处理工具调用的自定义工具节点模式
• 异步调用 ainvoke()

操作模式 （从 11-web-operator/4_Web_operator/agent2_correct/web_operator.py):

• 基于类的运算符 initialize() 和 run() 方法
• 跨调用的持久状态
• 与stdin/stdout的交互循环

测验代理架构

┌─────────────┐
│   START     │
└──────┬──────┘
       │
       ▼
┌──────────────┐     ┌────────────────┐
│ process_input│────▶│generate_question│
└──────┬───────┘     └───────┬────────┘
       │                     │
       ▼                     ▼
┌──────────────┐           END (wait for answer)
│evaluate_answer│
└──────┬───────┘
       │
       ▼
┌──────────────┐
│ find_source  │
└──────┬───────┘
       │
       ▼
┌────────────────┐
│provide_feedback│
└───────┬────────┘
        │
        ▼
       END (wait for next input)

状态模式

class QuizState(TypedDict):
    messages: Annotated[list, add_messages]  # LangGraph message reducer
    phase: str              # "idle" | "asking" | "awaiting_answer" | "evaluating" | "finding_source" | "providing_feedback" | "finished"
    question_number: int    # Current question (1-indexed)
    max_questions: int      # Total questions (default 10)
    current_question: str   # Question text
    current_answer: str     # Canonical answer
    current_difficulty: str # "easy" | "medium" | "hard"
    current_topic: str      # Category (History, Science, etc.)
    last_user_answer: str   # User's answer
    last_result: str        # "correct" | "partially_correct" | "partially_incorrect" | "incorrect" | "skipped"
    score: float            # Running score
    last_source_url: str    # Source URL from Tavily
    last_source_snippet: str # Source snippet
    pending_output: str     # Text to display to user

图形节点

评分

用法

设置

# Install dependencies
pip install -e .

# Or with uv
uv pip install -e .

# Set up environment
cp .env.example .env
# Edit .env with your API keys

跑

python quiz_main.py

命令

示例会话

Welcome to Pub Quiz! Type 'start' to begin.

> start

Question 1: What element has the chemical symbol Au?

> gold

Correct.
Source: https://en.wikipedia.org/wiki/Gold
Gold is a chemical element with symbol Au...
[Score: 1.0/1]

Question 2: In what year did the Berlin Wall fall?

> 1990

Partially correct. The full answer is 1989.
Source: https://en.wikipedia.org/wiki/Fall_of_the_Berlin_Wall
The Berlin Wall fell on November 9, 1989...
[Score: 1.5/2]

> harder

Difficulty set to hard.

Question 3: What is the Schwarzschild radius formula?

> skip

Skipped. The answer is rs = 2GM/c².
Source: https://en.wikipedia.org/wiki/Schwarzschild_radius
The Schwarzschild radius defines the radius of the event horizon...
[Score: 1.5/3]

iOS应用程序集成

这 create_quiz_graph() 函数返回一个编译后的StateGraph，可以封装在各种后端中：

REST API（FastAPI）

from fastapi import FastAPI
from graph import create_quiz_graph, get_initial_state

app = FastAPI()
sessions = {}  # Store quiz states by session ID

@app.post("/quiz/{session_id}/input")
async def process_input(session_id: str, user_input: str):
    if session_id not in sessions:
        sessions[session_id] = get_initial_state()
        sessions[session_id]["graph"] = await create_quiz_graph(mcp_tools)

    state = sessions[session_id]
    state["messages"].append(HumanMessage(content=user_input))
    result = await state["graph"].ainvoke(state)
    state.update(result)

    return {
        "response": state.get("pending_output", ""),
        "phase": state["phase"],
        "score": state["score"],
        "question_number": state["question_number"]
    }

WebSocket

@app.websocket("/quiz/ws")
async def quiz_websocket(websocket: WebSocket):
    await websocket.accept()
    state = get_initial_state()
    graph = await create_quiz_graph(mcp_tools)

    while True:
        user_input = await websocket.receive_text()
        state["messages"].append(HumanMessage(content=user_input))
        result = await graph.ainvoke(state)
        state.update(result)

        await websocket.send_json({
            "response": state.get("pending_output", ""),
            "phase": state["phase"],
            "score": state["score"]
        })

状态序列化

这 QuizState 可以序列化为JSON以实现持久性：

import json

# Save state
state_json = json.dumps({
    k: v for k, v in state.items()
    if k != "messages"  # Handle messages separately
})

# Messages need special handling for HumanMessage/AIMessage
messages_json = [
    {"type": m.__class__.__name__, "content": m.content}
    for m in state["messages"]
]

文件结构

quiz_agent/
├── graph.py          # State schema, nodes, graph construction
├── quiz_main.py      # Terminal entry point, QuizOperator class
├── pyproject.toml    # Dependencies
├── .env.example      # Environment template
└── README.md         # This file

与WebOperator的主要区别

环境变量

依赖项

• langchain>=0.3.0 -LLM框架
• langchain-openai>=0.2.0 -OpenAI集成
• langgraph>=0.2.0 -状态图框架
• langchain-mcp-adapters>=0.0.1 -MCP客户端适配器
• python-dotenv>=1.0.0 -环境载荷