本系列博客基于一个真实可运行的电商评论舆情分析项目。上一篇《第1篇电商评论舆情分析系统从 0 到 1 的架构设计与数据流》一、痛点开场让模型返回 JSON很容易但让模型稳定返回字段齐全、枚举合法、类型正确的 JSON 很难。新手常见的做法是resultllm.invoke(请分析这条评论的情感返回 JSON)textresult.content datajson.loads(text)# 祈祷它真的是合法 JSONsentimentdata[sentiment_label]# 祈祷字段名真的叫这个然后你就会遇到输出里混着解释文字、字段名从sentiment_label漂移成label、情感值变成Positive而不是positive、score 变成字符串……每个问题都在运行时爆炸。这篇讲我的解法Pydantic Schema with_structured_output(methodfunction_calling) 统一模型工厂。二、核心思路把输出契约变成代码结构化输出的本质是先定义模型必须返回什么再让模型按这个契约输出。在 LangChain 中llm.with_structured_output(schema, methodfunction_calling)会把 Pydantic Schema 转成函数调用参数强制模型按 schema 输出返回的就是BaseModel对象——不需要再手写 JSON 解析。三、Schema 设计ReviewAnalysis这是情感分析的目标输出结构backend/schemas/review.pyfrompydanticimportBaseModel,FieldfromtypingimportLiteral,ListclassAspectScore(BaseModel):name:strField(description属性名如 物流/价格/质量/客服)sentiment:Literal[positive,neutral,negative]score:floatField(ge0.0,le1.0)mentions:List[str]Field(default_factorylist,description原文证据片段)classReviewAnalysis(BaseModel):sentiment_label:Literal[positive,neutral,negative]sentiment_score:floatField(ge0.0,le1.0)emotion:Literal[angry,disappointed,regret,satisfied,praising,neutral]aspects:List[AspectScore]Field(default_factorylist)summary:strField(description一句话摘要)confidence:floatField(default0.9,ge0.0,le1.0)关键点枚举用Literal情感、情绪都锁死取值模型想输出Positive都会校验失败分数用Field(ge0, le1)范围越界直接报错mentions保留原文证据片段为后续验证模型没说谎留了钩子字段都有description这部分会进提示词指导模型理解每个字段含义四、统一模型工厂LLMFactory如果每个 Agent 各自init_chat_model后面切模型、统一超时、统一熔断都会失控。所以项目用LLMFactory收口backend/core/llm_factory.pyclassLLMFactory:_instances:dict[str,BaseChatModel]{}_failure_count:dict[str,int]{}_circuit_open_until:dict[str,datetime]{}_lockthreading.Lock()classmethoddefget_structured_llm(cls,agent_type,output_schema,temperature0):llmcls.get_llm(agent_type,temperaturetemperature,thinkingdisabled)returnllm.with_structured_output(output_schema,methodfunction_calling)模型按(agent_type → model_key, temperature, streaming)缓存实例避免每次请求重复初始化cache_keyf{model_key}_{temperature}_{streaming}ifcache_keynotincls._instances:llminit_chat_model(**kwargs)cls._instances[cache_key]llmreturncls._instances[cache_key]工厂还内置了熔断保护某个 agent_type 连续失败达到阈值后临时拒绝调用llm_circuit_breaker_threshold给外部 LLM 服务喘息时间避免雪崩def_check_circuit(cls,agent_type):untilcls._circuit_open_until.get(agent_type)ifuntilanduntilnow:raiseRuntimeError(fLLM circuit open for{agent_type}, retry in{remaining}s)五、在 Agent 节点里使用情感分析节点backend/agents/sentiment/nodes.py的调用方式frombackend.core.llm_factoryimportget_structured_llmfrombackend.schemas.reviewimportReviewAnalysis# 获取绑定 Schema 的结构化模型structured_llmget_structured_llm(sentiment,ReviewAnalysis)# 每条评论调用messages[SystemMessage(contentSYSTEM_PROMPT),HumanMessage(contentbuild_user_prompt(review)),]try:resultawaitstructured_llm.ainvoke(messages)# result 是 ReviewAnalysis 对象analysisresult.model_dump()analysis[model_name]deepseek# 标记来源exceptExceptionase:analysis_heuristic_analyze(review)# 降级词典规则注意这里没有json.loads——ainvoke返回的就是ReviewAnalysis实例model_dump()直接得到字典。模型输出与 Python 类型之间由 LangChain 的 function calling 桥接。六、为什么 “请返回 JSON” 不可靠问题现象结构化输出解法输出混入解释文字好的分析结果如下{...}function calling 只取工具参数字段名漂移sentiment_label变labelSchema 锁定字段名枚举不合法Positive/负面Literal校验类型错误score 是字符串Pydantic 强类型字段缺失没返回 summarySchema 必填字段校验七、重要提醒Schema 通过 ≠ 语义正确结构化输出只解决长得像不像不解决说得对不对。Schema 校验通过不代表mentions真的来自原文不代表sentiment_label真的符合评论语义不代表根因、摘要没有幻觉所以项目里做了三层防线Schema 层字段、类型、枚举、范围业务层mentions需在原文中出现、簇大小与统计一致、数字与数据库核对HITL 层低置信度 0.75结果进人工复核队列这也是下一篇的主题。八、踩坑记录cache key 维度早期把thinking也放进 cache key导致实例漂移后来简化成(model_key, temperature, streaming)thinking通过extra_body传入max_retries0工厂层故意关闭自动重试把重试策略交给上层统一控制避免重试风暴trust_envFalse关闭代理环境变量干扰保证base_url精确可控HTTP client 共享_HTTP_ASYNC_CLIENT全局复用注意应用关闭时要释放九、总结用 Schema 定义输出契约把模型输出变成类型安全的数据用工厂统一模型路由、缓存、熔断、超时一处管理结构化 ≠ 正确业务校验和 HITL 必须跟上下一篇预告《大模型系统的保命设计LLM 失败降级 HITL 人工复核闭环》
第2篇：LLM 结构化输出实战：Pydantic + Function Calling 告别“解析 JSON 地狱“
本系列博客基于一个真实可运行的电商评论舆情分析项目。上一篇《第1篇电商评论舆情分析系统从 0 到 1 的架构设计与数据流》一、痛点开场让模型返回 JSON很容易但让模型稳定返回字段齐全、枚举合法、类型正确的 JSON 很难。新手常见的做法是resultllm.invoke(请分析这条评论的情感返回 JSON)textresult.content datajson.loads(text)# 祈祷它真的是合法 JSONsentimentdata[sentiment_label]# 祈祷字段名真的叫这个然后你就会遇到输出里混着解释文字、字段名从sentiment_label漂移成label、情感值变成Positive而不是positive、score 变成字符串……每个问题都在运行时爆炸。这篇讲我的解法Pydantic Schema with_structured_output(methodfunction_calling) 统一模型工厂。二、核心思路把输出契约变成代码结构化输出的本质是先定义模型必须返回什么再让模型按这个契约输出。在 LangChain 中llm.with_structured_output(schema, methodfunction_calling)会把 Pydantic Schema 转成函数调用参数强制模型按 schema 输出返回的就是BaseModel对象——不需要再手写 JSON 解析。三、Schema 设计ReviewAnalysis这是情感分析的目标输出结构backend/schemas/review.pyfrompydanticimportBaseModel,FieldfromtypingimportLiteral,ListclassAspectScore(BaseModel):name:strField(description属性名如 物流/价格/质量/客服)sentiment:Literal[positive,neutral,negative]score:floatField(ge0.0,le1.0)mentions:List[str]Field(default_factorylist,description原文证据片段)classReviewAnalysis(BaseModel):sentiment_label:Literal[positive,neutral,negative]sentiment_score:floatField(ge0.0,le1.0)emotion:Literal[angry,disappointed,regret,satisfied,praising,neutral]aspects:List[AspectScore]Field(default_factorylist)summary:strField(description一句话摘要)confidence:floatField(default0.9,ge0.0,le1.0)关键点枚举用Literal情感、情绪都锁死取值模型想输出Positive都会校验失败分数用Field(ge0, le1)范围越界直接报错mentions保留原文证据片段为后续验证模型没说谎留了钩子字段都有description这部分会进提示词指导模型理解每个字段含义四、统一模型工厂LLMFactory如果每个 Agent 各自init_chat_model后面切模型、统一超时、统一熔断都会失控。所以项目用LLMFactory收口backend/core/llm_factory.pyclassLLMFactory:_instances:dict[str,BaseChatModel]{}_failure_count:dict[str,int]{}_circuit_open_until:dict[str,datetime]{}_lockthreading.Lock()classmethoddefget_structured_llm(cls,agent_type,output_schema,temperature0):llmcls.get_llm(agent_type,temperaturetemperature,thinkingdisabled)returnllm.with_structured_output(output_schema,methodfunction_calling)模型按(agent_type → model_key, temperature, streaming)缓存实例避免每次请求重复初始化cache_keyf{model_key}_{temperature}_{streaming}ifcache_keynotincls._instances:llminit_chat_model(**kwargs)cls._instances[cache_key]llmreturncls._instances[cache_key]工厂还内置了熔断保护某个 agent_type 连续失败达到阈值后临时拒绝调用llm_circuit_breaker_threshold给外部 LLM 服务喘息时间避免雪崩def_check_circuit(cls,agent_type):untilcls._circuit_open_until.get(agent_type)ifuntilanduntilnow:raiseRuntimeError(fLLM circuit open for{agent_type}, retry in{remaining}s)五、在 Agent 节点里使用情感分析节点backend/agents/sentiment/nodes.py的调用方式frombackend.core.llm_factoryimportget_structured_llmfrombackend.schemas.reviewimportReviewAnalysis# 获取绑定 Schema 的结构化模型structured_llmget_structured_llm(sentiment,ReviewAnalysis)# 每条评论调用messages[SystemMessage(contentSYSTEM_PROMPT),HumanMessage(contentbuild_user_prompt(review)),]try:resultawaitstructured_llm.ainvoke(messages)# result 是 ReviewAnalysis 对象analysisresult.model_dump()analysis[model_name]deepseek# 标记来源exceptExceptionase:analysis_heuristic_analyze(review)# 降级词典规则注意这里没有json.loads——ainvoke返回的就是ReviewAnalysis实例model_dump()直接得到字典。模型输出与 Python 类型之间由 LangChain 的 function calling 桥接。六、为什么 “请返回 JSON” 不可靠问题现象结构化输出解法输出混入解释文字好的分析结果如下{...}function calling 只取工具参数字段名漂移sentiment_label变labelSchema 锁定字段名枚举不合法Positive/负面Literal校验类型错误score 是字符串Pydantic 强类型字段缺失没返回 summarySchema 必填字段校验七、重要提醒Schema 通过 ≠ 语义正确结构化输出只解决长得像不像不解决说得对不对。Schema 校验通过不代表mentions真的来自原文不代表sentiment_label真的符合评论语义不代表根因、摘要没有幻觉所以项目里做了三层防线Schema 层字段、类型、枚举、范围业务层mentions需在原文中出现、簇大小与统计一致、数字与数据库核对HITL 层低置信度 0.75结果进人工复核队列这也是下一篇的主题。八、踩坑记录cache key 维度早期把thinking也放进 cache key导致实例漂移后来简化成(model_key, temperature, streaming)thinking通过extra_body传入max_retries0工厂层故意关闭自动重试把重试策略交给上层统一控制避免重试风暴trust_envFalse关闭代理环境变量干扰保证base_url精确可控HTTP client 共享_HTTP_ASYNC_CLIENT全局复用注意应用关闭时要释放九、总结用 Schema 定义输出契约把模型输出变成类型安全的数据用工厂统一模型路由、缓存、熔断、超时一处管理结构化 ≠ 正确业务校验和 HITL 必须跟上下一篇预告《大模型系统的保命设计LLM 失败降级 HITL 人工复核闭环》
