feat: 添加结构化日志系统,更新LLM配置与全部文档

新增:
- backend/logger.py — 集中日志模块 (JSON格式 + trace_id + 独立llm.log)
- @log_node / @_log_route 装饰器覆盖17个节点和8个路由

改进:
- backend/llm.py — _LLMLoggingWrapper 自动记录LLM输入输出
- backend/llm.py — API Key优先读ANTHROPIC_API_KEY,模型名改为MiniMax-M2.7
- backend/llm.py — get_llm() 新增caller参数标识调用来源
- backend/validation.py — 新增验证结果/连接失败日志
- backend/session.py — 新增会话创建/删除日志
- app.py — 新增用户交互日志 (输入/执行/异常/会话操作)
- app.py — 提前导入torchvision抑制transformers懒加载报错
- .env.example — 新增LOG_DIR/LOG_LEVEL/ANTHROPIC_API_KEY等配置项
- .gitignore — 新增logs/和db/忽略规则

文档:
- ROADMAP.md — 新增阶段四: 可观测性
- README.md — 补充日志架构/LLM配置/项目结构
- CLAUDE.md — 同步最新配置/日志/MAX_RETRY(3)
- CODE_GUIDE.md — 新增第15章日志系统,更新架构图/LLM/配置
This commit is contained in:
2026-05-19 23:40:01 +08:00
parent 6467fd4ae5
commit 067880bf2e
13 changed files with 753 additions and 82 deletions
+4
View File
@@ -54,6 +54,10 @@ CONTEXT_KEEP_RECENT=4
# 会话持久化目录
SESSIONS_DIR=./sessions
# 日志目录和级别
LOG_DIR=./logs
LOG_LEVEL=DEBUG
# 状态快照保留数量(用于撤销操作)
HISTORY_MAX_SNAPSHOTS=10
+7 -1
View File
@@ -11,7 +11,8 @@ dist/
# 数据库
db/chroma/
sessions/
logs/
db/
# RAG 管线中间产物 (rag 子模块内)
rag/jrxml_chunker_output/
rag/embeddings/
@@ -29,3 +30,8 @@ rag/jrxml_source_chunks/
# 系统文件
Thumbs.db
.DS_Store
db/chroma-bak/chroma.sqlite3
db/chroma-bak/ec8b65c1-913e-4fa3-b073-b7663c97cf15/data_level0.bin
db/chroma-bak/ec8b65c1-913e-4fa3-b073-b7663c97cf15/header.bin
db/chroma-bak/ec8b65c1-913e-4fa3-b073-b7663c97cf15/length.bin
db/chroma-bak/ec8b65c1-913e-4fa3-b073-b7663c97cf15/link_lists.bin
+19 -8
View File
@@ -4,7 +4,7 @@
一个**本地桌面应用**,通过自然语言多轮对话帮助非技术用户创建 JasperReports 模板(JRXML 文件)。核心技术栈:Streamlit UI + LangGraph 状态机 + LLM 生成/修改 + 自动验证修正循环。
**一句话**:用户用中文描述报表需求 → LLM 生成 JRXML → 自动验证 → 失败则自动修正(最多5次) → 重试耗尽后失败上下文自动注入下一轮 → 返回可编译的 JRXML 文件。
**一句话**:用户用中文描述报表需求 → LLM 生成 JRXML → 自动验证 → 失败则自动修正(最多3次) → 重试耗尽后失败上下文自动注入下一轮 → 返回可编译的 JRXML 文件。
## 启动命令
@@ -23,11 +23,12 @@ STREAMLIT_SERVER_HEADLESS=true streamlit run app.py --server.port 8501
- **OCR**: EasyOCR(优先,ch_sim+en)→ PaddleOCR(回退),两者均未安装时仅返回图片元信息
- **LLM**: `cloud` / `anthropic` → MiniMax Anthropic 兼容 API (`MiniMax-M2.7`)
- Base URL: `https://api.minimaxi.com/anthropic`
- 认证: 通过 `OPENAI_API_KEY` 传入 Anthropic SDK(注意不是 `ANTHROPIC_API_KEY`
- 绕过代理: 代码中设 `NO_PROXY=*`
- 认证: Anthropic SDK 自动读取 `ANTHROPIC_API_KEY`fallback `OPENAI_API_KEY`
- **嵌入模型**: `local` / `sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2`
- **向量库**: ChromaDB 持久化在 `./db/chroma`
- **验证服务**: FastAPI `localhost:8001`
- **日志**: JSON 格式化,`logs/app.log` + `logs/llm.log`,中国时区 (UTC+8)
- **MAX_RETRY**: 3
## 架构
@@ -47,10 +48,11 @@ agent/graph.py (LangGraph 状态机)
│ 验证修正循环: validate ─fail─► explain_error ─► correct_jrxml ─► validate
│ ▲ │
│ └──────── (retry < MAX_RETRY=5) ───────────────────┘
│ └──────── (retry < MAX_RETRY=3) ───────────────────┘
├──► prompts/loader.py Prompt 外部化:7 个 .md 文件热重载
├──► backend/llm.py LLM 工厂: Anthropic SDK / OpenAI / Ollama (统一 stream/invoke)
├──► backend/logger.py 集中日志: JSON + trace_id + llm.log/app.log 分离
├──► backend/rag_adapter.py 语义搜索: ChromaDB + SentenceTransformer
├──► backend/error_kb.py 错误知识库: 指纹去重 + ChromaDB 持久化
├──► backend/file_parser.py 文件解析: PDF/DOCX/图片/文本
@@ -70,7 +72,8 @@ agent/graph.py (LangGraph 状态机)
| `agent/graph.py` | 状态图编译 + 路由函数(预览跳过验证) | 中 |
| `prompts/loader.py` | Prompt 加载器(从 .md 文件热重载) | 低 |
| `prompts/*.md` | 7 个独立 Prompt 模板 | **高** |
| `backend/llm.py` | LLM 工厂,统一 `_BaseLLM` 接口(invoke + stream | 中 |
| `backend/llm.py` | LLM 工厂,统一 `_BaseLLM` 接口(invoke + stream+ `_LLMLoggingWrapper` | 中 |
| `backend/logger.py` | 集中日志模块:JSON 格式化 + trace_id + 独立 llm.log | 低 |
| `backend/rag_adapter.py` | RAGSearcher 单例,语义搜索接口 | 中 |
| `backend/error_kb.py` | ErrorKB — 错误指纹去重 + ChromaDB 持久化 + 语义检索 | 中 |
| `backend/file_parser.py` | 文件解析: PDF/DOCX/图片(EasyOCR→PaddleOCR回退)/文本 | 中 |
@@ -147,14 +150,22 @@ agent/graph.py (LangGraph 状态机)
### Ctrl+C 修复
- JS 注入拦截 Streamlit 裸 `c` 键清缓存,保留 Ctrl+C 复制
### 结构化日志系统
- `backend/logger.py` — JSON 格式化 + trace_id + 国际时区
- `_LLMLoggingWrapper` — 包装所有 LLM 后端,记录完整 prompt/response
- `@log_node` / `@_log_route` — 装饰器自动记录节点和路由
- 日志分离: `logs/app.log` (业务) + `logs/llm.log` (AI 调用)
## 已知注意点
- **Anthropic SDK**: 使用原始 `anthropic` 包(非 `langchain-anthropic`),因为需要直连 MiniMax 兼容端点。`backend/llm.py:31` 创建的 `Anthropic()` 必须传`api_key`SDK 不会自动读 `OPENAI_API_KEY`
- **Windows 环境**: NO_PROXY 设为 `*` 避免代理干扰 MiniMax API
- **Anthropic SDK**: 使用原始 `anthropic` 包(非 `langchain-anthropic`),因为需要直连 MiniMax 兼容端点。API Key 优先读 `ANTHROPIC_API_KEY`fallback `OPENAI_API_KEY`。Anthropic SDK 会自动将 key 放`x-api-key` header
- **MiniMax 模型名称**: `MiniMax-M2.7`(不是 `minimax-2.7`),大小写敏感
- **Streamlit headless**: Windows 下必须设 `STREAMLIT_SERVER_HEADLESS=true` 跳过邮箱采集提示。
- **日志分析**: 通过 `trace_id` 字段可追踪一次请求的全链路。LLM 调用日志在 `logs/llm.log`,包含完整 prompt 和 response(各截断 10000 字符)。
- **验证服务结构检查**: 字段引用一致性 (`$F{field}` vs `<field>` 声明)、SQL SELECT 存在性、pageWidth/pageHeight/name 属性。
- **XSD 校验可选**: 需要 `validation_service/schemas/jasperreport_7_0_6.xsd` 存在。
- **rag 子模块**: 内部有独立的管线脚本(`batch_chunker.py``embed_chunks.py``import_to_chroma.py`),通常不需要在主项目中运行。
- **OCR 引擎**: 优先使用 EasyOCRWindows 兼容性更好,`pip install easyocr`),回退 PaddleOCR。两者均未安装时仅返回图片元信息,建议至少安装 EasyOCR。
- **MAX_RETRY**: 默认 5 次。重试耗尽后 `pending_failure_context` 记录失败信息,下次用户输入时自动注入。
- **MAX_RETRY**: 默认 3 次。重试耗尽后 `pending_failure_context` 记录失败信息,下次用户输入时自动注入。
- **验证最小内容检查**: 验证服务额外检查至少 1 个 `<band>` + 1 个 `<textField>``<staticText>`,拦截空壳 JRXML。
- **torchvision**: `transformers` 库的懒加载需要 `torchvision`,已作为依赖安装。
+145 -42
View File
@@ -29,7 +29,7 @@
## 1. 项目是什么
**一句话**:用户用中文描述报表需求 → LLM 生成 JRXML 模板 → 自动验证 → 失败则自动修正(最多 5 次)→ 重试耗尽后失败上下文自动注入下一轮 → 返回可用的 JRXML 文件。
**一句话**:用户用中文描述报表需求 → LLM 生成 JRXML 模板 → 自动验证 → 失败则自动修正(最多 3 次)→ 重试耗尽后失败上下文自动注入下一轮 → 返回可用的 JRXML 文件。
**技术栈**StreamlitUI + LangGraph(状态机) + LLMMiniMax/OpenAI/Ollama + ChromaDB(向量库) + FastAPI(验证微服务)
@@ -114,9 +114,10 @@ streamlit run app.py --server.port 8501
┌──────────┐ ┌──────────────┐ ┌───────────────┐
│backend/ │ │prompts/ │ │validation_ │
│llm.py │ │loader.py │ │service/main.py│
rag_ │ │*.md (7个 │ │(FastAPI, │
adapter.py│ │Prompt模板) │ │独立进程) │
error_kb │ └──────────────┘ └───────────────┘
logger.py │ │*.md (7个 │ │(FastAPI, │
rag_ │ │Prompt模板) │ │独立进程) │
adapter.py│ └──────────────┘ └───────────────┘
│error_kb │
│.py │
│embeddings│
│.py │
@@ -133,7 +134,7 @@ streamlit run app.py --server.port 8501
### 关键设计决策
1. **LLM 调用不经过 LangChain**`backend/llm.py` 直接使用 Anthropic SDK 和 OpenAI SDK,仅 Ollama 保留 langchain-ollama 包装。原因是 Anthropic SDK 需要透传 `api_key` 到 MiniMax 兼容端点
1. **LLM 调用不经过 LangChain**`backend/llm.py` 直接使用 Anthropic SDK 和 OpenAI SDK,仅 Ollama 保留 langchain-ollama 包装。所有 LLM 调用通过 `_LLMLoggingWrapper` 自动记录输入输出到 `logs/llm.log`
2. **Prompt 热重载**`prompts/loader.py` 每次都从磁盘读取 `.md` 文件,修改 Prompt 无需重启。
@@ -141,6 +142,8 @@ streamlit run app.py --server.port 8501
4. **验证服务独立进程**FastAPI 运行在 8001 端口,主进程通过 HTTP 调用。这样可以把 Java 编译验证加进去而不影响 UI 进程。
5. **结构化日志**`backend/logger.py` 提供 JSON 格式化日志,`trace_id` 通过 contextvars 贯穿全链路,LLM 调用与业务日志分离。
---
## 4. 数据总线:AgentState
@@ -217,7 +220,7 @@ def route_after_correct(state) -> Literal["validate", "finalize"]:
return "validate" if state.get("retry_count", 0) < MAX_RETRY else "finalize"
```
**MAX_RETRY 默认为 5**`.env` 中配置)。重试耗尽后进入 finalize,finalize 会将失败上下文写入 `pending_failure_context`,下次用户输入时 `process_input` 自动注入。
**MAX_RETRY 默认为 3**`.env` 中配置)。重试耗尽后进入 finalize,finalize 会将失败上下文写入 `pending_failure_context`,下次用户输入时 `process_input` 自动注入。
```
**关键路由逻辑**
@@ -503,12 +506,21 @@ class _BaseLLM:
raise NotImplementedError
```
### 7.1 三个实现
### 7.1 日志包装器
所有 LLM 实例都通过 `_LLMLoggingWrapper` 包装,自动记录:
- 请求 prompt(完整内容,截断 10000 字符)
- 响应内容(完整内容,截断 10000 字符)
- 调用耗时(毫秒)
- 模型名称、后端、调用来源(caller 参数)
日志输出到 `logs/llm.log`(独立于业务日志)。
### 7.2 三个实现
**MiniMaxLLM**`LLM_PROVIDER=anthropic`):
- 使用原始 `anthropic` SDK`from anthropic import Anthropic`
- `api_key` 必须显式传入构造函数(SDK 不会自动读 `OPENAI_API_KEY`
- `NO_PROXY=*` 绕过 Windows 代理
- API Key 优先读 `ANTHROPIC_API_KEY`fallback `OPENAI_API_KEY`
- `invoke()` 遍历 `resp.content``type == "text"` 的 block
- `stream()` 使用 `client.messages.stream()` + `s.text_stream`
@@ -520,11 +532,11 @@ class _BaseLLM:
- 使用 langchain-ollama 的 `ChatOllama`
- 本地运行,无需网络
### 7.2 调用约定
### 7.3 调用约定
所有节点统一使用:
所有节点统一使用,传入 `caller` 参数标识调用来源
```python
llm = get_llm()
llm = get_llm(caller="classify_intent")
# 同步
resp = llm.invoke(prompt)
text = resp.content.strip()
@@ -815,11 +827,90 @@ generate_session_id() → str # UUID hex[:12]
---
## 15. Streamlit UIapp.py
## 15. 日志系统:logger.py
`app.py` 是整个系统的入口,约 500 行。分为几个区域:
`backend/logger.py` 提供结构化日志能力,是整个系统的"黑匣子"。
### 15.1 组件树
### 15.1 架构设计
```
backend/logger.py
├── JsonFormatter JSON 单行格式化,自动收集 extra 字段
├── get_logger(name) 获取 loggername="llm" → llm.log,其他 → app.log
├── generate_trace_id() 生成 16 位 hex trace_id
├── set_trace_id(tid) 通过 contextvars 设置当前请求的 trace_id
└── get_trace_id() 获取当前 trace_id(自动跨线程/协程传播)
```
### 15.2 日志文件
| 文件 | 对应 logger | 内容 |
|------|-----------|------|
| `logs/app.log` | `get_logger("agent")`, `get_logger("app")`, `get_logger("session")`, `get_logger("validation")` | 节点流转、路由决策、用户交互、会话操作、验证结果 |
| `logs/llm.log` | `get_logger("llm")` | LLM 请求 prompt、响应内容、耗时、异常 |
### 15.3 日志格式
每条日志是单行 JSON
```json
{
"timestamp": "2026-05-19T23:05:22.877+08:00",
"level": "INFO",
"logger": "jrxml.agent",
"trace_id": "b29010ab4a014249",
"message": "[节点入口] classify_intent",
"module": "nodes",
"function": "wrapper",
"line": 53,
"extra": {
"node": "classify_intent",
"phase": "entry",
"state": {
"session_id": "681e55231bab",
"intent": "",
"has_jrxml": false,
"retry_count": 0
}
}
}
```
### 15.4 trace_id 机制
每次在 [app.py](file:///d:/Idea%20Project/jaspersoft/app.py) 的 `run_agent()` 中调用 `set_trace_id(generate_trace_id())`,后续所有节点、路由、LLM 调用都自动带上同一个 trace_id。通过 `grep "b29010ab4a014249" logs/*.log` 可还原一次请求的完整链路。
### 15.5 `@log_node` 装饰器
[agent/nodes.py](file:///d:/Idea%20Project/jaspersoft/agent/nodes.py) 中 17 个节点均使用 `@log_node("节点名")` 装饰器,自动记录:
- **入口日志** — 节点开始执行时的 state 摘要
- **出口日志** — 节点完成时的 state 摘要 + 耗时 (duration_ms)
- **异常日志** — 节点抛异常时的错误信息 + state 摘要
### 15.6 `@_log_route` 装饰器
[agent/graph.py](file:///d:/Idea%20Project/jaspersoft/agent/graph.py) 中 8 个路由函数均使用 `@_log_route("路由名")`,自动记录每次路由决策(from → to)。
### 15.7 日志分析示例
```bash
# 按 trace_id 追踪一次完整请求
jq 'select(.trace_id=="b29010ab4a014249")' logs/app.log
# 统计各节点平均耗时
jq 'select(.extra.phase=="exit") | {node: .extra.node, ms: .extra.duration_ms}' logs/app.log | jq -s 'group_by(.node) | map({node: .[0].node, avg_ms: (map(.ms) | add / length)})'
# 查看所有 LLM 调用耗时
jq 'select(.extra.direction=="response") | {caller: .extra.caller, ms: .extra.duration_ms}' logs/llm.log
```
---
## 16. Streamlit UIapp.py
`app.py` 是整个系统的入口,约 560 行。分为几个区域:
### 16.1 组件树
```
st.set_page_config (wide layout)
@@ -842,7 +933,7 @@ st.set_page_config (wide layout)
└── 触发 run_agent(full_prompt)
```
### 15.2 run_agent() — 核心渲染函数
### 16.2 run_agent() — 核心渲染函数
```python
def run_agent(user_input):
@@ -859,7 +950,7 @@ def run_agent(user_input):
# 6. 渲染总结卡片:用 agent_state(完整状态)而非 node_state(仅含变更字段)
```
### 15.3 文件上传流程
### 16.3 文件上传流程
```
用户选择文件
@@ -883,7 +974,7 @@ template_type?
清空 uploaded_files
```
### 15.4 流式渲染细节
### 16.4 流式渲染细节
```python
# 在 graph.stream 循环中:
@@ -896,7 +987,7 @@ elif mode == "custom":
这里 `streaming_placeholder.code()` 每次都会被覆盖,显示累积到当前的所有文本,产生"逐字打字"的视觉效果。
### 15.5 Ctrl+C 修复
### 16.5 Ctrl+C 修复
Streamlit 默认会在非输入元素上按 `c` 键清除缓存。注入 JS 拦截裸 `c` 键(不含 Ctrl/Alt/Meta):
@@ -912,7 +1003,7 @@ parent.addEventListener('keydown', function(e) {
---
## 16. 配置参考
## 17. 配置参考
所有配置通过 `.env` 文件管理。完整配置项:
@@ -920,10 +1011,12 @@ parent.addEventListener('keydown', function(e) {
|------|--------|------|
| `LLM_BACKEND` | `cloud` | `cloud``local` |
| `LLM_PROVIDER` | `openai` | `openai``anthropic` |
| `LLM_MODEL` | `gpt-4o` | 云端模型名 |
| `LLM_MODEL` | `MiniMax-M2.7` | 云端模型名 |
| `LOCAL_LLM_MODEL` | `qwen2.5-coder:7b` | Ollama 模型名 |
| `OPENAI_API_KEY` | — | API 密钥(Anthropic 模式也用这个 |
| `OPENAI_BASE_URL` | `https://api.openai.com/v1` | API 端点 |
| `OPENAI_API_KEY` | — | API 密钥(Anthropic 模式的 fallback |
| `ANTHROPIC_API_KEY` | — | Anthropic 兼容 API 密钥(优先) |
| `OPENAI_BASE_URL` | `https://api.openai.com/v1` | OpenAI API 端点 |
| `ANTHROPIC_BASE_URL` | `https://api.minimaxi.com/anthropic` | Anthropic 兼容端点 |
| `EMBED_BACKEND` | `local` | `local``cloud` |
| `RAG_EMBED_MODEL` | `sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2` | 嵌入模型 |
| `RAG_CHROMA_PATH` | `./db/chroma` | ChromaDB 存储路径 |
@@ -931,17 +1024,19 @@ parent.addEventListener('keydown', function(e) {
| `RAG_USE_GPU` | `true` | GPU 加速 |
| `RAG_USE_FP16` | `true` | 半精度推理 |
| `VALIDATION_SERVICE_URL` | `http://localhost:8001/validate` | 验证服务地址 |
| `MAX_RETRY` | `5` | 最大自动修正次数 |
| `MAX_RETRY` | `3` | 最大自动修正次数 |
| `CONTEXT_MAX_TOKENS` | `6000` | 触发压缩的 token 阈值 |
| `CONTEXT_KEEP_RECENT` | `4` | 压缩时保留最近 N 轮 |
| `SESSIONS_DIR` | `./sessions` | 会话文件目录 |
| `HISTORY_MAX_SNAPSHOTS` | `10` | 撤销快照栈深度 |
| `LOG_DIR` | `./logs` | 日志目录 |
| `LOG_LEVEL` | `DEBUG` | 日志级别 (DEBUG/INFO/WARNING/ERROR) |
---
## 17. 如何添加新功能
## 18. 如何添加新功能
### 17.1 添加新的意图类型
### 18.1 添加新的意图类型
假设要添加"导出 Excel"功能:
@@ -951,7 +1046,7 @@ parent.addEventListener('keydown', function(e) {
4. **`app.py`** — 侧边栏添加快捷按钮
5. 如果 Excel 导出的逻辑复杂,可以新增节点 `handle_export_excel`
### 17.2 添加新的 LLM 后端
### 18.2 添加新的 LLM 后端
`backend/llm.py``get_llm()` 中添加新的 provider
@@ -965,7 +1060,7 @@ elif provider == "my_provider":
return MyProviderLLM()
```
### 17.3 添加新的文件格式支持
### 18.3 添加新的文件格式支持
`backend/file_parser.py` 中:
@@ -973,19 +1068,19 @@ elif provider == "my_provider":
2. 实现对应的 `_parse_xxx()` 函数
3.`app.py``file_uploader``type` 参数中加入新后缀
### 17.4 添加新的验证规则
### 18.4 添加新的验证规则
`validation_service/main.py``_check_structural_issues()` 中添加检查逻辑,返回描述问题的人类可读字符串即可。
### 17.5 修改 Prompt
### 18.5 修改 Prompt
直接编辑 `prompts/*.md`,保存后立即生效。Prompt 使用 Python `str.format()` 占位符,变量名必须与节点中 `.format()` 的参数名一致。
---
## 18. 调试指南
## 19. 调试指南
### 18.1 常见问题
### 19.1 常见问题
**Q: 验证服务连接失败**
```
@@ -1013,20 +1108,28 @@ elif provider == "my_provider":
**Q: 修改了 nodes.py 但不生效**
→ Streamlit 有热重载,保存文件后刷新浏览器即可。如果改的是 Prompt 文件,下次请求自动生效,无需做任何操作。
### 18.2 日志调试
### 19.2 日志调试
当前项目没有集中的日志系统。最简单的方式是在节点中加 `print()`
项目已集成结构化日志系统(详见第 15 章)。调试时
```python
# 在 nodes.py 中
def generate(state):
print(f"[DEBUG] Prompt length: {len(prompt)}")
print(f"[DEBUG] Generated JRXML length: {len(jrxml)}")
```bash
# 实时查看日志
tail -f logs/app.log | jq .
tail -f logs/llm.log | jq .
# 按 trace_id 追踪一次完整请求
jq 'select(.trace_id=="abc123")' logs/app.log
# 查看最近 5 次 LLM 调用
tail -5 logs/llm.log | jq '{caller: .extra.caller, model: .extra.model, ms: .extra.duration_ms}'
# 查看错误日志
jq 'select(.level=="ERROR")' logs/app.log
```
`print()` 输出会出现在 Streamlit 终端(终端 2
也可直接在 Streamlit 终端(终端 2添加 `print()` 快速调试
### 18.3 状态检查
### 19.3 状态检查
`app.py``run_agent()` 完成后,`st.session_state.agent_state` 包含最新状态。可以通过 Streamlit 的 `st.write()` 临时打印:
@@ -1035,7 +1138,7 @@ def generate(state):
st.json(state) # 打印完整状态(调试用,记得删除)
```
### 18.4 关键数据点
### 19.4 关键数据点
调试时最常需要检查的数据:
+22 -4
View File
@@ -111,21 +111,32 @@ jrxml-agent/
nodes.py 图节点(generate, validate, modify 等)
graph.py LangGraph 状态机
backend/
llm.py LLM 工厂(OpenAI / Ollama
llm.py LLM 工厂(Anthropic SDK / OpenAI / Ollama
logger.py 集中日志模块(JSON + trace_id
embeddings.py 嵌入模型工厂
validation.py 验证服务客户端
rag_adapter.py RAG 语义搜索适配器
error_kb.py 错误自增长知识库
file_parser.py 文件解析器(PDF/DOCX/图片)
layout_analyzer.py A4 模板布局分析
session.py 会话持久化 CRUD
prompts/
loader.py Prompt 加载器(热重载)
*.md 7 个 Prompt 模板文件
validation_service/
main.py FastAPI 验证服务器
validate.bat Windows 启动器
data/
sample_templates/ 知识库的 JRXML 模板
corrections/ 错误修正案例
logs/
app.log 应用日志(节点流转、路由、用户交互)
llm.log LLM 调用日志(完整 prompt / response
scripts/
init_kb.py Chroma 知识库初始化脚本
tests/
test_validation.py 验证服务测试
test_agent.py 代理集成测试
db/chroma/ Chroma 持久化目录
requirements.txt
.env.example
README.md
@@ -136,12 +147,19 @@ jrxml-agent/
| 变量 | 描述 | 默认值 |
|----------|-------------|---------|
| LLM_BACKEND | cloud 或 local | cloud |
| OPENAI_API_KEY | OpenAI API 密钥 | - |
| LLM_PROVIDER | openai 或 anthropic | openai |
| OPENAI_API_KEY | API 密钥(OpenAI 或 MiniMax | - |
| OPENAI_BASE_URL | API 基础 URL | https://api.openai.com/v1 |
| LLM_MODEL | 模型名称 | gpt-4o |
| ANTHROPIC_API_KEY | Anthropic 兼容 API 密钥(优先) | - |
| ANTHROPIC_BASE_URL | Anthropic 兼容 Base URL | https://api.minimaxi.com/anthropic |
| LLM_MODEL | 模型名称 | MiniMax-M2.7 |
| LOCAL_LLM_MODEL | Ollama 模型 | qwen2.5-coder:7b |
| EMBED_BACKEND | local 或 cloud | local |
| LOCAL_EMBED_MODEL | 嵌入模型 | Qwen/Qwen3-Embedding-0.6B |
| VALIDATION_SERVICE_URL | 验证端点 | http://localhost:8001/validate |
| CHROMA_PERSIST_DIR | Chroma 存储位置 | ./db/chroma |
| MAX_RETRY | 自动修正尝试次数 | 3 |
| CONTEXT_MAX_TOKENS | 上下文压缩阈值 | 6000 |
| LOG_DIR | 日志目录 | ./logs |
| LOG_LEVEL | 日志级别 | DEBUG |
| SESSIONS_DIR | 会话持久化目录 | ./sessions |
+34 -1
View File
@@ -70,6 +70,36 @@
---
## 阶段四:可观测性
### 10. 结构化日志系统 ✓
- [x] `backend/logger.py` — 集中日志配置模块
- [x] JSON 格式化(每行一条记录,便于 jq/pandas 分析)
- [x] 请求级 trace_idcontextvars 自动传播,一次用户请求贯穿全链路)
- [x] 独立 LLM 日志文件 `logs/llm.log`(记录完整 prompt 和 response
- [x] 时区:UTC+8(中国时区)
- [x] 日志轮转(单文件 10MB,保留 5 备份)
- [x] `backend/llm.py``_LLMLoggingWrapper` 包装所有 LLM 后端
- [x] 记录每次 invoke/stream 的请求 prompt、响应内容、耗时、模型、调用来源
- [x] 异常时也记录完整 prompt
- [x] `agent/nodes.py``@log_node` 装饰器覆盖 17 个节点
- [x] 入口/出口/异常三个阶段的日志
- [x] 自动记录 state 关键字段摘要(session_id、intent、status、jrxml_length 等)
- [x] 每个节点耗时(duration_ms
- [x] `agent/graph.py``@_log_route` 装饰器覆盖 8 个路由函数
- [x] 记录每次路由决策(来源 → 目标)
- [x] `app.py` — 用户交互日志
- [x] 收到用户输入(含上传文件信息)
- [x] 代理执行开始/完成(含最终 intent、status、jrxml_length
- [x] 异常时记录错误详情
- [x] 会话新建/切换/删除操作日志
- [x] `backend/session.py` — 会话创建/删除日志
- [x] `backend/validation.py` — 验证完成/连接失败日志
- [x] `.env.example` — 新增 `LOG_DIR``LOG_LEVEL` 配置项
- [x] `.gitignore` — 新增 `logs/` 忽略规则
---
## 执行顺序建议
```
@@ -87,6 +117,9 @@
9. Ctrl+C 修复
10. 结构化日志系统
```
阶段一立即可做,无外部依赖。阶段二是主要工作量。阶段三是收尾。
阶段一立即可做,无外部依赖。阶段二是主要工作量。阶段三是收尾。阶段四是可观测性基础。
+35
View File
@@ -1,5 +1,6 @@
"""LangGraph JRXML 生成代理的状态图定义。"""
import functools
import os
from typing import Literal
@@ -25,14 +26,41 @@ from agent.nodes import (
correct_jrxml,
finalize,
)
from backend.logger import get_logger
load_dotenv()
MAX_RETRY = int(os.getenv("MAX_RETRY", "3"))
_graph_log = get_logger("agent")
def _log_route(route_name: str):
"""装饰器:自动记录路由决策。"""
def decorator(func):
@functools.wraps(func)
def wrapper(state: AgentState, *args, **kwargs):
target = func(state, *args, **kwargs)
_graph_log.info(
f"[路由] {route_name}{target}",
extra={
"route": route_name,
"target": target,
"session_id": state.get("session_id", ""),
"intent": state.get("intent", ""),
"status": state.get("status", ""),
"has_jrxml": bool(state.get("current_jrxml", "").strip()),
"retry_count": state.get("retry_count", 0),
},
)
return target
return wrapper
return decorator
# ============================================================
# 路由函数
# ============================================================
@_log_route("route_by_intent")
def route_by_intent(state: AgentState) -> Literal[
"retrieve", "modify_jrxml", "save_session",
"handle_consult", "handle_undo", "handle_reset"
@@ -59,18 +87,22 @@ def route_by_intent(state: AgentState) -> Literal[
return "retrieve"
@_log_route("route_after_generate")
def route_after_generate(state: AgentState) -> Literal["save_session"]:
return "save_session"
@_log_route("route_after_modify")
def route_after_modify(state: AgentState) -> Literal["save_session"]:
return "save_session"
@_log_route("route_after_undo")
def route_after_undo(state: AgentState) -> Literal["save_session"]:
return "save_session"
@_log_route("route_after_save")
def route_after_save(state: AgentState) -> Literal["validate", "finalize"]:
# 预览/导出意图跳过验证,直接完成
intent = state.get("intent", "")
@@ -79,16 +111,19 @@ def route_after_save(state: AgentState) -> Literal["validate", "finalize"]:
return "validate"
@_log_route("route_after_validate")
def route_after_validate(state: AgentState) -> Literal["finalize", "explain_error"]:
if state.get("status") == "pass":
return "finalize"
return "explain_error"
@_log_route("route_after_explain")
def route_after_explain(state: AgentState) -> Literal["correct_jrxml"]:
return "correct_jrxml"
@_log_route("route_after_correct")
def route_after_correct(state: AgentState) -> Literal["validate", "finalize"]:
retry = state.get("retry_count", 0)
if retry >= MAX_RETRY:
+86 -7
View File
@@ -1,9 +1,11 @@
"""LangGraph JRXML 生成工作流的节点函数。"""
import copy
import functools
import json
import os
import re
import time
from datetime import datetime, timezone
from typing import Dict
@@ -11,21 +13,82 @@ from dotenv import load_dotenv
from agent.state import AgentState
from backend.llm import get_llm
from backend.logger import get_logger, set_trace_id
from backend.validation import validate_jrxml
from prompts.loader import load_prompt
load_dotenv()
_node_log = get_logger("agent")
MAX_RETRY = int(os.getenv("MAX_RETRY", "3"))
CONTEXT_MAX_TOKENS = int(os.getenv("CONTEXT_MAX_TOKENS", "6000"))
CONTEXT_KEEP_RECENT = int(os.getenv("CONTEXT_KEEP_RECENT", "4"))
HISTORY_MAX_SNAPSHOTS = int(os.getenv("HISTORY_MAX_SNAPSHOTS", "10"))
def _state_summary(state: AgentState) -> dict:
"""提取 state 中的关键字段用于日志摘要。"""
user_input = state.get("user_input", "")
return {
"session_id": state.get("session_id", ""),
"intent": state.get("intent", ""),
"status": state.get("status", ""),
"has_jrxml": bool(state.get("current_jrxml", "").strip()),
"jrxml_length": len(state.get("current_jrxml", "")),
"retry_count": state.get("retry_count", 0),
"user_input_preview": user_input[:100] if user_input else "",
"conversation_turns": len(state.get("conversation_history", [])),
"history_snapshots": len(state.get("history_states", [])),
"versions": len(state.get("jrxml_versions", [])),
}
def log_node(node_name: str):
"""装饰器:自动记录节点入口、出口和耗时。"""
def decorator(func):
@functools.wraps(func)
def wrapper(state: AgentState, *args, **kwargs):
t0 = time.time()
_node_log.info(
f"[节点入口] {node_name}",
extra={"node": node_name, "phase": "entry", "state": _state_summary(state)},
)
try:
result = func(state, *args, **kwargs)
elapsed_ms = round((time.time() - t0) * 1000)
_node_log.info(
f"[节点出口] {node_name}",
extra={
"node": node_name,
"phase": "exit",
"duration_ms": elapsed_ms,
"state": _state_summary(state),
},
)
return result
except Exception as e:
elapsed_ms = round((time.time() - t0) * 1000)
_node_log.error(
f"[节点异常] {node_name}: {e}",
extra={
"node": node_name,
"phase": "error",
"duration_ms": elapsed_ms,
"error": str(e),
"state": _state_summary(state),
},
)
raise
return wrapper
return decorator
# ============================================================
# 核心工作流节点
# ============================================================
@log_node("process_input")
def process_input(state: AgentState) -> Dict:
"""记录用户输入到对话历史,重置本轮请求状态。如有上次失败上下文则自动注入。"""
user_input = state.get("user_input", "")
@@ -58,6 +121,7 @@ def process_input(state: AgentState) -> Dict:
return state
@log_node("save_state_snapshot")
def save_state_snapshot(state: AgentState) -> Dict:
"""保存当前状态快照到 history_states,用于撤销操作。最多保留 N 个版本。"""
snapshots = state.get("history_states", [])
@@ -82,6 +146,7 @@ def save_state_snapshot(state: AgentState) -> Dict:
return state
@log_node("classify_intent")
def classify_intent(state: AgentState) -> Dict:
"""使用 LLM 对用户输入进行意图分类(8 种意图)。"""
user_input = state.get("user_input", "")
@@ -89,7 +154,7 @@ def classify_intent(state: AgentState) -> Dict:
intent = "initial_generation"
try:
llm = get_llm()
llm = get_llm(caller="classify_intent")
prompt = load_prompt("intent_classify").format(
has_report=has_report,
user_input=user_input[:500],
@@ -116,11 +181,12 @@ def classify_intent(state: AgentState) -> Dict:
return state
@log_node("handle_consult")
def handle_consult(state: AgentState) -> Dict:
"""处理咨询类问题:调用 LLM 直接回答,不走报表生成流程。"""
user_input = state.get("user_input", "")
try:
llm = get_llm()
llm = get_llm(caller="handle_consult")
prompt = load_prompt("consult").format(question=user_input)
resp = llm.invoke(prompt)
answer = resp.content.strip()
@@ -135,6 +201,7 @@ def handle_consult(state: AgentState) -> Dict:
return state
@log_node("handle_undo")
def handle_undo(state: AgentState) -> Dict:
"""撤销上一步修改:从 history_states 恢复最近一个快照。"""
snapshots = state.get("history_states", [])
@@ -162,6 +229,7 @@ def handle_undo(state: AgentState) -> Dict:
return state
@log_node("handle_reset")
def handle_reset(state: AgentState) -> Dict:
"""重置当前会话:清空报表相关状态,保留会话信息。"""
state["current_jrxml"] = ""
@@ -186,6 +254,7 @@ def handle_reset(state: AgentState) -> Dict:
return state
@log_node("count_tokens")
def count_tokens(state: AgentState) -> int:
"""使用 tiktoken(gpt-4o 编码器)计算当前上下文 token 数量。"""
try:
@@ -208,6 +277,7 @@ def count_tokens(state: AgentState) -> int:
return len(enc.encode(text))
@log_node("manage_context")
def manage_context(state: AgentState) -> Dict:
"""当 token 数量超过阈值时,压缩较早的对话轮次。"""
token_count = count_tokens(state)
@@ -230,7 +300,7 @@ def manage_context(state: AgentState) -> Dict:
conv_text = json.dumps(older, ensure_ascii=False, indent=2)
try:
llm = get_llm()
llm = get_llm(caller="manage_context")
prompt = load_prompt("compression").format(conversation_text=conv_text)
resp = llm.invoke(prompt)
new_compressed = resp.content.strip()[:300]
@@ -249,6 +319,7 @@ def manage_context(state: AgentState) -> Dict:
return state
@log_node("load_session_node")
def load_session_node(state: AgentState) -> Dict:
"""在请求开始时从磁盘加载会话状态。"""
session_id = state.get("session_id", "")
@@ -273,6 +344,7 @@ def load_session_node(state: AgentState) -> Dict:
return state
@log_node("save_session_node")
def save_session_node(state: AgentState) -> Dict:
"""将当前代理状态持久化到磁盘。"""
session_id = state.get("session_id", "")
@@ -319,6 +391,7 @@ def _now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
@log_node("retrieve")
def retrieve(state: AgentState) -> Dict:
"""在 ChromaDB + 错误知识库中搜索相关的 JRXML 模板和组件。"""
try:
@@ -341,12 +414,13 @@ def retrieve(state: AgentState) -> Dict:
return state
@log_node("generate")
def generate(state: AgentState) -> Dict:
"""根据用户需求和检索到的上下文生成初始 JRXML。"""
from langgraph.config import get_stream_writer
writer = get_stream_writer()
llm = get_llm()
llm = get_llm(caller="generate")
prompt = load_prompt("initial_generation").format(
context=state.get("retrieved_context", ""),
user_request=state.get("user_input", ""),
@@ -361,12 +435,13 @@ def generate(state: AgentState) -> Dict:
return state
@log_node("modify_jrxml")
def modify_jrxml(state: AgentState) -> Dict:
"""根据用户的修改请求修改现有 JRXML。"""
from langgraph.config import get_stream_writer
writer = get_stream_writer()
llm = get_llm()
llm = get_llm(caller="modify_jrxml")
# 构建对话上下文:压缩摘要 + 最近对话
compressed = state.get("compressed_history", "")
recent = state.get("conversation_history", [])[-6:]
@@ -405,6 +480,7 @@ def modify_jrxml(state: AgentState) -> Dict:
return state
@log_node("validate")
def validate(state: AgentState) -> Dict:
"""根据 FastAPI 验证服务验证当前 JRXML。"""
jrxml = state.get("current_jrxml", "")
@@ -448,9 +524,10 @@ def validate(state: AgentState) -> Dict:
return state
@log_node("explain_error")
def explain_error(state: AgentState) -> Dict:
"""生成验证错误的可读解释。"""
llm = get_llm()
llm = get_llm(caller="explain_error")
jrxml = state.get("current_jrxml", "")
lines = jrxml.split("\n")[:80]
snippet = "\n".join(lines)
@@ -464,12 +541,13 @@ def explain_error(state: AgentState) -> Dict:
return state
@log_node("correct_jrxml")
def correct_jrxml(state: AgentState) -> Dict:
"""尝试自动修正验证失败的 JRXML。"""
from langgraph.config import get_stream_writer
writer = get_stream_writer()
llm = get_llm()
llm = get_llm(caller="correct_jrxml")
prompt = load_prompt("correction").format(
current_jrxml=state.get("current_jrxml", ""),
error_msg=state.get("error_msg", ""),
@@ -495,6 +573,7 @@ def correct_jrxml(state: AgentState) -> Dict:
return state
@log_node("finalize")
def finalize(state: AgentState) -> Dict:
"""保存最终验证通过的 JRXML 并更新对话历史 + 版本记录。"""
jrxml = state.get("current_jrxml", "")
+69 -2
View File
@@ -9,6 +9,15 @@
import os
import sys
os.environ.setdefault("TRANSFORMERS_VERBOSITY", "error")
try:
import torchvision
except Exception:
pass
import time
from pathlib import Path
import streamlit as st
@@ -23,6 +32,9 @@ from backend.session import (
delete_session,
list_all_sessions,
)
from backend.logger import get_logger, set_trace_id, generate_trace_id
_app_log = get_logger("app")
st.set_page_config(
page_title="JRXML 代理",
@@ -32,7 +44,7 @@ st.set_page_config(
)
# 阻止 Streamlit 裸 'c' 键清除缓存,保留 Ctrl+C 复制行为
st.components.v1.html("""
st.html("""
<script>
(function() {
const parent = window.parent.document;
@@ -48,7 +60,7 @@ st.components.v1.html("""
}, true);
})();
</script>
""", height=0)
""")
# ---- 节点名称 → 中文标签 ----
NODE_LABELS = {
@@ -130,7 +142,22 @@ current_session_id = st.session_state.agent_state.get("session_id", "")
def run_agent(user_input: str):
"""运行代理图:流式渲染节点进度 + LLM 文本。"""
trace_id = generate_trace_id()
set_trace_id(trace_id)
agent_state = st.session_state.agent_state
session_id = agent_state.get("session_id", "")
_app_log.info(
"代理执行开始",
extra={
"session_id": session_id,
"trace_id": trace_id,
"user_input_preview": user_input[:200],
"user_input_length": len(user_input),
"has_jrxml": bool(agent_state.get("current_jrxml", "").strip()),
"intent": agent_state.get("intent", ""),
},
)
if agent_state.get("current_jrxml") and agent_state.get("status") == "pass":
agent_state["user_modification_request"] = user_input
@@ -222,6 +249,10 @@ def run_agent(user_input: str):
except Exception as e:
progress_placeholder.empty()
_app_log.error(
f"代理执行异常: {e}",
extra={"session_id": session_id, "error": str(e)},
)
st.error(f"工作流异常: {e}")
return
@@ -296,6 +327,17 @@ def run_agent(user_input: str):
else:
st.error("未产生结果,请重试。")
_app_log.info(
"代理执行完成",
extra={
"session_id": session_id,
"intent": final_state.get("intent", ""),
"status": final_state.get("status", ""),
"jrxml_length": len(final_state.get("current_jrxml", "")),
"retry_count": final_state.get("retry_count", 0),
},
)
# ---- 侧边栏 ----
with st.sidebar:
@@ -330,6 +372,10 @@ with st.sidebar:
new_sid = session_options[selected]
data = load_session(new_sid)
if data and data.get("agent_state"):
_app_log.info(
"切换会话",
extra={"from_session": current_session_id, "to_session": new_sid},
)
st.session_state.agent_state = data["agent_state"]
st.session_state.messages = []
st.rerun()
@@ -338,6 +384,10 @@ with st.sidebar:
with col1:
if st.button(" 新建", use_container_width=True):
new_data = create_session(name="", agent_state=create_initial_state())
_app_log.info(
"新建会话",
extra={"session_id": new_data["session_id"]},
)
st.session_state.agent_state = create_initial_state()
st.session_state.agent_state["session_id"] = new_data["session_id"]
st.session_state.agent_state["session_name"] = new_data["session_name"]
@@ -347,6 +397,10 @@ with st.sidebar:
with col2:
if st.button("🗑 删除", use_container_width=True):
if current_session_id:
_app_log.info(
"删除会话",
extra={"session_id": current_session_id},
)
delete_session(current_session_id)
st.session_state.agent_state = create_initial_state()
new_data = create_session(name="", agent_state=st.session_state.agent_state)
@@ -541,15 +595,28 @@ for msg in st.session_state.messages:
if prompt := st.chat_input("描述您的报表需求..."):
# 拼接上传文件的文本
uploaded_texts = []
uploaded_files_info = []
if st.session_state.get("uploaded_files"):
for f in st.session_state.uploaded_files:
uploaded_texts.append(f"[上传文件: {f['name']}]\n{f['text']}")
uploaded_files_info.append({"name": f["name"], "type": f["type"], "length": len(f["text"])})
if uploaded_texts:
full_prompt = "\n\n".join(uploaded_texts) + "\n\n---\n用户需求:\n" + prompt
st.session_state.uploaded_files = [] # 用后即清
else:
full_prompt = prompt
_app_log.info(
"收到用户输入",
extra={
"session_id": current_session_id,
"prompt_preview": prompt[:200],
"prompt_length": len(prompt),
"has_uploaded_files": bool(uploaded_files_info),
"uploaded_files": uploaded_files_info,
},
)
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.markdown(prompt)
+138 -11
View File
@@ -1,12 +1,17 @@
"""大语言模型工厂:支持 OpenAI 兼容的云端 API、Anthropic 兼容 API 和本地 Ollama。"""
import os
import time
from typing import Any
from dotenv import load_dotenv
from backend.logger import get_logger
load_dotenv()
_llm_log = get_logger("llm")
class _BaseLLM:
"""LLM 统一接口基类 — 所有后端都提供 invoke() 和 stream()。"""
@@ -18,7 +23,124 @@ class _BaseLLM:
raise NotImplementedError
def get_llm():
class _LLMLoggingWrapper(_BaseLLM):
"""包装任何 LLM 后端,自动记录输入/输出到 llm.log。"""
def __init__(self, inner: _BaseLLM, model: str, backend: str, caller: str = ""):
self._inner = inner
self._model = model
self._backend = backend
self._caller = caller
def invoke(self, prompt: str) -> Any:
t0 = time.time()
prompt_len = len(prompt)
prompt_preview = prompt[:500]
_llm_log.debug(
"LLM invoke 请求",
extra={
"direction": "request",
"model": self._model,
"backend": self._backend,
"caller": self._caller,
"prompt_length": prompt_len,
"prompt_preview": prompt_preview,
"prompt": prompt[:10000],
},
)
try:
result = self._inner.invoke(prompt)
elapsed = round((time.time() - t0) * 1000)
content = getattr(result, "content", str(result))
resp_len = len(content)
resp_preview = content[:500]
_llm_log.info(
"LLM invoke 完成",
extra={
"direction": "response",
"model": self._model,
"backend": self._backend,
"caller": self._caller,
"duration_ms": elapsed,
"response_length": resp_len,
"response_preview": resp_preview,
"response": content[:10000],
},
)
return result
except Exception as e:
elapsed = round((time.time() - t0) * 1000)
_llm_log.error(
"LLM invoke 异常",
extra={
"direction": "error",
"model": self._model,
"backend": self._backend,
"caller": self._caller,
"duration_ms": elapsed,
"error": str(e),
"prompt": prompt[:10000],
},
)
raise
def stream(self, prompt: str):
t0 = time.time()
prompt_len = len(prompt)
prompt_preview = prompt[:500]
_llm_log.debug(
"LLM stream 请求",
extra={
"direction": "request",
"model": self._model,
"backend": self._backend,
"caller": self._caller,
"prompt_length": prompt_len,
"prompt_preview": prompt_preview,
"prompt": prompt[:10000],
},
)
full = []
try:
for chunk in self._inner.stream(prompt):
full.append(chunk)
yield chunk
elapsed = round((time.time() - t0) * 1000)
resp_text = "".join(full)
resp_len = len(resp_text)
resp_preview = resp_text[:500]
_llm_log.info(
"LLM stream 完成",
extra={
"direction": "response",
"model": self._model,
"backend": self._backend,
"caller": self._caller,
"duration_ms": elapsed,
"response_length": resp_len,
"response_preview": resp_preview,
"response": resp_text[:10000],
},
)
except Exception as e:
elapsed = round((time.time() - t0) * 1000)
_llm_log.error(
"LLM stream 异常",
extra={
"direction": "error",
"model": self._model,
"backend": self._backend,
"caller": self._caller,
"duration_ms": elapsed,
"error": str(e),
"prompt": prompt[:10000],
},
)
raise
def _build_raw_llm(caller: str = "") -> tuple[_BaseLLM, str, str]:
"""构造原始 LLM 实例,返回 (实例, model名, backend名)。"""
backend = os.getenv("LLM_BACKEND", "cloud")
if backend == "local":
from langchain_ollama import ChatOllama
@@ -34,20 +156,18 @@ def get_llm():
for chunk in raw.stream(prompt):
yield chunk.content
return OllamaWrapper()
return OllamaWrapper(), model, f"local/{model}"
provider = os.getenv("LLM_PROVIDER", "openai")
if provider == "anthropic":
from anthropic import Anthropic
api_key = os.getenv("OPENAI_API_KEY", "")
base_url = os.getenv("OPENAI_BASE_URL", "https://api.minimaxi.com/anthropic")
model = os.getenv("LLM_MODEL", "minimax-2.7")
api_key = os.getenv("ANTHROPIC_API_KEY") or os.getenv("OPENAI_API_KEY", "")
base_url = os.getenv("ANTHROPIC_BASE_URL") or os.getenv("OPENAI_BASE_URL", "https://api.minimaxi.com/anthropic")
model = os.getenv("LLM_MODEL", "MiniMax-M2.7")
temperature = 0.1
max_tokens = 4096
os.environ["NO_PROXY"] = "*"
client = Anthropic(api_key=api_key, base_url=base_url, timeout=120)
class MiniMaxLLM(_BaseLLM):
@@ -80,12 +200,13 @@ def get_llm():
)
return resp.input_tokens
return MiniMaxLLM()
return MiniMaxLLM(), model, f"cloud/anthropic/{model}"
else:
from langchain_openai import ChatOpenAI
model = os.getenv("LLM_MODEL", "gpt-4o")
raw = ChatOpenAI(
model=os.getenv("LLM_MODEL", "gpt-4o"),
model=model,
api_key=os.getenv("OPENAI_API_KEY"),
base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
temperature=0.1,
@@ -99,8 +220,14 @@ def get_llm():
for chunk in raw.stream(prompt):
yield chunk.content
return OpenAIWrapper()
return OpenAIWrapper(), model, f"cloud/openai/{model}"
def get_llm(caller: str = "") -> _BaseLLM:
"""返回带日志的 LLM 实例。caller 用于标识调用来源(如 generate、classify_intent)。"""
inner, model, backend = _build_raw_llm(caller)
return _LLMLoggingWrapper(inner, model=model, backend=backend, caller=caller)
def get_llm_for_correction():
return get_llm()
return get_llm(caller="correction")
+167
View File
@@ -0,0 +1,167 @@
"""集中日志模块。
提供:
- 结构化 JSON 日志(每行一条记录)
- 请求级 trace_id(通过 contextvars 自动传播)
- 独立的 LLM 调用日志文件
- 日志轮转(按大小 10MB,保留 5 个备份)
用法:
from backend.logger import get_logger, set_trace_id
# 业务日志
log = get_logger("agent")
log.info("节点开始执行", extra={"node": "classify_intent", "session_id": "xxx"})
# LLM 日志
llm_log = get_logger("llm")
llm_log.info("LLM 请求", extra={"prompt": "...", "model": "gpt-4o"})
"""
import json
import logging
import os
import sys
import uuid
from contextvars import ContextVar
from datetime import datetime, timezone, timedelta
from logging.handlers import RotatingFileHandler
from pathlib import Path
from typing import Optional
LOG_DIR = Path(os.getenv("LOG_DIR", "./logs"))
LOG_LEVEL = os.getenv("LOG_LEVEL", "DEBUG")
LLM_LOG_FILE = "llm.log"
APP_LOG_FILE = "app.log"
CHINA_TZ = timezone(timedelta(hours=8))
_trace_id_var: ContextVar[str] = ContextVar("trace_id", default="")
def generate_trace_id() -> str:
return uuid.uuid4().hex[:16]
def get_trace_id() -> str:
tid = _trace_id_var.get()
if not tid:
tid = generate_trace_id()
_trace_id_var.set(tid)
return tid
def set_trace_id(trace_id: str):
_trace_id_var.set(trace_id)
class JsonFormatter(logging.Formatter):
"""将日志记录格式化为单行 JSON,便于后续分析。
LogRecord 标准属性的键(不放入 extra)。
通过 logging.Logger.debug(msg, extra={...}) 传入的键会自动设为
LogRecord 属性,由本格式化器收集到 extra 字段中。
"""
_STANDARD_ATTRS: set[str] = frozenset({
"args", "asctime", "created", "exc_info", "exc_text", "filename",
"funcName", "levelname", "levelno", "lineno", "module", "msecs",
"message", "msg", "name", "pathname", "process", "processName",
"relativeCreated", "stack_info", "thread", "threadName",
"extra_fields", "taskName",
})
def _collect_extra(self, record: logging.LogRecord) -> dict:
"""从 LogRecord 上收集非标准属性 → 合并为 extra dict。"""
extra = dict(getattr(record, "extra_fields", {}))
for key, val in record.__dict__.items():
if key not in self._STANDARD_ATTRS and not key.startswith("_"):
extra[key] = val
return extra
def format(self, record: logging.LogRecord) -> str:
log_entry = {
"timestamp": datetime.now(CHINA_TZ).isoformat(),
"level": record.levelname,
"logger": record.name,
"trace_id": get_trace_id(),
"message": record.getMessage(),
"module": record.module,
"function": record.funcName,
"line": record.lineno,
}
extra = self._collect_extra(record)
if extra:
log_entry["extra"] = extra
if record.exc_info and record.exc_info[0]:
import traceback
log_entry["exception"] = traceback.format_exception(
record.exc_info[0], record.exc_info[1], record.exc_info[2]
)
return json.dumps(log_entry, ensure_ascii=False)
def _create_handler(filename: str, level: int) -> RotatingFileHandler:
handler = RotatingFileHandler(
filename=str(LOG_DIR / filename),
maxBytes=10 * 1024 * 1024,
backupCount=5,
encoding="utf-8",
)
handler.setLevel(level)
handler.setFormatter(JsonFormatter())
return handler
def _get_level() -> int:
return getattr(logging, LOG_LEVEL.upper(), logging.DEBUG)
def get_logger(name: str) -> logging.Logger:
"""获取指定名称的 logger,自动配置了 JSON 格式化 + 文件轮转。
name="llm" → 输出到 logs/llm.log(仅 LLM 调用相关)
其他 name → 输出到 logs/app.log
"""
logger = logging.getLogger(f"jrxml.{name}")
if logger.handlers:
return logger
LOG_DIR.mkdir(parents=True, exist_ok=True)
level = _get_level()
logger.setLevel(level)
logger.propagate = False
if name == "llm":
logger.addHandler(_create_handler(LLM_LOG_FILE, level))
else:
logger.addHandler(_create_handler(APP_LOG_FILE, level))
return logger
class _ExtraAdapter(logging.LoggerAdapter):
"""支持通过 adapter.extra 合并 extra 字段的适配器。"""
def process(self, msg, kwargs):
extra = kwargs.pop("extra", {})
merged = {**self.extra, **extra} if self.extra or extra else None
if merged:
kwargs["extra"] = {"extra_fields": merged}
return msg, kwargs
def get_trace_logger(name: str) -> _ExtraAdapter:
"""返回一个自动附带 trace_id 的 logger 适配器。
用法:
log = get_trace_logger("agent")
log.info("节点完成", extra={"node": "generate"})
"""
logger = get_logger(name)
return _ExtraAdapter(logger, {"trace_id": get_trace_id()})
+6
View File
@@ -12,8 +12,12 @@ from typing import Optional
from dotenv import load_dotenv
from backend.logger import get_logger
load_dotenv()
_session_log = get_logger("session")
SESSIONS_DIR = Path(os.getenv("SESSIONS_DIR", "./sessions"))
@@ -43,6 +47,7 @@ def create_session(name: str = "", agent_state: Optional[dict] = None) -> dict:
}
with open(_session_path(sid), "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
_session_log.info("创建会话", extra={"session_id": sid, "session_name": data["session_name"]})
return data
@@ -104,6 +109,7 @@ def delete_session(session_id: str) -> bool:
fp = _session_path(session_id)
if fp.exists():
fp.unlink()
_session_log.info("删除会话", extra={"session_id": session_id})
return True
return False
+21 -6
View File
@@ -5,22 +5,37 @@ import os
import httpx
from dotenv import load_dotenv
from backend.logger import get_logger
load_dotenv()
_val_log = get_logger("validation")
VALIDATION_URL = os.getenv("VALIDATION_SERVICE_URL", "http://localhost:8001/validate")
def validate_jrxml(jrxml_text: str) -> dict:
"""将 JRXML 发送到验证服务并返回 {valid: bool, error: str}。"""
jrxml_length = len(jrxml_text)
try:
with httpx.Client(timeout=30.0) as client:
resp = client.post(VALIDATION_URL, json={"jrxml": jrxml_text})
resp.raise_for_status()
return resp.json()
result = resp.json()
_val_log.info(
"验证完成",
extra={
"valid": result.get("valid"),
"error": result.get("error", ""),
"jrxml_length": jrxml_length,
},
)
return result
except httpx.ConnectError:
return {
"valid": False,
"error": f"无法连接到验证服务 ({VALIDATION_URL})。是否正在运行?",
}
error_msg = f"无法连接到验证服务 ({VALIDATION_URL})。是否正在运行?"
_val_log.error("验证服务连接失败", extra={"error": error_msg, "url": VALIDATION_URL})
return {"valid": False, "error": error_msg}
except Exception as e:
return {"valid": False, "error": f"验证请求失败: {str(e)}"}
error_msg = f"验证请求失败: {str(e)}"
_val_log.error("验证请求异常", extra={"error": str(e), "url": VALIDATION_URL})
return {"valid": False, "error": error_msg}