Evolve paper browser into knowledge atlas
This commit is contained in:
wuyang
+435
-1
@@ -5,7 +5,6 @@ from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import html
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
@@ -15,6 +14,8 @@ import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
import xml.etree.ElementTree as ET
|
||||
from collections import Counter, defaultdict
|
||||
from itertools import combinations
|
||||
from http import HTTPStatus
|
||||
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
@@ -34,6 +35,10 @@ DEFAULT_MODELS = {
|
||||
"translate": "ChatGPT-5.6:light",
|
||||
"summary": "ChatGPT-5.6:fast",
|
||||
"deep": "ChatGPT-5.6:large",
|
||||
"atlas": "ChatGPT-5.6:fast",
|
||||
"topic": "ChatGPT-5.6:large",
|
||||
"path": "ChatGPT-5.6:large",
|
||||
"compare": "ChatGPT-5.6:large",
|
||||
}
|
||||
|
||||
NS = {
|
||||
@@ -44,6 +49,124 @@ NS = {
|
||||
INDEX_LOCK = threading.Lock()
|
||||
OLLAMA_LOCK = threading.Lock()
|
||||
INDEX_CACHE: dict[str, Any] = {"mtime": 0.0, "papers": [], "by_id": {}}
|
||||
ATLAS_CACHE: dict[str, Any] = {"mtime": 0.0, "atlas": None}
|
||||
|
||||
TOPIC_PROFILES = {
|
||||
"agent-evaluation": {
|
||||
"label": "Evaluation",
|
||||
"stance": "评估已经成为 Agent 研究的中心层：不只是看最终答案，而是看轨迹、工具调用、失败点、成本和安全边界。",
|
||||
"question": "如何证明一个 Agent 真的能在长程、不确定、带工具的环境里可靠完成任务？",
|
||||
},
|
||||
"tool-use": {
|
||||
"label": "Tool Use",
|
||||
"stance": "工具调用正在从 prompt 技巧变成运行时系统问题：schema、权限、依赖、失败恢复和可观测性都变成核心。",
|
||||
"question": "模型什么时候应该调用工具，如何调用，失败后如何恢复，工具链如何被治理？",
|
||||
},
|
||||
"memory": {
|
||||
"label": "Memory",
|
||||
"stance": "Memory 不再只是 RAG 的附属，而是长程 Agent 的状态管理、经验沉淀和安全攻击面。",
|
||||
"question": "Agent 应该记什么、忘什么、如何写入、如何检索、如何防止记忆污染？",
|
||||
},
|
||||
"rag": {
|
||||
"label": "Agentic RAG",
|
||||
"stance": "RAG 正在从相似度检索走向主动检索、证据规划、多跳推理和工具化知识访问。",
|
||||
"question": "Agent 如何把检索变成可验证、可迭代、可追踪的推理过程？",
|
||||
},
|
||||
"planning": {
|
||||
"label": "Planning",
|
||||
"stance": "Planning 的重点从一次性计划转向执行中的重规划、状态校正和长程任务稳定性。",
|
||||
"question": "长程任务里，计划如何与真实环境反馈、工具失败和上下文限制协同？",
|
||||
},
|
||||
"agent-safety": {
|
||||
"label": "Safety",
|
||||
"stance": "Agent safety 已经从拒答问题转向运行时风险：权限、注入、记忆污染、越权动作和治理平面。",
|
||||
"question": "如何限制 Agent 的行动边界，同时保留足够的自主性和生产力？",
|
||||
},
|
||||
"coding-agent": {
|
||||
"label": "Coding Agent",
|
||||
"stance": "Coding Agent 正在从单题修 bug 走向仓库级、长程、交互式的软件工程流程。",
|
||||
"question": "如何评估和构建能长期维护真实仓库的编码 Agent？",
|
||||
},
|
||||
"computer-use": {
|
||||
"label": "Computer Use",
|
||||
"stance": "GUI/browser/OS Agent 正在成为独立层：视觉状态、动作边界、任务进度和安全确认都需要重新建模。",
|
||||
"question": "Agent 如何可靠理解屏幕状态，并在真实界面中安全行动？",
|
||||
},
|
||||
"multi-agent": {
|
||||
"label": "Multi-Agent",
|
||||
"stance": "Multi-agent 正在从角色扮演走向共享记忆、协作协议、路由、治理和组织行为。",
|
||||
"question": "多个 Agent 如何共享状态、分工、纠错，并避免多数投票和上下文碎片带来的失败？",
|
||||
},
|
||||
"workflow-agent": {
|
||||
"label": "Workflow",
|
||||
"stance": "Workflow Agent 把模型能力接进企业流程，核心不在聊天，而在状态、权限、审计和可恢复执行。",
|
||||
"question": "怎样把 Agent 放进真实业务流程，并让它可观察、可治理、可回滚？",
|
||||
},
|
||||
"reasoning": {
|
||||
"label": "Reasoning",
|
||||
"stance": "Reasoning 在 Agent 中必须和工具、记忆、验证器、环境反馈一起看，而不是孤立的思维链。",
|
||||
"question": "推理如何真正改善行动，而不是只让轨迹看起来更复杂？",
|
||||
},
|
||||
"world-model": {
|
||||
"label": "World Model",
|
||||
"stance": "World model 让 Agent 从反应式调用走向对环境、规则和未来状态的内部建模。",
|
||||
"question": "Agent 需要什么样的世界模型才能更好地预测行动后果？",
|
||||
},
|
||||
"embodied-agent": {
|
||||
"label": "Embodied",
|
||||
"stance": "Embodied Agent 把语言、视觉、规划和物理约束连接起来，是 Agent 能力外延的重要方向。",
|
||||
"question": "语言模型如何在有物理约束的环境中做可执行决策？",
|
||||
},
|
||||
}
|
||||
|
||||
TOPIC_POSITIONS = {
|
||||
"agent-evaluation": (51, 37),
|
||||
"tool-use": (37, 52),
|
||||
"memory": (52, 61),
|
||||
"rag": (68, 57),
|
||||
"planning": (48, 48),
|
||||
"agent-safety": (62, 35),
|
||||
"coding-agent": (27, 31),
|
||||
"computer-use": (29, 68),
|
||||
"multi-agent": (74, 42),
|
||||
"workflow-agent": (62, 74),
|
||||
"reasoning": (42, 24),
|
||||
"world-model": (78, 68),
|
||||
"embodied-agent": (82, 82),
|
||||
}
|
||||
|
||||
KNOWLEDGE_ROUTES = [
|
||||
{
|
||||
"id": "evaluation-runtime",
|
||||
"title": "从答案评测到运行时评估",
|
||||
"topics": ["agent-evaluation", "tool-use", "agent-safety", "workflow-agent"],
|
||||
"summary": "Agent 的评估对象从 final answer 扩展到轨迹、工具调用、权限、失败恢复和上线治理。",
|
||||
},
|
||||
{
|
||||
"id": "memory-state",
|
||||
"title": "从 RAG 到状态管理",
|
||||
"topics": ["rag", "memory", "planning", "agent-safety"],
|
||||
"summary": "长期记忆、主动检索、写入策略和记忆安全正在把 RAG 推成 Agent 的状态层。",
|
||||
},
|
||||
{
|
||||
"id": "coding-long-horizon",
|
||||
"title": "从 SWE-Bench 到长程软件工程",
|
||||
"topics": ["coding-agent", "agent-evaluation", "memory", "tool-use"],
|
||||
"summary": "编码 Agent 的问题从单 bug 修复扩展到仓库理解、长程维护、交互澄清和 CI 反馈循环。",
|
||||
},
|
||||
{
|
||||
"id": "computer-use",
|
||||
"title": "从工具 API 到真实电脑操作",
|
||||
"topics": ["computer-use", "tool-use", "planning", "agent-safety"],
|
||||
"summary": "GUI、浏览器和 OS Agent 让状态识别、动作边界、确认机制和视觉记忆成为新的系统层。",
|
||||
},
|
||||
{
|
||||
"id": "collective-agents",
|
||||
"title": "从单体智能到协作系统",
|
||||
"topics": ["multi-agent", "memory", "agent-evaluation", "workflow-agent"],
|
||||
"summary": "多 Agent 研究正在走向共享记忆、角色分工、路由、群体失败归因和组织行为评估。",
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
def ensure_cache_dirs() -> None:
|
||||
@@ -177,6 +300,249 @@ def topic_counts(papers: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
]
|
||||
|
||||
|
||||
def month_key(value: str) -> str:
|
||||
if re.match(r"^\d{4}-\d{2}", value or ""):
|
||||
return value[:7]
|
||||
return "unknown"
|
||||
|
||||
|
||||
def profile_for(topic: str) -> dict[str, str]:
|
||||
fallback = {
|
||||
"label": topic.replace("-", " ").title(),
|
||||
"stance": f"{topic} 是当前语料中自动识别出的主题，需要进一步人工导读。",
|
||||
"question": "这个主题与 Agent 系统能力的关系是什么？",
|
||||
}
|
||||
return {**fallback, **TOPIC_PROFILES.get(topic, {})}
|
||||
|
||||
|
||||
def top_papers_for(papers: list[dict[str, Any]], limit: int = 8, recent: bool = False) -> list[dict[str, Any]]:
|
||||
ranked = list(papers)
|
||||
if recent:
|
||||
ranked.sort(key=lambda item: (item["published_at"], item["collection_score"]), reverse=True)
|
||||
else:
|
||||
ranked.sort(key=lambda item: (item["collection_score"], item["published_at"]), reverse=True)
|
||||
return [paper_public(paper) for paper in ranked[:limit]]
|
||||
|
||||
|
||||
def build_atlas() -> dict[str, Any]:
|
||||
papers, _ = load_papers()
|
||||
mtime = INDEX_PATH.stat().st_mtime
|
||||
if ATLAS_CACHE["mtime"] == mtime and ATLAS_CACHE["atlas"]:
|
||||
return ATLAS_CACHE["atlas"]
|
||||
|
||||
topic_to_papers: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
||||
topic_months: dict[str, Counter] = defaultdict(Counter)
|
||||
edge_counts: Counter = Counter()
|
||||
month_counts: Counter = Counter()
|
||||
|
||||
for paper in papers:
|
||||
month = month_key(paper["published_at"])
|
||||
month_counts[month] += 1
|
||||
topics = sorted(set(paper["topics"]))
|
||||
for topic in topics:
|
||||
topic_to_papers[topic].append(paper)
|
||||
topic_months[topic][month] += 1
|
||||
for left, right in combinations(topics, 2):
|
||||
edge_counts[(left, right)] += 1
|
||||
|
||||
months = [month for month in sorted(month_counts) if month != "unknown"]
|
||||
recent_months = set(months[-3:])
|
||||
max_count = max((len(items) for items in topic_to_papers.values()), default=1)
|
||||
max_recent = max(
|
||||
(sum(topic_months[topic][month] for month in recent_months) for topic in topic_to_papers),
|
||||
default=1,
|
||||
)
|
||||
|
||||
topics = []
|
||||
fallback_index = 0
|
||||
for topic, topic_papers in sorted(topic_to_papers.items(), key=lambda item: (-len(item[1]), item[0])):
|
||||
profile = profile_for(topic)
|
||||
recent_count = sum(topic_months[topic][month] for month in recent_months)
|
||||
if topic in TOPIC_POSITIONS:
|
||||
x, y = TOPIC_POSITIONS[topic]
|
||||
else:
|
||||
x = 18 + (fallback_index % 5) * 16
|
||||
y = 16 + (fallback_index // 5) * 14
|
||||
fallback_index += 1
|
||||
topics.append(
|
||||
{
|
||||
"id": topic,
|
||||
"label": profile["label"],
|
||||
"count": len(topic_papers),
|
||||
"recent_count": recent_count,
|
||||
"share": round(len(topic_papers) / max(len(papers), 1), 3),
|
||||
"heat": round(recent_count / max_recent, 3),
|
||||
"radius": round(16 + 34 * (len(topic_papers) / max_count), 1),
|
||||
"x": x,
|
||||
"y": y,
|
||||
"stance": profile["stance"],
|
||||
"question": profile["question"],
|
||||
"key_papers": top_papers_for(topic_papers, 5),
|
||||
}
|
||||
)
|
||||
|
||||
top_topic_ids = {topic["id"] for topic in topics[:14]}
|
||||
max_edge_count = max(edge_counts.values(), default=1)
|
||||
edges = [
|
||||
{
|
||||
"source": left,
|
||||
"target": right,
|
||||
"count": count,
|
||||
"strength": round(count / max_edge_count, 3),
|
||||
}
|
||||
for (left, right), count in edge_counts.most_common(80)
|
||||
if left in top_topic_ids and right in top_topic_ids
|
||||
]
|
||||
|
||||
timeline = []
|
||||
for month in months:
|
||||
row = {"month": month, "total": month_counts[month], "topics": {}}
|
||||
for topic in top_topic_ids:
|
||||
row["topics"][topic] = topic_months[topic][month]
|
||||
timeline.append(row)
|
||||
|
||||
atlas = {
|
||||
"generated_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
|
||||
"paper_count": len(papers),
|
||||
"topic_count": len(topics),
|
||||
"months": months,
|
||||
"recent_months": sorted(recent_months),
|
||||
"topics": topics,
|
||||
"edges": edges,
|
||||
"timeline": timeline,
|
||||
"routes": KNOWLEDGE_ROUTES,
|
||||
"key_papers": top_papers_for(papers, 12),
|
||||
"recent_papers": top_papers_for(papers, 12, recent=True),
|
||||
"design_policy": {
|
||||
"precomputed": [
|
||||
"主题地图",
|
||||
"主题规模",
|
||||
"最近热度",
|
||||
"主题共现关系",
|
||||
"月度趋势",
|
||||
"关键论文候选",
|
||||
"主题证据列表",
|
||||
],
|
||||
"ai_on_demand": [
|
||||
"解释当前地图",
|
||||
"生成主题导读",
|
||||
"生成阅读路径",
|
||||
"比较技术路线",
|
||||
"单篇论文研读",
|
||||
],
|
||||
},
|
||||
}
|
||||
ATLAS_CACHE.update({"mtime": mtime, "atlas": atlas})
|
||||
return atlas
|
||||
|
||||
|
||||
def topic_focus(topic: str) -> dict[str, Any]:
|
||||
atlas = build_atlas()
|
||||
papers, _ = load_papers()
|
||||
topic_papers = [paper for paper in papers if topic in paper["topics"]]
|
||||
if not topic_papers:
|
||||
raise KeyError(topic)
|
||||
|
||||
months = {row["month"]: row["topics"].get(topic, 0) for row in atlas["timeline"]}
|
||||
neighbor_counts: Counter = Counter()
|
||||
for paper in topic_papers:
|
||||
for other in paper["topics"]:
|
||||
if other != topic:
|
||||
neighbor_counts[other] += 1
|
||||
|
||||
profile = profile_for(topic)
|
||||
routes = [route for route in KNOWLEDGE_ROUTES if topic in route["topics"]]
|
||||
focus = {
|
||||
"topic": topic,
|
||||
"label": profile["label"],
|
||||
"stance": profile["stance"],
|
||||
"question": profile["question"],
|
||||
"count": len(topic_papers),
|
||||
"months": months,
|
||||
"neighbors": [
|
||||
{
|
||||
"topic": name,
|
||||
"label": profile_for(name)["label"],
|
||||
"count": count,
|
||||
"stance": profile_for(name)["stance"],
|
||||
}
|
||||
for name, count in neighbor_counts.most_common(10)
|
||||
],
|
||||
"routes": routes,
|
||||
"key_papers": top_papers_for(topic_papers, 12),
|
||||
"recent_papers": top_papers_for(topic_papers, 12, recent=True),
|
||||
"evidence": top_papers_for(topic_papers, 40),
|
||||
"reading_lens": [
|
||||
"先看主题问题，再看高分论文，不要从论文列表开始迷路。",
|
||||
"优先对比相邻主题，它们通常代表真实系统里的交叉问题。",
|
||||
"把论文当作证据：它支持了哪条技术路线，暴露了哪个未解决问题？",
|
||||
],
|
||||
}
|
||||
return focus
|
||||
|
||||
|
||||
def atlas_context(topic: str = "", topic_b: str = "") -> str:
|
||||
atlas = build_atlas()
|
||||
compact = {
|
||||
"paper_count": atlas["paper_count"],
|
||||
"recent_months": atlas["recent_months"],
|
||||
"top_topics": [
|
||||
{
|
||||
"topic": item["id"],
|
||||
"label": item["label"],
|
||||
"count": item["count"],
|
||||
"recent_count": item["recent_count"],
|
||||
"stance": item["stance"],
|
||||
}
|
||||
for item in atlas["topics"][:13]
|
||||
],
|
||||
"strong_edges": atlas["edges"][:24],
|
||||
"routes": atlas["routes"],
|
||||
"key_papers": [
|
||||
{
|
||||
"title": paper["title"],
|
||||
"topics": paper["topics"],
|
||||
"score": paper["collection_score"],
|
||||
"published_at": paper["published_at"],
|
||||
}
|
||||
for paper in atlas["key_papers"][:10]
|
||||
],
|
||||
}
|
||||
if topic:
|
||||
compact["topic_focus"] = topic_focus(topic)
|
||||
if topic_b:
|
||||
compact["compare_topic"] = topic_focus(topic_b)
|
||||
return json.dumps(compact, ensure_ascii=False, indent=2)[:12000]
|
||||
|
||||
|
||||
def atlas_prompt_for(mode: str, context: str) -> str:
|
||||
no_think = "不要输出思考过程，不要输出 <think> 标签，只输出最终结果。"
|
||||
if mode == "topic":
|
||||
return (
|
||||
"你是 Agent 研究编辑。请根据给定知识库统计，为当前主题写一份中文导读：\n"
|
||||
"1. 主题定位\n2. 为什么最近重要\n3. 主要技术路线\n4. 关键争议\n"
|
||||
"5. 必读论文阅读顺序\n6. 和相邻主题的关系\n7. 可以做的实验。\n"
|
||||
f"只基于给定数据，不要编造论文细节。{no_think}\n\n{context}"
|
||||
)
|
||||
if mode == "path":
|
||||
return (
|
||||
"你是研究导师。请基于给定知识库，为一个想系统学习 Agent 技术的人生成阅读路径。\n"
|
||||
"输出 4 个阶段：入门地图、核心系统、风险与评估、专题深入。每阶段给出主题、阅读目标、代表论文标题。\n"
|
||||
f"保持可执行，避免泛泛而谈。{no_think}\n\n{context}"
|
||||
)
|
||||
if mode == "compare":
|
||||
return (
|
||||
"你是技术路线分析师。请比较给定两个 Agent 主题：共同问题、差异、交叉区域、代表论文、工程启发。\n"
|
||||
f"只使用给定数据。{no_think}\n\n{context}"
|
||||
)
|
||||
return (
|
||||
"你是 Agent 知识库的研究编辑。请解释这张 Agent 论文 Atlas：\n"
|
||||
"1. 当前版图的主轴是什么\n2. 哪些主题是中心节点\n3. 哪些方向最近升温\n"
|
||||
"4. 哪些主题之间关系最强\n5. 下一步应该优先精读什么。\n"
|
||||
f"只基于给定数据。{no_think}\n\n{context}"
|
||||
)
|
||||
|
||||
|
||||
def arxiv_id_from_paper(paper: dict[str, Any]) -> str:
|
||||
url = str(paper.get("url") or "")
|
||||
match = re.search(r"arxiv\.org/(?:abs|html|pdf)/([0-9]{4}\.[0-9]+)", url)
|
||||
@@ -288,6 +654,8 @@ def call_ollama(model: str, prompt: str, mode: str) -> dict[str, Any]:
|
||||
options.update({"num_ctx": 3072, "num_predict": 700, "temperature": 0.1})
|
||||
elif mode == "deep":
|
||||
options.update({"num_ctx": 8192, "num_predict": 900, "temperature": 0.25})
|
||||
elif mode in {"atlas", "topic", "path", "compare"}:
|
||||
options.update({"num_ctx": 8192, "num_predict": 950, "temperature": 0.22})
|
||||
|
||||
payload = {
|
||||
"model": model,
|
||||
@@ -354,6 +722,12 @@ class PaperBrowserHandler(SimpleHTTPRequestHandler):
|
||||
papers, _ = load_papers()
|
||||
write_json(self, {"topics": topic_counts(papers)})
|
||||
return
|
||||
if path == "/api/atlas":
|
||||
write_json(self, build_atlas())
|
||||
return
|
||||
if path == "/api/topic":
|
||||
self.handle_topic(query)
|
||||
return
|
||||
if path == "/api/papers":
|
||||
self.handle_paper_search(query)
|
||||
return
|
||||
@@ -371,6 +745,9 @@ class PaperBrowserHandler(SimpleHTTPRequestHandler):
|
||||
if parsed.path == "/api/ai":
|
||||
self.handle_ai()
|
||||
return
|
||||
if parsed.path == "/api/atlas/ai":
|
||||
self.handle_atlas_ai()
|
||||
return
|
||||
write_error(self, HTTPStatus.NOT_FOUND, "not found")
|
||||
|
||||
def serve_file(self, target: Path, content_type: str) -> None:
|
||||
@@ -479,6 +856,16 @@ class PaperBrowserHandler(SimpleHTTPRequestHandler):
|
||||
text = (ROOT / paper["path"]).read_text(encoding="utf-8")
|
||||
write_json(self, {"paper": paper_public(paper), "markdown": text})
|
||||
|
||||
def handle_topic(self, query: dict[str, list[str]]) -> None:
|
||||
topic = (query.get("topic") or [""])[0].strip()
|
||||
if not topic:
|
||||
write_error(self, HTTPStatus.BAD_REQUEST, "topic is required")
|
||||
return
|
||||
try:
|
||||
write_json(self, topic_focus(topic))
|
||||
except KeyError:
|
||||
write_error(self, HTTPStatus.NOT_FOUND, "topic not found")
|
||||
|
||||
def handle_abstract(self, query: dict[str, list[str]]) -> None:
|
||||
paper_id = (query.get("id") or [""])[0]
|
||||
_, by_id = load_papers()
|
||||
@@ -545,6 +932,53 @@ class PaperBrowserHandler(SimpleHTTPRequestHandler):
|
||||
except Exception as exc: # noqa: BLE001
|
||||
write_error(self, HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
|
||||
|
||||
def handle_atlas_ai(self) -> None:
|
||||
try:
|
||||
payload = read_json_body(self)
|
||||
mode = str(payload.get("mode") or "atlas")
|
||||
topic = str(payload.get("topic") or "")
|
||||
topic_b = str(payload.get("topic_b") or "")
|
||||
refresh = bool(payload.get("refresh"))
|
||||
model = str(payload.get("model") or DEFAULT_MODELS.get(mode) or DEFAULT_MODELS["atlas"])
|
||||
if mode not in {"atlas", "topic", "path", "compare"}:
|
||||
write_error(self, HTTPStatus.BAD_REQUEST, "invalid atlas ai mode")
|
||||
return
|
||||
if mode == "topic" and not topic:
|
||||
write_error(self, HTTPStatus.BAD_REQUEST, "topic is required")
|
||||
return
|
||||
if mode == "compare" and (not topic or not topic_b):
|
||||
write_error(self, HTTPStatus.BAD_REQUEST, "topic and topic_b are required")
|
||||
return
|
||||
|
||||
ensure_cache_dirs()
|
||||
cache_key = f"atlas:{mode}:{topic}:{topic_b}"
|
||||
cache_path = ai_cache_path(cache_key, mode, model)
|
||||
if cache_path.exists() and not refresh:
|
||||
result = json.loads(cache_path.read_text(encoding="utf-8"))
|
||||
result["cached"] = True
|
||||
write_json(self, result)
|
||||
return
|
||||
|
||||
if not OLLAMA_LOCK.acquire(blocking=False):
|
||||
write_error(self, HTTPStatus.TOO_MANY_REQUESTS, "ollama is busy; try again later")
|
||||
return
|
||||
try:
|
||||
context = atlas_context(topic, topic_b)
|
||||
result = call_ollama(model, atlas_prompt_for(mode, context), mode)
|
||||
result.update(
|
||||
{
|
||||
"scope": cache_key,
|
||||
"cached": False,
|
||||
"created_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
|
||||
}
|
||||
)
|
||||
cache_path.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||||
write_json(self, result)
|
||||
finally:
|
||||
OLLAMA_LOCK.release()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
write_error(self, HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser()
|
||||
|
||||
Reference in New Issue
Block a user