EduAgent 项目全解析（二）：工程地基——配置中心、LLM 工厂与三层兜底

上一篇介绍了 EduAgent 的整体架构。今天深入项目的地基层看三个贯穿全项目的核心文件config.py配置中心、llm_factory.pyLLM 工厂、retry.py三层兜底。这些代码是理解后面所有 Agent 的基础——所有 Agent 都通过 LLM 工厂拿模型所有图调用都套着三层兜底。一、配置中心pydantic-settings 的正确用法1.1 配置模型backend/config.py继承pydantic-settings的BaseSettings把.env.local里的配置项变成强类型的类属性frompydantic_settingsimportBaseSettingsfromfunctoolsimportlru_cacheclassSettings(BaseSettings):# ── 数据库PostgreSQL──db_host:strlocalhost# 写了默认值 可选项db_port:int5432db_name:streduagentdb_user:str# 没有默认值 必填db_password:strpropertydefdatabase_url(self)-str:把散件拼成 SQLAlchemy 连接串property 可像属性一样访问return(fpostgresqlasyncpg://f{self.db_user}:{self.db_password}f{self.db_host}:{self.db_port}/{self.db_name})# ── 大模型DeepSeek──deepseek_api_key:stros.getenv(DASHSCOPE_API_KEY)deepseek_base_url:strhttps://api.deepseek.com/v1deepseek_model_chat:strdeepseek-v4-flashdeepseek_model_coder:strdeepseek-v4-flashclassConfig:env_fileget_abs_path(.env.local)# 从该文件读取env_file_encodingutf-8case_sensitiveFalseextraignore# 文件里多余的字段不报错lru_cache()defget_settings()-Settings:全局唯一配置对象任何模块都调用这个函数取配置returnSettings()几个值得学的点强类型校验db_port: int 5432—— 如果.env.local里写了DB_PORTabc启动时直接报类型错误把配置错误提前到启动阶段暴露而不是运行到一半才发现。必填 vs 可选写了默认值的字段可选没写默认值的如db_user、db_password、jwt_secret_key是必填缺失会直接报错。lru_cache()单例get_settings()实际只执行一次之后每次调用都返回缓存对象保证全进程共享同一份配置。property组合字段database_url把散件拼成完整连接串调用方写settings.database_url即可不用自己拼。get_abs_path()统一路径基准基于__file__向上推导工程根目录任何相对路径都基于根目录解析避免在哪个目录启动程序导致路径失效的经典坑。小坑提示deepseek_api_key用了os.getenv(DASHSCOPE_API_KEY)——配置文件里的值优先于环境变量BaseSettings的默认优先级顺序这行其实是环境变量兜底注释里也留了 TODODeepSeek 涨价后要换 Qwen 模型改配置即可业务代码零改动——这就是配置中心的价值。二、LLM 工厂统一管理大模型实例2.1 为什么要做工厂如果每个 Agent 都自己调init_chat_model(...)会出现三个问题参数散落各处、模型实例反复创建浪费资源、想统一加日志/换模型很困难。EduAgent 的做法是所有 Agent 必须通过LLMFactory.get_llm(qa)拿模型。2.2 自定义 httpx 客户端绕过系统代理# 异步客户端给 ainvoke/astream 用_HTTP_ASYNC_CLIENThttpx.AsyncClient(trust_envFalse,# ★ 无视系统代理直连 DeepSeektimeouthttpx.Timeout(120.0,connect15.0),# 总超时120s建连超时15s)_HTTP_SYNC_CLIENThttpx.Client(trust_envFalse,timeouthttpx.Timeout(120.0,connect15.0),)init_chat_model底层用 httpx 发请求而 httpx 默认会读系统/环境变量里的代理HTTP_PROXY/HTTPS_PROXY、Windows 系统代理。走代理容易导致SSL/TLS 握手报错而且 DeepSeek 国内能直连所以trust_envFalse强制关掉代理读取——这是很多人在本地调 DeepSeek 报SSL: CERTIFICATE_VERIFY_FAILED的解法之一。2.3 模型路由表与缓存# Agent 类型 → 模型标识符_AGENT_MODEL_ROUTING:dict[str,str]{qa:chat,# 智能问答exam_subjective:chat,# 简答题批改exam_code:code,# 代码题批改resume:chat,# 简历审查interview:chat,# 模拟面试intent:chat,# 意图识别summarize:chat,# 摘要压缩}classLLMFactory:_instances:dict[str,BaseChatModel]{}# 实例缓存classmethoddefget_llm(cls,agent_type,temperature0,streamingFalse,thinkingenabled)-BaseChatModel:ifagent_typenotin_AGENT_MODEL_ROUTING:raiseValueError(f未知 agent_type: {agent_type})model_key_AGENT_MODEL_ROUTING[agent_type]# 缓存键 模型_温度_是否流式_是否thinking不同组合各缓存一份cache_keyf{model_key}_{temperature}_{streaming}_{thinking}ifcache_keynotincls._instances:kwargscls._build_model_kwargs(model_key)kwargs[streaming]streaming kwargs[extra_body]{enable_thinking:thinkingenabled}kwargs[temperature]temperature llminit_chat_model(**kwargs)# ** 把字典展开成关键字参数cls._instances[cache_key]llmreturncls._instances[cache_key]缓存键的设计很讲究f{model_key}_{temperature}_{streaming}_{thinking}—— 同一个 Agent 可能既需要思考模式对话生成又需要非思考模式结构化输出必须区分开否则会拿到错误的模型实例。2.4 结构化输出绑定 Pydantic Schemaclassmethoddefget_structured_llm(cls,agent_type,output_schema:Type[BaseModel],temperature0)-Runnable:# 先拿普通模型结构化输出必须关 thinkingDeepSeek 要求llmcls.get_llm(agent_type,temperaturetemperature,thinkingdisabled)# 绑定 Pydantic 结构methodfunction_calling 是 DeepSeek 必须的returnllm.with_structured_output(output_schema,methodfunction_calling)这个方法的威力在于ainvoke()之后直接返回一个 Pydantic 对象不用手动解析 JSON、不用提心吊胆处理格式错误。后面简历审查的ResumeStructured、试卷批改的SubjectiveReviewResult全都靠它。关键约束DeepSeek 的 function calling 结构化输出必须关闭 thinking 模式注释里写得很清楚——这是踩过坑才有的经验。三、三层兜底永远不让用户拿到空响应生产环境里 LLM API 会超时、Milvus 会断连、网络会抖动。EduAgent 用retry.py实现了一套三层兜底第一层自动重试间隔1s/3s最多重试2次单次超时30s ↓ 重试耗尽 第二层Agent 级降级每个 Agent 有不同兜底策略 ↓ 降级也失败 第三层系统级兜底友好提示永不抛异常3.1 异常分类什么值得重试# 可重试多半是短暂故障重试一下可能就好RETRYABLE_ERRORS(LLMAPIError,MilvusConnectionError,TimeoutError,ConnectionError,)# 不可重试重试也没用输入非法、认证失败应立即抛出NON_RETRYABLE_ERRORS(InvalidInputError,AuthenticationError,)这个分类是整层兜底的基石网络类错误值得重试业务类错误重试只会浪费时间。3.2 装饰器核心逻辑defwith_retry(agent_type:str):defdecorator(func:Callable)-Callable:wraps(func)asyncdefwrapper(*args,**kwargs)-Any:last_errorNone# ── 第一层自动重试attempt 0, 1, 2──forattemptinrange(MAX_RETRIES1):try:resultawaitasyncio.wait_for(func(*args,**kwargs),timeoutTIMEOUT_PER_ATTEMPT,# 30s)returnresultexceptNON_RETRYABLE_ERRORSase:raise# 不可重试立即抛给上层exceptExceptionase:last_erroreifattemptMAX_RETRIES:awaitasyncio.sleep(RETRY_DELAYS[attempt])# 1s/3s# 到上限跳出循环进入降级# ── 第二层Agent 级降级 ──try:returnawaitAgentFallbackHandler.handle(agent_typeagent_type,original_errorlast_error,funcfunc,argsargs,kwargskwargs)exceptException:pass# 降级失败 → 第三层# ── 第三层系统级兜底 ──return_system_fallback_response(agent_type)returnwrapperreturndecorator用法把图调用包进函数再装饰with_retry(agent_typeqa)asyncdef_invoke():returnawaitgraph.ainvoke(initial_state,configconfig)result_stateawait_invoke()3.3 Agent 级降级每个业务有自己的兜底classAgentFallbackHandler:fallback_map{qa:cls._qa_fallback,exam_code:cls._exam_code_fallback,exam_subjective:cls._exam_subjective_fallback,resume:cls._resume_fallback,interview:cls._interview_fallback,}classmethodasyncdef_qa_fallback(cls,error,func,args,kwargs)-dict:问答降级跳过 RAG直接用 LLM 参数知识回答并加 ⚠️ 提示stateargs[0]ifargselsekwargs.get(state,{})messagesstate.get(messages,[])llmget_llm(qa)responseawaitllm.ainvoke(messages)fallback_content(⚠️ 知识库检索暂时不可用以下为 AI 直接生成的回答仅供参考建议与教师确认\n\n(response.contentifhasattr(response,content)elsestr(response)))return{messages:messages[AIMessage(contentfallback_content)],fallback_used:True,# 标记本次走了降级structured_output:None,}各 Agent 的降级策略紧扣业务QA跳检索直答并打 ⚠️ 标记Exam标记需教师人工复核Resume提示服务不可用Interview记录对话、稍后出报告。降级后fallback_usedTrue前端能感知本次结果来自兜底。四、记忆管理MemorySaver 滑动窗口 摘要压缩memory.py管理对话记忆包含三块① 每个 Agent 独立的 MemorySaver_memory_savers:dict[str,MemorySaver]{}defget_memory_saver(agent_typedefault)-MemorySaver:ifagent_typenotin_memory_savers:_memory_savers[agent_type]MemorySaver()return_memory_savers[agent_type]不同 Agent 的 State schema 不同共用同一个 MemorySaver 会在 msgpack 序列化时字段冲突必须按 Agent 隔离。生产环境换成AsyncPostgresSaver即可持久化业务代码零修改。② 滑动窗口裁剪保留最近 N 轮对话SystemMessage始终保留在最前不参与裁剪。③ 摘要压缩对话超过 10 轮时用 LLM 把历史压缩成学员画像摘要保留薄弱点/项目背景丢弃已解决内容支持增量压缩传入上次摘要防止重复记录。五、本篇小结工程地基的三个文件解决三类问题文件解决什么问题核心技巧config.py配置散落、类型不校验BaseSettings强类型 lru_cache单例 get_abs_path统一路径llm_factory.py模型实例重复创建、无法统一管理路由表 组合缓存键 with_structured_outputtrust_envFalseretry.pyLLM 抖动导致用户拿到空响应异常分类 三层兜底装饰器 业务级降级策略下篇预告下一篇《EduAgent 项目全解析三RAG 智能问答 Agent——混合检索、HyDE 与 Multi-Query》是系列的精华篇。我们会看到 QA Agent 的 10 个节点如何在 LangGraph 里编排三层意图分类规则→MiniLM→LLM、BGE-M3 混合检索 WeightedRanker 融合、BGE-Reranker 精排与置信度路由、HyDE 假设文档生成、Multi-Query 子问题改写。项目源码仅供参考欢迎评论区交流讨论。