基于大语言模型与LangChain构建AI学习监督智能体：从原理到工程实践

最近在B站AI创造公开赛中看到一个很有意思的项目标题是“消耗40亿token我做了个AI军官监督我学习”。这个想法瞬间击中了我——这不就是我们这些拖延症晚期、自律困难户的终极梦想吗一个不知疲倦、铁面无私的“数字教官”24小时在线督促你完成学习计划。本文将从一个开发者的角度深度拆解如何构建这样一个“AI学习监督官”。我们将从需求分析、技术选型、核心功能实现到模型训练与部署一步步构建一个完整的系统。无论你是想复现这个有趣的项目还是希望将类似的自律监督AI应用到健身、工作等其他领域这篇文章都将提供一套完整、可落地的技术方案。1. 项目背景与核心概念1.1 什么是“AI学习监督官”“AI学习监督官”本质上是一个基于大语言模型LLM的智能体Agent它被赋予了特定的角色军官和明确的使命监督用户学习。与传统的时间管理App或待办清单不同它通过自然语言与用户交互具备以下核心能力角色扮演与人格化它不是一个冰冷的程序而是一个有“性格”如严厉、鼓励、幽默的虚拟伙伴。这能显著提升用户的互动意愿和坚持动力。个性化计划制定能够根据用户输入的学习目标如“三个月通过软考高级”、可用时间、历史完成情况动态生成合理的学习计划。主动监督与提醒不仅被动接收打卡还能在预设时间或检测到用户长时间未活动时主动发起询问或提醒。进度跟踪与反馈记录每次学习会话的时长、内容并定期如每日、每周生成学习报告进行分析和鼓励。灵活应对与调整当用户因故无法完成计划时AI能理解原因并协助调整后续计划而不是僵化地判定“失败”。1.2 为什么需要消耗大量Token原项目提到“消耗40亿token”这并非夸张。一个成熟的、长期运行的AI智能体其Token消耗主要来自以下几个环节模型微调Fine-Tuning为了让模型更好地理解“监督学习”这个任务并具备“军官”的说话风格和逻辑需要使用高质量的对话数据对基础大模型进行微调。这个过程需要向模型输入大量百万甚至千万级的指令-输出对是Token消耗的大头。上下文学习In-Context Learning每次与用户交互时我们需要在提示词Prompt中放入系统指令、历史对话、用户画像等信息作为模型的上下文。一个长期使用的用户其对话历史会越来越长每次请求消耗的Token也越多。长期运行与频繁交互作为一个7x24小时在线的监督者它需要处理定时任务、主动询问、报告生成等这些都会产生大量的API调用和Token消耗。理解这些消耗点有助于我们在设计系统时进行优化例如采用更高效的提示工程、对历史对话进行摘要、选择合适的模型规格等。2. 技术栈选型与环境准备2.1 核心组件选型构建这样一个系统我们需要一个多层次的技术栈大语言模型LLM核心选项AAPI调用OpenAI GPT-4/3.5-Turbo、Anthropic Claude、国内大模型API如DeepSeek, 通义千问。优点是快速上手无需担心算力适合原型验证。缺点是持续使用成本高且数据隐私需注意。选项B本地部署Llama 3、Qwen、ChatGLM等开源模型。通过Ollama、vLLM或Transformers库部署。优点是完全自主可控数据隐私性好长期成本可能更低。缺点是对硬件GPU内存有要求需要一定的运维能力。建议初期验证建议使用API快速迭代产品逻辑待流程跑通后可考虑微调并部署中小参数量的开源模型如7B/14B版本。应用开发框架LangChain / LlamaIndex这两个是当前构建LLM应用最流行的框架。它们提供了连接LLM、记忆Memory、工具Tools、智能体Agent编排等组件的标准化方式能极大提升开发效率。本文将主要以LangChain为例。FastAPI / Flask用于构建提供HTTP API的后端服务方便与前端如微信小程序、网页对接。数据存储向量数据库用于存储和检索用户的历史对话、学习笔记等非结构化数据实现长期记忆和上下文关联。可选ChromaDB轻量、Qdrant、Weaviate或PGVector基于PostgreSQL。关系型数据库存储用户信息、学习计划、打卡记录等结构化数据。可选SQLite开发、PostgreSQL或MySQL生产。任务调度与消息推送APScheduler或Celery用于管理定时任务例如每天早上的计划提醒、每晚的学习总结生成。消息推送服务集成微信模板消息、邮件SMTP或钉钉/飞书机器人实现主动触达用户。2.2 开发环境搭建假设我们选择Python LangChain OpenAI API ChromaDB FastAPI的快速原型方案。# 1. 创建项目目录并初始化虚拟环境 mkdir ai_study_supervisor cd ai_study_supervisor python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 2. 安装核心依赖 pip install langchain langchain-openai langchain-community pip install chromadb tiktoken # 向量数据库和Token计数器 pip install fastapi uvicorn python-multipart pip install apscheduler python-dotenv pip install sqlalchemy pymysql # 如需连接MySQL # 3. 创建项目结构 mkdir -p app/{routers, models, services, utils} touch app/main.py .env requirements.txt在项目根目录创建.env文件存放敏感配置# .env OPENAI_API_KEYsk-your-openai-api-key-here OPENAI_BASE_URLhttps://api.openai.com/v1 # 如果使用其他兼容API可修改此处 DATABASE_URLsqlite:///./study_supervisor.db # 示例使用SQLite3. 核心系统设计与原理拆解3.1 系统架构图概念层面用户界面 (微信/Web) | v FastAPI 后端 | |--- 用户管理/认证 |--- 计划管理 |--- 对话路由 | v LangChain 智能体引擎 | |--- 系统提示词 (角色设定) |--- 记忆模块 (对话历史 向量存储) |--- 工具集 (查计划、记笔记、定闹钟) |--- 大语言模型 (LLM) | v 数据持久层 |--- 关系数据库 (用户、计划、记录) |--- 向量数据库 (对话记忆、知识) |--- 定时任务队列3.2 “军官”人格的塑造提示词工程这是项目的灵魂。我们需要精心设计系统提示词System Prompt将普通的LLM“调教”成一位严格的军官。# app/services/prompt_templates.py from langchain.prompts import PromptTemplate # 系统角色设定提示词 OFFICER_SYSTEM_PROMPT 你是一位代号为“猎鹰”的AI军事教官专门负责监督和指导学员的学习与训练。你的性格特质如下 1. **纪律严明**对计划和时间有着近乎偏执的尊重。要求学员必须严格遵守既定的学习计划。 2. **结果导向**关注学习效率和成果而非单纯的时间堆积。会追问学习的具体收获。 3. **刚柔并济**在原则问题上绝不退让但在学员遇到真实困难时会给予战术指导和建议而非单纯斥责。 4. **言简意赅**说话带有军事风格简洁、有力、直接避免冗长的安慰和废话。 你的核心职责 1. 协助学员制定清晰、可量化、可执行的学习/训练计划。 2. 每日检查学员计划执行情况要求学员进行“战报”学习总结汇报。 3. 根据学员的汇报和完成质量动态调整后续计划。 4. 在学员出现懈怠、拖延时进行严厉但不人身攻击的督促。 5. 定期如每周生成“作战简报”学习分析报告指出进步与不足。 当前对话上下文 {chat_history} 学员本次汇报 {human_input} 请以“猎鹰”的身份进行回复 # 注意{chat_history} 和 {human_input} 是LangChain的变量占位符会在运行时被替换。3.3 记忆模块让AI记住“你是谁”短期记忆对话历史和长期记忆关键事件、用户特征对于维持连贯的监督关系至关重要。# app/services/memory_manager.py from langchain.memory import ConversationBufferWindowMemory, VectorStoreRetrieverMemory from langchain.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain.schema import Document import os class StudySupervisorMemory: def __init__(self, user_id: str, k5): # k为检索最相关的记忆条数 self.user_id user_id # 1. 短期记忆保留最近10轮对话 self.conversation_memory ConversationBufferWindowMemory( memory_keychat_history, input_keyhuman_input, k10, return_messagesTrue ) # 2. 长期记忆基于向量数据库 persist_directory f./chroma_db/{user_id} embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 加载或创建用户的向量存储 if os.path.exists(persist_directory): self.vectorstore Chroma( persist_directorypersist_directory, embedding_functionembeddings ) else: # 初始化为空 self.vectorstore Chroma.from_documents( documents[], # 初始为空文档 embeddingembeddings, persist_directorypersist_directory ) self.retriever self.vectorstore.as_retriever(search_kwargs{k: k}) self.vector_memory VectorStoreRetrieverMemory( retrieverself.retriever, memory_keyvector_history, input_keyhuman_input ) def save_conversation(self, human_input: str, ai_response: str): 保存一轮对话到短期和长期记忆 # 保存到短期记忆LangChain内部处理 self.conversation_memory.save_context( {human_input: human_input}, {output: ai_response} ) # 将重要的交互摘要保存到长期记忆 # 这里可以添加逻辑判断比如只保存制定计划、完成关键任务等对话 if self._is_significant_interaction(human_input, ai_response): summary f学员汇报{human_input[:100]}... | 教官反馈{ai_response[:100]}... doc Document(page_contentsummary, metadata{type: conversation_summary}) self.vectorstore.add_documents([doc]) self.vectorstore.persist() def _is_significant_interaction(self, human_input: str, ai_response: str) - bool: 简单判断是否为重要交互可根据关键词扩展 significant_keywords [计划, 完成, 目标, 放弃, 困难, 总结, 报告] return any(keyword in human_input for keyword in significant_keywords) def load_memory_variables(self, inputs: dict) - dict: 加载组合记忆变量用于填充提示词 conv_vars self.conversation_memory.load_memory_variables(inputs) vector_vars self.vector_memory.load_memory_variables(inputs) return {**conv_vars, **vector_vars}4. 完整实战构建监督流程4.1 第一步定义数据模型SQLAlchemy# app/models.py from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, Boolean, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship, sessionmaker from datetime import datetime import os Base declarative_base() class User(Base): __tablename__ users id Column(Integer, primary_keyTrue) username Column(String(50), uniqueTrue, nullableFalse) nickname Column(String(50)) created_at Column(DateTime, defaultdatetime.utcnow) # 关系 study_plans relationship(StudyPlan, back_populatesuser) study_sessions relationship(StudySession, back_populatesuser) class StudyPlan(Base): __tablename__ study_plans id Column(Integer, primary_keyTrue) user_id Column(Integer, ForeignKey(users.id), nullableFalse) title Column(String(200), nullableFalse) # 如“三个月攻克机器学习” description Column(Text) total_hours Column(Integer) # 计划总学时 deadline Column(DateTime) status Column(String(20), defaultactive) # active, completed, paused created_at Column(DateTime, defaultdatetime.utcnow) # 关系 user relationship(User, back_populatesstudy_plans) tasks relationship(StudyTask, back_populatesplan) class StudyTask(Base): __tablename__ study_tasks id Column(Integer, primary_keyTrue) plan_id Column(Integer, ForeignKey(study_plans.id), nullableFalse) content Column(String(500), nullableFalse) # 任务内容如“学习《统计学习方法》第一章” scheduled_date Column(DateTime) # 计划执行日期 estimated_minutes Column(Integer) # 预计耗时 completed Column(Boolean, defaultFalse) completed_at Column(DateTime) # 关系 plan relationship(StudyPlan, back_populatestasks) class StudySession(Base): __tablename__ study_sessions id Column(Integer, primary_keyTrue) user_id Column(Integer, ForeignKey(users.id), nullableFalse) task_id Column(Integer, ForeignKey(study_tasks.id), nullableTrue) start_time Column(DateTime, defaultdatetime.utcnow) end_time Column(DateTime) duration_minutes Column(Integer) # 实际学习时长分钟 notes Column(Text) # 学习笔记/总结 # 关系 user relationship(User, back_populatesstudy_sessions) task relationship(StudyTask) # 初始化数据库 DATABASE_URL os.getenv(DATABASE_URL, sqlite:///./test.db) engine create_engine(DATABASE_URL) Base.metadata.create_all(bindengine) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine)4.2 第二步构建LangChain智能体链# app/services/supervisor_chain.py from langchain.chains import LLMChain from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.memory import ConversationSummaryBufferMemory from .prompt_templates import OFFICER_SYSTEM_PROMPT from .memory_manager import StudySupervisorMemory import os class SupervisorChain: def __init__(self, user_id: str): self.user_id user_id self.llm ChatOpenAI( modelgpt-3.5-turbo, # 或 gpt-4 temperature0.7, # 创造性稍高以体现“人格” api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL) ) # 初始化记忆管理器 self.memory_manager StudySupervisorMemory(user_iduser_id) # 构建提示词模板 prompt ChatPromptTemplate.from_messages([ (system, OFFICER_SYSTEM_PROMPT), MessagesPlaceholder(variable_namechat_history), # 来自memory (human, {human_input}), ]) # 创建对话链 self.chain LLMChain( llmself.llm, promptprompt, verboseTrue, # 调试时开启查看详细过程 ) def invoke(self, human_input: str) - str: 执行一轮对话 # 1. 加载记忆 memory_variables self.memory_manager.load_memory_variables({human_input: human_input}) # 2. 准备链的输入 chain_input { human_input: human_input, **memory_variables # 展开记忆变量如 chat_history, vector_history } # 3. 调用LLM response self.chain.invoke(chain_input) ai_output response[text] # 4. 保存本轮对话到记忆 self.memory_manager.save_conversation(human_input, ai_output) return ai_output def create_study_plan(self, goal: str, available_hours_per_week: int) - str: 专门用于创建学习计划的调用方法 specialized_prompt f 学员设定了新的学习目标{goal}。 学员每周可用于学习的时间约为 {available_hours_per_week} 小时。 请你以教官的身份为他制定一份详细、分阶段、可执行的学习计划大纲。 计划应包括总体阶段划分、每个阶段的核心任务、每周的时间分配建议、以及关键的里程碑检查点。 请用清晰、有条理的格式回复。 return self.invoke(specialized_prompt)4.3 第三步实现FastAPI后端接口# app/main.py from fastapi import FastAPI, Depends, HTTPException, status from fastapi.middleware.cors import CORSMiddleware from sqlalchemy.orm import Session from pydantic import BaseModel from typing import Optional from datetime import datetime import uvicorn from app.models import SessionLocal, User, StudyPlan, StudyTask from app.services.supervisor_chain import SupervisorChain app FastAPI(titleAI学习监督官API) # 允许跨域如果前端独立部署 app.add_middleware( CORSMiddleware, allow_origins[*], # 生产环境应指定具体域名 allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 依赖项获取数据库会话 def get_db(): db SessionLocal() try: yield db finally: db.close() # 依赖项获取或创建用户的AI监督链可放入缓存如Redis def get_supervisor_for_user(user_id: int): # 这里简化处理实际应从数据库读取用户信息 # 使用user_id作为标识符创建或获取链实例 chain_key fsupervisor_chain_{user_id} # 伪代码检查缓存若无则创建新实例 # if not cache.exists(chain_key): # cache.set(chain_key, SupervisorChain(user_idstr(user_id))) # return cache.get(chain_key) return SupervisorChain(user_idstr(user_id)) # Pydantic请求/响应模型 class ChatRequest(BaseModel): message: str user_id: int class ChatResponse(BaseModel): reply: str timestamp: datetime class PlanRequest(BaseModel): goal: str weekly_hours: int user_id: int # API端点 app.post(/chat, response_modelChatResponse) async def chat_with_supervisor(request: ChatRequest, db: Session Depends(get_db)): 与AI监督官对话 # 1. 验证用户简化 user db.query(User).filter(User.id request.user_id).first() if not user: raise HTTPException(status_code404, detail用户不存在) # 2. 获取该用户的AI链 supervisor get_supervisor_for_user(request.user_id) # 3. 调用AI ai_reply supervisor.invoke(request.message) # 4. 可选将对话记录存入数据库 # new_session StudySession(user_iduser.id, notesfQ: {request.message}\nA: {ai_reply}) # db.add(new_session) # db.commit() return ChatResponse(replyai_reply, timestampdatetime.utcnow()) app.post(/create_plan, response_modelChatResponse) async def create_study_plan(request: PlanRequest, db: Session Depends(get_db)): 请求AI制定学习计划 user db.query(User).filter(User.id request.user_id).first() if not user: raise HTTPException(status_code404, detail用户不存在) supervisor get_supervisor_for_user(request.user_id) # 调用专门的方法 plan_text supervisor.create_study_plan(request.goal, request.weekly_hours) # 此处可以添加更复杂的逻辑解析AI返回的计划文本结构化后存入数据库的StudyPlan和StudyTask表 # parsed_plan parse_plan_from_text(plan_text) # 需要实现一个解析函数或让AI返回JSON # ... 数据库保存逻辑 ... return ChatResponse(replyplan_text, timestampdatetime.utcnow()) if __name__ __main__: uvicorn.run(app.main:app, host0.0.0.0, port8000, reloadTrue)4.4 第四步添加定时任务与主动监督使用APScheduler实现定时提醒。# app/services/scheduler.py from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger from datetime import datetime import logging from app.models import SessionLocal, User # 假设有一个发送消息的工具函数 from app.utils.notification import send_wechat_message logging.basicConfig() logger logging.getLogger(__name__) def create_daily_morning_reminder(): 创建每日早晨的计划提醒任务 scheduler BackgroundScheduler() # 每天上午8点触发 trigger CronTrigger(hour8, minute0) def job_function(): logger.info(f执行每日提醒任务时间{datetime.now()}) db SessionLocal() try: # 获取所有活跃用户 active_users db.query(User).all() # 生产环境应过滤例如有活跃计划的用户 for user in active_users: # 获取用户今日任务需要实现相关查询 # today_tasks get_today_tasks(db, user.id) # task_list \n.join([f- {t.content} for t in today_tasks]) message f{user.nickname}学员早\n这里是猎鹰教官。\n请汇报今日的学习作战计划。 # 发送提醒这里简化 # send_wechat_message(user.openid, message) logger.info(f已向用户 {user.id} 发送提醒) except Exception as e: logger.error(f定时任务执行失败: {e}) finally: db.close() scheduler.add_job(job_function, trigger, iddaily_morning_reminder) scheduler.start() logger.info(每日早晨提醒调度器已启动) return scheduler # 在应用启动时运行 # scheduler create_daily_morning_reminder()5. 常见问题与排查思路在开发和运行此类AI应用时你会遇到一些典型问题。问题现象可能原因排查与解决思路AI回复不符合“军官”人设1. 系统提示词不够强或被后续对话淹没。2. Temperature参数值过高导致输出随机性太大。3. 上下文历史中包含了太多无关或削弱角色的对话。1. 强化系统提示词在关键指令前后使用###或强调。2. 适当降低Temperature如从0.9调到0.7。3. 使用ConversationBufferWindowMemory限制历史长度或使用ConversationSummaryMemory对历史进行总结保留核心信息。Token消耗过快成本高昂1. 上下文历史过长每次请求都携带全部历史。2. 向量检索返回的文档内容过多。3. 模型选择过大的版本如总是用GPT-4。1. 对长对话历史进行摘要而不是全文传递。2. 优化向量检索的k值只返回最相关的1-3条记忆。3. 非核心创意性对话使用更经济的模型如GPT-3.5-Turbo制定计划或分析时再用强模型。AI忘记之前的约定或计划1. 长期记忆向量存储未正确保存或检索失败。2. 记忆的存储格式不利于检索。1. 检查向量数据库的持久化路径和权限。2. 优化存入长期记忆的内容使用清晰、包含关键实体如“Python学习计划”、“截止日期”的摘要。3. 在提示词中明确要求AI先检索长期记忆再回答。定时任务不执行或重复执行1. APScheduler在多进程/多线程环境下配置不当。2. 服务器时间时区问题。3. 任务函数内部异常导致中断。1. 确保在生产环境如GunicornUvicorn中使用适合的调度器持久化后端如Redis。2. 使用UTC时间并在代码中做好时区转换。3. 在任务函数内部添加完善的try-except日志记录。数据库连接池耗尽1. 每次请求都创建新连接未关闭。2. 定时任务中未正确管理会话生命周期。1. 使用FastAPI的Depends和yield确保请求后关闭会话。2. 在定时任务函数中使用try...finally手动关闭数据库会话。6. 最佳实践与进阶优化6.1 提示词工程优化少样本学习Few-Shot在系统提示词中提供几个高质量的对话示例让AI更好地模仿语气和逻辑。OFFICER_SYSTEM_PROMPT 对话示例 学员教官我今天好累不想学了。 猎鹰疲惫是意志的试金石。你的目标不会因为你的疲惫而改变。我允许你休息20分钟之后必须回到战位完成至少30分钟的基础训练。这是命令。 学员我今天完成了《神经网络》第二章的学习做了课后习题。 猎鹰收到。汇报具体收获和遇到的难点。习题正确率如何我需要评估你的理解深度。 输出结构化对于需要后续程序处理的信息如解析出的计划可以要求AI以特定格式如JSON、YAML输出方便自动化。structured_prompt 请将制定的计划以如下JSON格式输出 { phases: [ { name: 阶段名称, duration_weeks: 2, tasks: [任务1, 任务2] } ], milestones: [里程碑1, 里程碑2] } 6.2 性能与成本优化分层模型使用将任务分类。简单的日常问候、打卡确认使用小模型如gpt-3.5-turbo复杂的计划制定、报告分析使用大模型如gpt-4。流式响应Streaming对于较长的回复使用SSEServer-Sent Events或WebSocket实现流式输出提升用户体验。缓存对常见、通用的回答如“如何开始学习Python”进行缓存避免重复调用LLM。6.3 工程化与部署配置管理将所有配置API Key、模型参数、数据库连接放入环境变量或配置中心。日志与监控记录每一次LLM调用的输入、输出、Token用量和耗时便于分析和优化。错误处理与降级当主要LLM API不可用时应有降级策略如切换备用API、返回预定义的提示信息。容器化使用Docker封装应用确保环境一致性。# Dockerfile 示例 FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 8000]6.4 安全与隐私数据加密用户的学习数据、对话记录在数据库和传输过程中应加密。输入过滤对用户输入进行基本的过滤防止Prompt注入攻击诱导AI执行不当指令。权限控制确保用户只能访问自己的数据和AI实例。构建一个“AI学习监督官”是一次充满乐趣的技术探索它融合了LLM应用开发、提示词工程、数据持久化和系统设计等多个方面。从简单的对话机器人到具备记忆、个性化和主动监督能力的智能体每一步的优化都能带来体验的显著提升。本文提供的代码和架构是一个完整的起点你可以在此基础上继续深化前端界面开发一个微信小程序或网页提供更友好的交互。多模态接入语音输入输出让监督更像真实的通话。数据分析深入分析用户学习数据用图表展示趋势提供更科学的建议。社交功能引入“学习兵团”让多个学员在AI教官的带领下互相监督、竞赛。技术的最终目的是为人服务。希望这个项目不仅能成为你学习路上的“铁面教官”更能成为你探索AI应用边界的一次成功实践。