0. 系列闭环（不公开源码也能跟读）
端到端链路：Vue 前端 → api/routes/chat.py → Guide 多轮 SSE → run_analysis_pipeline（解析→分析→匹配→报告）→ tools/pdf_exporter PDF。
本篇：第 12/17 篇 · 结构化环 · JSON
| 阶段 |
用户可见 |
代码入口 |
对应篇 |
| 建会话 |
欢迎语 |
POST /api/sessions |
09 |
| 多轮对话 |
SSE 流式 |
chat/stream → run_guide_single_turn |
06, 14 |
| 信息充分 |
开始分析 |
_run_analysis_background |
05, 07 |
| 履历解析 |
进度 30% |
run_resume_parser |
12 |
| 画像/RIASEC |
进度 50% |
run_profile_analyzer |
03, 13 |
| 职业匹配 |
进度 70% |
run_career_matcher |
02 |
| 报告 |
进度 90% |
run_reporter |
11 |
| 下载 PDF |
文件 |
GET …/report/pdf |
11, 15 |
|
说明 |
| 读本篇前 |
第 08 篇 invoke_llm_with_json |
| 读完本篇 |
手工走通 parse_json_from_text 四层策略 |
| 下一环 |
第 03/13 篇：业务 JSON schema（第 13 篇） |
全系列闭环索引：SERIES-LOOP.md
1. 要解决什么问题
在 iCan 主流程里，resume_parser_node 要把 Guide 阶段收集的自然语言履历，转成 structured_profile 供 profile_analyzer_node 消费。输入是非结构化文本，输出必须是固定 schema 的 dict。
实际联调时常见失败形态：
- 模型把 JSON 包在
```json 代码块里，或直接混在解释文字后面；
- Ollama 本地模型不支持
response_format={"type": "json_object"}，bind 抛错；
- JSON 语法有小瑕疵（尾逗号、单引号），
json.loads 直接失败；
- LLM 两次调用都返回空 dict，整条解析链路断掉。
iCan 的策略是 Prompt 约束 schema + 调用层 JSON 模式 + 四层文本提取 + 正则兜底，而不是指望模型「一次就完美」。
2. 实现位置
| 模块 |
职责 |
llm/prompts.py |
RESUME_PARSER_SYSTEM_PROMPT：完整 JSON 示例 + 字段规则 |
llm/providers.py |
invoke_llm_with_json：response_format 优先，失败降级 |
llm/parsers.py |
parse_json_from_text 四层提取；validate_structured_profile 校验 |
agents/resume_parser.py |
组装 messages、选 get_light_model()、重试与 _regex_extract_profile 兜底 |
子图顺序（create_resume_parser_graph）：load_input → extract_information → build_profile → validate_profile。
3. Prompt 设计：ResumeParser 的 schema 契约
Prompt 定义在 llm/prompts.py 的 RESUME_PARSER_SYSTEM_PROMPT。核心不是「请输出 JSON」一句话，而是四件事同时写清：
- 完整示例：
basic_info、work_experience、skill_set、certifications、career_progression、parsing_confidence 全字段展示；
- 缺失策略：「如未提及则为 null，不要编造」；
- 推断标注：
parsing_confidence.inferred_fields 列出推断字段；
- 中文与格式：技能区分 technical/soft，多轮对话要整合去重。
Prompt 里嵌了带 ```json 的完整样例——这恰好与 llm/parsers.py 策略 1 的正则 r"```json\s*([\s\S]*?)\s*```" 对齐：模型若照 Prompt 输出代码块，解析器第一层就能命中。
agents/resume_parser.py 的 extract_information 把 system prompt 与用户原文拼成 messages：
1
2
3
4
5
6
| messages = [
{"role": "system", "content": RESUME_PARSER_SYSTEM_PROMPT},
{"role": "user", "content": f"请从以下文本中提取结构化个人信息：\n\n{document_content}"},
]
model = get_light_model()
parsed_data = await invoke_llm_with_json(model, messages)
|
模型选择：履历解析走 get_light_model()（代码默认 LLM_MODEL_LIGHT=gpt-4o-mini），不是 chat 模型。.env 里常见改成 DeepSeek 或 Docker 里的 Ollama qwen3.5:9b——换模型不影响 Prompt/schema，但会影响 JSON 模式兼容性（见踩坑）。
4. 调用层：invoke_llm_with_json 的双通道
llm/providers.py 里 JSON 调用不是简单 ainvoke，而是三层递进：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| try:
json_model = model.bind(response_format={"type": "json_object"})
response = await asyncio.wait_for(json_model.ainvoke(processed, **kwargs), timeout=60)
raw_content = response.content
except Exception as bind_err:
logger.warning("response_format JSON 模式不支持，回退到文本模式: %s", bind_err)
response = await asyncio.wait_for(model.ainvoke(processed, **kwargs), timeout=60)
raw_content = response.content
try:
result = json.loads(raw_content)
except (json.JSONDecodeError, TypeError):
result = parse_json_from_text(raw_content)
if not result:
raise ValueError(f"无法从 LLM 回复中提取有效 JSON，原始内容: {raw_content[:300]}")
|
流程可以概括为：
1
2
3
4
5
6
7
| bind(json_object) → json.loads(content)
↓ 不支持或解析失败
普通 ainvoke → json.loads
↓ 仍失败
parse_json_from_text（四层）
↓ 空 dict
ValueError / 上游重试
|
另外，当 LLM_BASE_URL 含 11434 且模型名含 qwen3 时，_inject_no_think 会在 system 消息前加 /no_think，避免 Qwen3 思考块污染 JSON——这是 JSON 稳定性在本地 Ollama 上的额外一层。
5. 四层降级解析器：parse_json_from_text
llm/parsers.py 的 parse_json_from_text 是最后一道网，按顺序尝试：
| 策略 |
正则/逻辑 |
典型场景 |
| 1 |
r"```json\s*([\s\S]*?)\s*```" |
ChatGPT 风格输出 |
| 2 |
普通 ``` ... ```，内容以 { 或 [ 开头 |
未标注 json 的代码块 |
| 3 |
r"\{[\s\S]*\}" 贪婪匹配最外层花括号 |
「好的，结果如下：{…}」 |
| 4 |
json.loads(text.strip()) |
纯 JSON 回复 |
| 兜底 |
返回 {} |
完全无法解析 |
与通用教程不同，实现里每一层失败不会抛到外层，只在当前策略 json.loads 失败时进入下一层；最外层 JSONDecodeError 也会捕获并返回 {}。这意味着调用方必须检查空 dict——invoke_llm_with_json 会再抛 ValueError，extract_information 则进入重试或正则 fallback。
源码里每层都有 logger.info 标注策略编号（策略1～4），排查时可对照日志确认走到了哪一层。
6. Agent 侧重试与正则兜底
agents/resume_parser.py 的 extract_information 在 LLM 层之上还有业务重试：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| for attempt in range(2):
try:
model = get_light_model()
parsed_data = await invoke_llm_with_json(model, messages)
if parsed_data and len(parsed_data) > 0:
break
logger.warning("[extract_information] 第%d次尝试返回空数据，重试", attempt + 1)
except TimeoutError as te:
...
except Exception as e:
...
if not parsed_data or len(parsed_data) == 0:
parsed_data = _regex_extract_profile(document_content)
|
_regex_extract_profile 用正则抽姓名、学历、工作经历等——字段名与 Prompt schema 不完全一致（例如产出 skills 而非 skill_set.technical_skills）。build_profile 会对缺失 key 填默认空结构，这是刻意的「有总比没有强」，但 validate_profile 大概率仍会报缺失必填字段。
7. 解析后的质量闭环
LLM 自评的 parsing_confidence 在 build_profile 被摘到 confidence_scores；validate_profile 调用 llm/parsers.py 的 validate_structured_profile 做代码侧校验，必填字段包括：
basic_info.education、basic_info.major- 非空
work_experience 列表
skill_set.technical_skills、skill_set.soft_skillscareer_progression.total_years
缺失项写入 parse_errors，validation_passed 写入 confidence_scores。Prompt 里的置信度与 Python 校验是互补的：前者反映模型自评，后者保证下游 Agent 不会收到「空壳 profile」。
8. 在流水线中的位置
顶层 workflow.py：guide_node 信息足够后进入 resume_parser_node，输出 structured_profile 写入 iCanWorkflowState，再交给 profile_analyzer_node。
数据流：
1
2
3
4
5
| Guide 对话文本 (raw_input)
→ run_resume_parser
→ invoke_llm_with_json + parse_json_from_text
→ structured_profile + confidence_scores + parse_errors
→ ProfileAnalyzer
|
同一套 invoke_llm_with_json + parse_json_from_text 也被 ProfileAnalyzer、CareerMatcher 等需要 JSON 的节点复用（详见第 8 篇 LLM 层）；ResumeParser 是 schema 最复杂、兜底链最长的调用点。
9. 踩坑与边界
踩坑 1：response_format 不是通用能力。 Ollama 部分模型 bind 失败会走文本模式，此时更依赖 Prompt 里的 JSON 示例和 parse_json_from_text。Docker 默认 qwen3.5:9b 联调时应在日志里确认是否出现「回退到文本模式」警告。
踩坑 2：策略 3 贪婪匹配可能截错。 \{[\s\S]*\} 从第一个 { 到最后一个 }，若模型在 JSON 前后还嵌入了其他花括号文本，可能整段解析失败并掉进 {}。Prompt 要求「仅输出 JSON」仍必要，解析器不能替代 Prompt 约束。
踩坑 3：正则 fallback 与 schema 不对齐。 _regex_extract_profile 产出 skills 等字段，不会自动映射到 skill_set.technical_skills；下游校验失败是预期行为，应引导用户补充信息或重试 LLM，而不是把 fallback 当成功解析。
踩坑 4：空 dict 与重试。 extract_information 最多 2 次尝试；若 invoke_llm_with_json 返回空 dict（未抛异常），会 warning 后重试。超时 TimeoutError 单独捕获，不会无限阻塞。
10. 小结
- Prompt 用完整 JSON 示例 + null/推断规则锁定 schema，定义在
llm/prompts.py。
llm/providers.py 的 invoke_llm_with_json 先 json_object 模式，不支持则普通调用，再 json.loads → parse_json_from_text。llm/parsers.py 四层递进，失败返回 {}，调用方必须处理空结果。agents/resume_parser.py 用 get_light_model()，并有 2 次重试 + _regex_extract_profile 最后一道兜底。validate_structured_profile 用代码规则校验必填字段，与 parsing_confidence 自评并行。
下一篇进入 RIASEC 测评的 Prompt 工程（第 13 篇）。
附录：关键源码（逐行注释）
以下代码摘自 iCan 实现，每行上方均有中文注释，不公开仓库也可跟读。
生成命令：python3 bin/build-ican-annotated-snippets.py
parse_json_from_text 四层策略
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
def parse_json_from_text(text: str) -> dict:
try:
logger.info(f"[parse_json_from_text] 开始执行，入参: 文本长度={len(text)}")
logger.debug(f"[parse_json_from_text] 文本预览: {text[:300]}")
if not text or not text.strip():
logger.warning("[parse_json_from_text] 入参文本为空，返回空字典")
return {}
json_code_block_pattern = r"```json\s*([\s\S]*?)\s*```"
match = re.search(json_code_block_pattern, text)
if match:
json_str = match.group(1).strip()
logger.debug(f"[parse_json_from_text] 从 json 代码块中提取到内容，长度: {len(json_str)}")
result = json.loads(json_str)
logger.info(f"[parse_json_from_text] 执行完成（策略1: json代码块），返回字段数: {len(result)}")
return result
code_block_pattern = r"```\s*([\s\S]*?)\s*```"
match = re.search(code_block_pattern, text)
if match:
inner = match.group(1).strip()
if inner.startswith("{") or inner.startswith("["):
logger.debug(f"[parse_json_from_text] 从普通代码块中提取到 JSON 内容，长度: {len(inner)}")
result = json.loads(inner)
logger.info(f"[parse_json_from_text] 执行完成（策略2: 普通代码块），返回字段数: {len(result)}")
return result
brace_pattern = r"\{[\s\S]*\}"
match = re.search(brace_pattern, text)
if match:
json_str = match.group(0)
logger.debug(f"[parse_json_from_text] 从文本中直接提取到 JSON 内容，长度: {len(json_str)}")
result = json.loads(json_str)
logger.info(f"[parse_json_from_text] 执行完成（策略3: 直接提取），返回字段数: {len(result)}")
return result
try:
result = json.loads(text.strip())
logger.info(f"[parse_json_from_text] 执行完成（策略4: 直接解析），返回字段数: {len(result)}")
return result
except json.JSONDecodeError:
pass
logger.warning("[parse_json_from_text] 未能从文本中提取到有效 JSON，返回空字典")
return {}
except json.JSONDecodeError as e:
logger.error(f"[parse_json_from_text] JSON 解析失败，异常: {e}", exc_info=True)
return {}
except Exception as e:
logger.error(f"[parse_json_from_text] 提取 JSON 异常: {e}", exc_info=True)
return {}
|
invoke_llm_with_json
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
|
async def invoke_llm_with_json(model: ChatOpenAI, messages: list, **kwargs) -> dict:
import json
from ican.llm.parsers import parse_json_from_text
try:
logger.info(
f"[invoke_llm_with_json] 开始执行，入参: model={model.model_name}，"
f"messages 数量: {len(messages)}，kwargs: {kwargs}"
)
logger.debug(f"[invoke_llm_with_json] 消息详情: {messages}")
processed = _inject_no_think(messages)
raw_content = None
import asyncio as _asyncio
try:
json_model = model.bind(response_format={"type": "json_object"})
try:
response = await _asyncio.wait_for(json_model.ainvoke(processed, **kwargs), timeout=60)
except _asyncio.TimeoutError:
raise TimeoutError("AI 模型响应超时，请稍后重试")
raw_content = response.content
except TimeoutError:
raise
except Exception as bind_err:
logger.warning(
f"[invoke_llm_with_json] response_format JSON 模式不支持，回退到文本模式: {bind_err}"
)
try:
response = await _asyncio.wait_for(model.ainvoke(processed, **kwargs), timeout=60)
except _asyncio.TimeoutError:
raise TimeoutError("AI 模型响应超时，请稍后重试")
raw_content = response.content
logger.debug(
f"[invoke_llm_with_json] 原始回复长度: {len(raw_content) if raw_content else 0}"
)
try:
result = json.loads(raw_content)
except (json.JSONDecodeError, TypeError):
logger.info("[invoke_llm_with_json] 直接 JSON 解析失败，尝试 parse_json_from_text 提取")
result = parse_json_from_text(raw_content)
if not result:
raise ValueError(f"无法从 LLM 回复中提取有效 JSON，原始内容: {raw_content[:300]}")
logger.info(
f"[invoke_llm_with_json] 执行完成，返回 JSON 字段数: {len(result)}"
)
logger.debug(f"[invoke_llm_with_json] 返回 JSON 预览: {str(result)[:300]}")
return result
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
async def extract_information(state: ResumeParserState) -> dict:
try:
logger.info("[extract_information] 开始执行，入参: state=%s", {k: str(v)[:100] for k, v in state.items()})
document_content = state.get("document_content", "")
if not document_content or not document_content.strip():
logger.warning("[extract_information] 文档内容为空，跳过提取")
return {
"parsed_sections": {},
"parse_errors": ["文档内容为空，无法提取信息"],
}
messages = [
{"role": "system", "content": RESUME_PARSER_SYSTEM_PROMPT},
{"role": "user", "content": f"请从以下文本中提取结构化个人信息：\n\n{document_content}"},
]
logger.info("[extract_information] 调用 LLM 提取结构化信息，文档长度: %d", len(document_content))
parsed_data = {}
last_err = None
for attempt in range(2):
try:
model = get_light_model()
parsed_data = await invoke_llm_with_json(model, messages)
if parsed_data and len(parsed_data) > 0:
break
logger.warning("[extract_information] 第%d次尝试返回空数据，重试", attempt + 1)
except TimeoutError as te:
last_err = te
logger.warning("[extract_information] 第%d次 LLM 调用超时: %s", attempt + 1, te)
except Exception as e:
last_err = e
logger.warning("[extract_information] 第%d次 LLM 调用异常: %s", attempt + 1, e)
if not parsed_data or len(parsed_data) == 0:
logger.warning("[extract_information] LLM 提取失败，使用正则 fallback")
parsed_data = _regex_extract_profile(document_content)
logger.info("[extract_information] 结构化数据字段数: %d", len(parsed_data))
logger.debug("[extract_information] 结构化数据预览: %s", json.dumps(parsed_data, ensure_ascii=False)[:500])
result = {
"parsed_sections": parsed_data,
}
logger.info("[extract_information] 执行完成，出参: parsed_sections字段数=%d", len(parsed_data))
return result
except Exception as e:
logger.error("[extract_information] LLM 提取结构化信息异常: %s", e, exc_info=True)
fallback = _regex_extract_profile(state.get("document_content", ""))
if fallback:
logger.info("[extract_information] 使用正则 fallback 提取到 %d 个字段", len(fallback))
return {"parsed_sections": fallback}
return {
"parsed_sections": {},
"parse_errors": [f"LLM 提取信息异常: {str(e)}"],
}
|
系列导航
← 返回 iCan 专题