一、引言
在AI技术飞速发展的今天，如何将大语言模型的能力真正落地到教育行业的核心业务场景中，是每个教育科技从业者都在思考的问题。RAG（检索增强生成）让AI能够查阅专有文档，Agent让AI具备了自主决策和调用外部工具的能力，而Function Calling则打通了LLM与外部系统的交互通道。

本文将带你从零开始，使用Python、FastAPI、LangChain和Qdrant等主流技术栈，构建一个面向教育场景的全栈AI系统。我们将深入拆解RAG知识库构建、Agent智能体开发、Function Calling工具调用三大核心模块，并提供完整的可运行代码。

本文所有代码基于Python 3.11+、LangChain、FastAPI和Qdrant，完整项目结构参考了企业级AI Agent后端工程的实践经验。

二、系统架构概览
在动手写代码之前，我们先明确系统的整体架构。一个完整的教育AI全栈系统通常采用前后端分离的微服务架构：

前端层：React/Vue提供交互界面，支持流式对话和知识库管理

API网关层：FastAPI提供RESTful接口和SSE事件流

Agent核心层：LangChain负责Agent的思考-行动-观察循环

RAG引擎层：Qdrant存储文档向量，实现语义检索

工具层：Function Calling封装各类教育业务动作

基础设施层：Redis缓存会话，MySQL存储业务数据

text
用户 → React/Vue前端 → FastAPI后端 → Agent(LLM+Tools) → RAG检索 → 知识库
↓
Function Calling → 外部系统(数据库/API)
三、RAG核心实现：教育知识库的构建
RAG的本质是给大模型外挂一个知识库。对于教育场景，这意味着将课程资料、教学大纲、政策文件等专有文档转化为AI可以检索和引用的知识。

3.1 文档解析与文本分块
第一步是将各种格式的文档统一处理成纯文本。以下代码支持PDF、Word和TXT格式的解析：

python
import os
import PyPDF2
import docx

def load_plain_text(file_path: str) -> str:
“”“加载纯文本文件”“”
with open(file_path, ‘r’, encoding=‘utf-8’) as fp:
return fp.read()

def extract_text_from_pdf(file_path: str) -> str:
“”“从PDF提取文本”“”
texts = []
with open(file_path, ‘rb’) as fp:
reader = PyPDF2.PdfReader(fp)
for pg in reader.pages:
page_txt = pg.extract_text() or “”
texts.append(page_txt)
return “\n”.join(texts)

def extract_text_from_docx(file_path: str) -> str:
“”“从Word文档提取文本”“”
doc = docx.Document(file_path)
paras = [p.text for p in doc.paragraphs]
return “\n”.join(paras)

def load_document(file_path: str) -> str:
“”“根据文件扩展名自动选择解析方式”“”
_, extension = os.path.splitext(file_path)
extension = extension.lower()
if extension == ‘.txt’:
return load_plain_text(file_path)
elif extension == ‘.pdf’:
return extract_text_from_pdf(file_path)
elif extension == ‘.docx’:
return extract_text_from_docx(file_path)
else:
raise ValueError(f"不支持的文件类型: {extension}")
文本分块是RAG中最关键的环节之一。分块太大，检索精度下降；分块太小，上下文信息不足。以下是一个基于语义边界的递归分块实现：

python
from langchain.text_splitter import RecursiveCharacterTextSplitter

def chunk_document(text: str, chunk_size: int = 500, chunk_overlap: int = 50):
“”“将文档切分为适合检索的文本块”“”
splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=[“\n\n”, “\n”, “。”, “！”, “？”, “；”, “，”, " ", “”]
)
return splitter.split_text(text)
3.2 向量化与存储
将文本块转化为向量，并存入向量数据库，是RAG检索的核心步骤。以下使用Qdrant作为向量数据库，配合OpenAI-compatible的Embedding服务：

python
from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance
from openai import OpenAI

初始化Embedding客户端

embedding_client = OpenAI(
base_url=os.getenv(“EMBEDDING_BASE_URL”),
api_key=os.getenv(“EMBEDDING_API_KEY”)
)

初始化Qdrant客户端

qdrant_client = QdrantClient(
url=os.getenv(“QDRANT_URL”),
prefer_grpc=False
)

def generate_embeddings(texts: list[str]) -> list[list[float]]:
“”“批量生成文本向量”“”
response = embedding_client.embeddings.create(
model=os.getenv(“EMBEDDING_MODEL”),
input=texts
)
return [item.embedding for item in response.data]

def create_or_get_collection(collection_name: str, vector_size: int):
“”“创建或获取向量集合”“”
collections = qdrant_client.get_collections().collections
if not any(c.name == collection_name for c in collections):
qdrant_client.create_collection(
collection_name=collection_name,
vectors_config=VectorParams(
size=vector_size,
distance=Distance.COSINE
)
)
return collection_name
3.3 检索与重排序
当用户提出问题时，系统需要从知识库中检索最相关的文档片段。为了提升检索质量，可以采用双阶段检索策略：先用向量检索快速召回候选集，再用重排序模型精排：

python
def retrieve_context(query: str, collection_name: str, top_k: int = 5) -> list[str]:
“”“检索与查询最相关的文档片段”“”
# 1. 生成查询向量
query_embedding = generate_embeddings([query])[0]

# 2. 向量检索
search_result = qdrant_client.search(
    collection_name=collection_name,
    query_vector=query_embedding,
    limit=top_k * 2  # 召回双倍候选，供重排序
)

# 3. 提取文本
candidates = [hit.payload.get("text", "") for hit in search_result]

# 4. 简单重排序（可按业务规则调整）
# 实际生产中可以接入Cross-Encoder模型做精排
return candidates[:top_k]

四、Agent智能体与Function Calling
有了RAG知识库，AI可以"知道"知识。但要让AI真正"做事"——比如查询学生信息、生成学习报告、推送课程通知——就需要Agent和Function Calling了。

4.1 定义教育场景的工具（Tools）
Function Calling的核心思想是：LLM根据用户意图，决定调用哪个预定义函数，并生成正确的调用参数。以下定义三个教育场景的典型工具：

python
from langchain.tools import tool
from pydantic import BaseModel, Field
from typing import Optional

工具1：查询学生信息

class StudentQueryInput(BaseModel):
student_id: str = Field(description=“学生学号”)
query_type: str = Field(description=“查询类型:成绩/出勤/课程表”)

@tool(args_schema=StudentQueryInput)
def query_student_info(student_id: str, query_type: str) -> str:
“”"
查询学生信息。当用户询问某个学生的成绩、出勤或课程表时使用。
“”"
# 实际生产环境这里会查询数据库
mock_data = {
“成绩”: “张三，学号2024001，高等数学: 92分，数据结构: 88分”,
“出勤”: “张三，学号2024001，本学期出勤率: 95%，缺勤2次”,
“课程表”: “张三，学号2024001，周一: 高数(8:00)，数据结构(10:00)”
}
return mock_data.get(query_type, “未找到相关信息”)

工具2：生成学习报告

class ReportInput(BaseModel):
student_id: str = Field(description=“学生学号”)
period: str = Field(description=“报告周期: weekly/monthly/semester”)

@tool(args_schema=ReportInput)
def generate_learning_report(student_id: str, period: str) -> str:
“”"
生成学生学习报告。当用户要求生成学习总结或进步报告时使用。
“”"
return f"已生成{student_id}的{period}学习报告：该生表现良好，建议继续加强编程实践。"

工具3：RAG知识库检索（封装为工具）

@tool
def search_knowledge_base(query: str) -> str:
“”"
从教育知识库中检索相关信息。当用户询问课程内容、政策规定等知识性问题时使用。
“”"
contexts = retrieve_context(query, “education_knowledge”, top_k=3)
if not contexts:
return “未在知识库中找到相关信息。”
return “\n—\n”.join(contexts)
4.2 构建Agent执行器
有了工具之后，我们需要构建Agent的"大脑"——它负责理解用户意图、决定调用哪些工具、以及如何组合工具完成任务：

python
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI

初始化LLM

llm = ChatOpenAI(
model=os.getenv(“LLM_MODEL”, “gpt-4o-mini”),
base_url=os.getenv(“LLM_BASE_URL”),
api_key=os.getenv(“LLM_API_KEY”),
temperature=0.3
)

定义系统提示词——教育场景的人设与工具使用策略

SYSTEM_TEMPLATE = “”"
你是一位专业的AI教育助手，名叫"小智"。

你的职责

  • 帮助学生解答学习问题
  • 协助教师管理教学事务
  • 提供个性化的学习建议

工具使用原则

  1. 遇到学生个人信息查询 → 调用 query_student_info
  2. 需要生成学习报告 → 调用 generate_learning_report
  3. 遇到课程知识、政策规定 → 优先调用 search_knowledge_base
  4. 如果不确定，先调用 search_knowledge_base 获取相关信息

回答风格

  • 用亲切、专业的语气
  • 引用知识库来源时标注出处
  • 涉及学生隐私时注意脱敏
    “”"

组装工具列表

tools = [query_student_info, generate_learning_report, search_knowledge_base]

创建提示模板

prompt = ChatPromptTemplate.from_messages([
(“system”, SYSTEM_TEMPLATE),
MessagesPlaceholder(variable_name=“chat_history”),
(“human”, “{input}”),
MessagesPlaceholder(variable_name=“agent_scratchpad”)
])

创建Agent

agent = create_tool_calling_agent(llm, tools, prompt)

创建Agent执行器

agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=5,
handle_parsing_errors=True
)

def run_agent(query: str, session_id: str = None) -> str:
“”“运行Agent，返回响应”“”
# 实际生产环境会从Redis加载会话历史
result = agent_executor.invoke({
“input”: query,
“chat_history”: [] # 传入历史消息列表
})
return result[“output”]
Agentic RAG与传统RAG的关键区别在于：传统RAG是固定流程（检索→生成），而Agentic RAG让LLM自主决定何时检索、检索什么、以及如何利用检索结果。上面的search_knowledge_base工具就是把RAG检索封装成了Agent可以主动调用的能力。

五、全栈API服务
5.1 FastAPI后端
将Agent封装为RESTful API服务，同时支持SSE流式响应：

python
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional
import asyncio
from sse_starlette.sse import EventSourceResponse

app = FastAPI(title=“教育AI助手API”, version=“1.0.0”)

app.add_middleware(
CORSMiddleware,
allow_origins=[““],
allow_methods=[””],
allow_headers=[“*”],
)

class ChatRequest(BaseModel):
query: str
session_id: Optional[str] = None
stream: bool = False

class ChatResponse(BaseModel):
answer: str
session_id: str
sources: Optional[list[str]] = None

@app.post(“/api/chat”, response_model=ChatResponse)
async def chat(request: ChatRequest):
“”“同步对话接口”“”
try:
answer = run_agent(request.query, request.session_id)
return ChatResponse(
answer=answer,
session_id=request.session_id or “default”,
sources=[] # 可以提取引用来源
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))

@app.post(“/api/chat/stream”)
async def chat_stream(request: ChatRequest):
“”“流式对话接口（SSE）”“”
async def event_generator():
# 实际生产环境使用流式LLM输出
answer = run_agent(request.query, request.session_id)
# 模拟流式输出
for chunk in answer.split():
yield {“event”: “message”, “data”: chunk}
yield {“event”: “done”, “data”: “[DONE]”}

return EventSourceResponse(event_generator())

5.2 会话管理与记忆
多轮对话需要记忆能力。使用Redis存储会话历史：

python
import redis
import json
from typing import List, Dict

redis_client = redis.Redis.from_url(os.getenv(“REDIS_URL”))

def get_session_history(session_id: str) -> List[Dict]:
“”“获取会话历史”“”
key = f"session:{session_id}"
data = redis_client.get(key)
if data:
return json.loads(data)
return []

def append_to_session(session_id: str, role: str, content: str, max_turns: int = 10):
“”“追加消息到会话，保留最近N轮”“”
key = f"session:{session_id}"
history = get_session_history(session_id)
history.append({“role”: role, “content”: content})
# 保留最近max_turns轮对话
if len(history) > max_turns * 2:
history = history[-max_turns * 2:]
redis_client.setex(key, 3600 * 24, json.dumps(history)) # 24小时过期
六、部署与运维
使用Docker Compose一键编排所有服务：

yaml

docker-compose.yml

version: ‘3.8’
services:
redis:
image: redis:7-alpine
ports:
- “6379:6379”

qdrant:
image: qdrant/qdrant:latest
ports:
- “6333:6333”
volumes:
- qdrant_data:/qdrant/storage

mysql:
image: mysql:8.0
environment:
MYSQL_ROOT_PASSWORD: root
MYSQL_DATABASE: education_ai
ports:
- “3306:3306”
volumes:
- mysql_data:/var/lib/mysql

backend:
build: .
ports:
- “8000:8000”
environment:
- REDIS_URL=redis://redis:6379/0
- QDRANT_URL=http://qdrant:6333
- DATABASE_URL=mysql+pymysql://root:root@mysql:3306/education_ai
depends_on:
- redis
- qdrant
- mysql
command: uvicorn main:app --host 0.0.0.0 --port 8000 --reload

volumes:
qdrant_data:
mysql_data:
启动命令：

bash
docker-compose up -d
七、总结
本文从零开始构建了一个完整的教育AI全栈系统，核心模块包括：

RAG知识库：通过文档解析、文本分块、向量化存储和语义检索，让AI能够"知道"教育领域的专有知识。

Agent智能体：利用LangChain的create_tool_calling_agent构建具备自主决策能力的AI助手。

Function Calling：将教育业务动作（查成绩、生成报告、检索知识）封装为工具，让LLM能够"做事"而非仅仅"说话"。

全栈服务：FastAPI提供RESTful API和SSE流式接口，Docker Compose实现一键部署。

这套架构的价值在于：它不只是"调API"，而是把AI能力深度融入教育业务的全链路——从知识管理到智能问答，从数据查询到报告生成，形成了一个可扩展、可运维的闭环系统。

当然，生产环境还需要考虑更多：多租户隔离、更精细的权限控制、RAG检索的持续优化、以及针对教育场景的提示词工程调优。这些都可以在现有架构上逐步迭代扩展。

希望这篇文章能为正在探索教育AI落地的开发者提供一些参考。代码即架构，架构即思考——愿我们都能从"调包侠"成长为真正的AI全栈工程师。

# 人工智能
Logo
人工智能6S服务平台

作为“人工智能6S店”的官方数字引擎，为AI开发者与企业提供一个覆盖软硬件全栈、一站式门户。

更多推荐

cover

vLLM适配昇腾NPU：DeepSeek-V3 PD分离部署完整流程

avatar 人工智能6S服务平台
cover

昇腾NPU上编译Apex：从踩坑到搞定

avatar 人工智能6S服务平台
cover

鸿蒙跨设备文件高级共享：分布式文件共享/权限临时授权时效管控/共享生命周期管理方案

avatar 人工智能6S服务平台