完成和大模型的通信
This commit is contained in:
@@ -1,5 +1,10 @@
|
|||||||
{
|
{
|
||||||
"language": "Chinese",
|
"language": "Chinese",
|
||||||
"raw_dir": "raw",
|
"raw_dir": "raw",
|
||||||
"wiki_dir": "wiki"
|
"wiki_dir": "wiki",
|
||||||
|
"llm": {
|
||||||
|
"model": "qwen/qwen3.6-27b",
|
||||||
|
"api_base": "http://100.123.83.113:1234/v1",
|
||||||
|
"api_key": "123456"
|
||||||
|
}
|
||||||
}
|
}
|
||||||
@@ -5,10 +5,11 @@ description = "本项目用于构建 LLM wiki 知识库"
|
|||||||
requires-python = ">=3.12"
|
requires-python = ">=3.12"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"llama-index>=0.12.0",
|
"llama-index>=0.12.0",
|
||||||
|
"llama-index-llms-openai-like==0.7.2",
|
||||||
]
|
]
|
||||||
|
|
||||||
[tool.uv]
|
[tool.uv]
|
||||||
index-url = "https://pypi.tuna.tsinghua.edu.cn/simple"
|
index-url = "https://mirrors.aliyun.com/pypi/simple"
|
||||||
link-mode = "copy"
|
link-mode = "copy"
|
||||||
|
|
||||||
[dependency-groups]
|
[dependency-groups]
|
||||||
|
|||||||
@@ -5,20 +5,26 @@ from pathlib import Path
|
|||||||
|
|
||||||
CONFIG_PATH = Path(__file__).parent.parent / "config.json"
|
CONFIG_PATH = Path(__file__).parent.parent / "config.json"
|
||||||
|
|
||||||
DEFAULT_CONFIG: dict[str, str] = {
|
DEFAULT_CONFIG: dict = {
|
||||||
"language": "Chinese",
|
"language": "Chinese",
|
||||||
"raw_dir": "raw",
|
"raw_dir": "raw",
|
||||||
"wiki_dir": "wiki",
|
"wiki_dir": "wiki",
|
||||||
}
|
}
|
||||||
|
|
||||||
|
DEFAULT_LLM_CONFIG: dict = {
|
||||||
|
"model": "qwen/qwen3.6-27b",
|
||||||
|
"api_base": "http://100.123.83.113:1234/v1",
|
||||||
|
"api_key": "123456",
|
||||||
|
}
|
||||||
|
|
||||||
def load_config() -> dict[str, str]:
|
|
||||||
|
def load_config() -> dict:
|
||||||
"""
|
"""
|
||||||
加载项目配置。
|
加载项目配置。
|
||||||
|
|
||||||
返回
|
返回
|
||||||
-------
|
-------
|
||||||
dict[str, str]
|
dict
|
||||||
配置字典,包含 language、raw_dir、wiki_dir 等键。
|
配置字典,包含 language、raw_dir、wiki_dir 等键。
|
||||||
"""
|
"""
|
||||||
if CONFIG_PATH.is_file():
|
if CONFIG_PATH.is_file():
|
||||||
@@ -38,3 +44,19 @@ def get_language() -> str | None:
|
|||||||
if language.lower() == "auto":
|
if language.lower() == "auto":
|
||||||
return None
|
return None
|
||||||
return language
|
return language
|
||||||
|
|
||||||
|
|
||||||
|
def get_llm_config() -> dict[str, str]:
|
||||||
|
"""
|
||||||
|
获取 LLM 配置。
|
||||||
|
|
||||||
|
返回
|
||||||
|
-------
|
||||||
|
dict[str, str]
|
||||||
|
包含 model, api_base, api_key 的字典。
|
||||||
|
"""
|
||||||
|
config = load_config()
|
||||||
|
llm_raw = config.get("llm", {})
|
||||||
|
# 过滤空值,避免空字符串覆盖默认配置
|
||||||
|
llm_filtered = {k: v for k, v in llm_raw.items() if v}
|
||||||
|
return {**DEFAULT_LLM_CONFIG, **llm_filtered}
|
||||||
|
|||||||
86
src/llm.py
Normal file
86
src/llm.py
Normal file
@@ -0,0 +1,86 @@
|
|||||||
|
"""LLM client wrapper."""
|
||||||
|
|
||||||
|
import logging
|
||||||
|
|
||||||
|
from prompt import build_analysis_prompt
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
def _create_llm():
|
||||||
|
"""根据配置文件创建 LLM 实例。"""
|
||||||
|
from config import get_llm_config
|
||||||
|
|
||||||
|
from llama_index.llms.openai_like import OpenAILike
|
||||||
|
|
||||||
|
llm_config = get_llm_config()
|
||||||
|
return OpenAILike(
|
||||||
|
model=llm_config["model"],
|
||||||
|
api_base=llm_config["api_base"],
|
||||||
|
api_key=llm_config.get("api_key", "lm-studio"),
|
||||||
|
temperature=0.1,
|
||||||
|
max_tokens=128000,
|
||||||
|
request_timeout=300.0,
|
||||||
|
is_chat_model=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def query_analysis(
|
||||||
|
purpose: str,
|
||||||
|
index: str,
|
||||||
|
source_content: str = "",
|
||||||
|
configured_language: str | None = None,
|
||||||
|
) -> str:
|
||||||
|
"""
|
||||||
|
构建分析提示并发送给 LLM,返回分析报告。
|
||||||
|
|
||||||
|
参数
|
||||||
|
----------
|
||||||
|
purpose : str
|
||||||
|
wiki/purpose.md 的内容。
|
||||||
|
index : str
|
||||||
|
wiki/index.md 的内容。
|
||||||
|
source_content : str
|
||||||
|
源文档文本。
|
||||||
|
configured_language : str | None
|
||||||
|
显式覆盖语言设置。
|
||||||
|
|
||||||
|
返回
|
||||||
|
-------
|
||||||
|
str
|
||||||
|
LLM 返回的分析报告。
|
||||||
|
"""
|
||||||
|
from config import get_llm_config
|
||||||
|
|
||||||
|
from llama_index.core.llms import ChatMessage
|
||||||
|
|
||||||
|
system_prompt = build_analysis_prompt(
|
||||||
|
purpose,
|
||||||
|
index,
|
||||||
|
source_content=source_content,
|
||||||
|
configured_language=configured_language,
|
||||||
|
)
|
||||||
|
|
||||||
|
messages = [
|
||||||
|
ChatMessage(role="system", content=system_prompt),
|
||||||
|
ChatMessage(
|
||||||
|
role="user",
|
||||||
|
content=f"Analyze this source document:\n\n---\n\n{source_content}",
|
||||||
|
),
|
||||||
|
]
|
||||||
|
|
||||||
|
llm_config = get_llm_config()
|
||||||
|
llm = _create_llm()
|
||||||
|
logger.info("开始请求 LLM (model=%s, base=%s)", llm_config["model"], llm_config["api_base"])
|
||||||
|
try:
|
||||||
|
parts = []
|
||||||
|
for resp in llm.stream_chat(messages):
|
||||||
|
delta = resp.delta
|
||||||
|
if delta:
|
||||||
|
parts.append(delta)
|
||||||
|
text = "".join(parts)
|
||||||
|
logger.info("LLM 请求完成,响应总长度: %d 字符", len(text))
|
||||||
|
return text
|
||||||
|
except Exception as e:
|
||||||
|
logger.error("LLM 请求失败: %s", e)
|
||||||
|
raise
|
||||||
182
src/main.py
182
src/main.py
@@ -1,115 +1,93 @@
|
|||||||
# ---------------------------------------------------------------------------
|
|
||||||
# 辅助函数
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
|
|
||||||
|
|
||||||
def language_rule(source_content: str, configured_language: str | None = None) -> str:
|
|
||||||
"""
|
"""
|
||||||
生成语言规则提示。
|
Unified entry point.
|
||||||
|
|
||||||
参数
|
Re-exports from sub-modules for backward compatibility.
|
||||||
----------
|
|
||||||
source_content : str
|
|
||||||
用于语言检测回退的源文档文本。
|
|
||||||
configured_language : str | None
|
|
||||||
明确配置的输出语言(如 "Chinese"、"English"、"auto")。
|
|
||||||
传入 ``None`` 或 ``"auto"`` 以自动检测。
|
|
||||||
|
|
||||||
返回
|
|
||||||
-------
|
|
||||||
str
|
|
||||||
语言规则提示字符串。
|
|
||||||
"""
|
"""
|
||||||
if configured_language and configured_language.lower() != "auto":
|
|
||||||
return f"使用 {configured_language} 语言输出分析报告。"
|
|
||||||
|
|
||||||
# 简单的语言检测回退
|
import logging
|
||||||
if source_content and any("\u4e00" <= char <= "\u9fff" for char in source_content[:500]):
|
|
||||||
return "源文档包含中文内容,使用简体中文输出分析报告。"
|
|
||||||
|
|
||||||
return "使用与源文档相同的语言输出分析报告。"
|
from prompt import build_analysis_prompt, language_rule
|
||||||
|
from llm import query_analysis
|
||||||
|
|
||||||
|
__all__ = ["language_rule", "build_analysis_prompt", "query_analysis"]
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
def _setup_logging() -> None:
|
||||||
# 主函数:build_analysis_prompt
|
"""配置日志格式和级别,同时输出到终端和文件。"""
|
||||||
# ---------------------------------------------------------------------------
|
from pathlib import Path
|
||||||
|
|
||||||
|
log_path = Path(__file__).parent.parent / "llmwiki.log"
|
||||||
|
|
||||||
|
formatter = logging.Formatter(
|
||||||
|
"%(asctime)s [%(levelname)s] %(message)s",
|
||||||
|
datefmt="%Y-%m-%d %H:%M:%S",
|
||||||
|
)
|
||||||
|
|
||||||
|
file_handler = logging.FileHandler(log_path, encoding="utf-8")
|
||||||
|
file_handler.setFormatter(formatter)
|
||||||
|
|
||||||
|
stream_handler = logging.StreamHandler()
|
||||||
|
stream_handler.setFormatter(formatter)
|
||||||
|
|
||||||
|
logging.basicConfig(
|
||||||
|
level=logging.INFO,
|
||||||
|
format="%(asctime)s [%(levelname)s] %(message)s",
|
||||||
|
datefmt="%Y-%m-%d %H:%M:%S",
|
||||||
|
handlers=[file_handler, stream_handler],
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def build_analysis_prompt(
|
def main() -> None:
|
||||||
purpose: str,
|
"""执行入口:遍历 raw 目录,对每个源文件进行分析。"""
|
||||||
index: str,
|
import sys
|
||||||
source_content: str = "",
|
from pathlib import Path
|
||||||
configured_language: str | None = None,
|
|
||||||
) -> str:
|
|
||||||
"""
|
|
||||||
第一步提示:AI 阅读源文档并生成结构化分析报告。
|
|
||||||
这是「讨论」步骤——AI 在撰写 wiki 页面之前先对源文档进行推理。
|
|
||||||
|
|
||||||
参数
|
_setup_logging()
|
||||||
----------
|
|
||||||
purpose : str
|
|
||||||
wiki/purpose.md 的内容(可能为空)。
|
|
||||||
index : str
|
|
||||||
wiki/index.md 的内容(可能为空)。
|
|
||||||
source_content : str
|
|
||||||
用于语言检测回退的源文档文本。
|
|
||||||
configured_language : str | None
|
|
||||||
显式覆盖语言设置。传入 ``None`` 时从配置文件读取。
|
|
||||||
|
|
||||||
返回
|
from config import load_config
|
||||||
-------
|
|
||||||
str
|
|
||||||
完整的系统提示字符串。
|
|
||||||
"""
|
|
||||||
from config import get_language
|
|
||||||
|
|
||||||
if configured_language is None:
|
config = load_config()
|
||||||
configured_language = get_language()
|
raw_dir = Path(config.get("raw_dir", "raw"))
|
||||||
parts: list[str] = [
|
wiki_dir = Path(config.get("wiki_dir", "wiki"))
|
||||||
"你是一位专业研究分析师。阅读源文档并产出一份结构化的分析报告。",
|
|
||||||
"不要输出思维链、隐藏推理或思考过程。在内部进行推理,只撰写简洁的最终分析。",
|
|
||||||
"",
|
|
||||||
language_rule(source_content, configured_language),
|
|
||||||
"",
|
|
||||||
"你的分析需要包含:",
|
|
||||||
"",
|
|
||||||
"## 关键实体",
|
|
||||||
"列出文档中提到的人物、组织、产品、数据集、工具、芯片、算法等。对于每一项:",
|
|
||||||
"- 名称和类型",
|
|
||||||
"- 在源文档中的角色(核心还是边缘)",
|
|
||||||
"- 是否可能已存在于 wiki 中(查阅索引)",
|
|
||||||
"",
|
|
||||||
"## 关键概念",
|
|
||||||
"列出理论、方法、技术、现象。对于每一项:",
|
|
||||||
"- 名称和简要定义",
|
|
||||||
"- 为什么在源文档中重要",
|
|
||||||
"- 是否可能已存在于 wiki 中",
|
|
||||||
"",
|
|
||||||
"## 主要论点与发现",
|
|
||||||
"- 核心主张或结果是什么?",
|
|
||||||
"- 有什么证据支持这些主张?",
|
|
||||||
"- 证据的强度如何?",
|
|
||||||
"",
|
|
||||||
"## 与现有 Wiki 的关联",
|
|
||||||
"- 源文档与哪些现有页面相关?",
|
|
||||||
"- 它是否强化、挑战或扩展了现有知识?",
|
|
||||||
"",
|
|
||||||
"## 矛盾与张力",
|
|
||||||
"- 源文档中是否有任何内容与现有 wiki 内容冲突?",
|
|
||||||
"- 是否存在内部矛盾或注意事项?",
|
|
||||||
"",
|
|
||||||
"## 建议",
|
|
||||||
"- 应该创建或更新哪些 wiki 页面?",
|
|
||||||
"- 应该强调什么,弱化什么?",
|
|
||||||
"- 是否有任何需要向用户指出的开放性问题?",
|
|
||||||
"",
|
|
||||||
"分析要全面但简洁,专注于真正重要的内容。",
|
|
||||||
"",
|
|
||||||
"如果提供了文件夹上下文,请将其作为分类的提示——文件夹结构通常反映了用户的组织意图(例如,'papers/energy' 表示该文件是与能源相关的论文)。",
|
|
||||||
"",
|
|
||||||
f"## Wiki 目的(供参考)\n{purpose}" if purpose else "",
|
|
||||||
(f"## 当前 Wiki 索引(用于检查现有内容)\n{index}" if index else ""),
|
|
||||||
]
|
|
||||||
|
|
||||||
# 过滤空字符串并用换行符连接
|
# 读取 wiki 上下文
|
||||||
return "\n".join(part for part in parts if part)
|
purpose_path = wiki_dir / "purpose.md"
|
||||||
|
index_path = wiki_dir / "index.md"
|
||||||
|
|
||||||
|
purpose = purpose_path.read_text(encoding="utf-8") if purpose_path.exists() else ""
|
||||||
|
index = index_path.read_text(encoding="utf-8") if index_path.exists() else ""
|
||||||
|
|
||||||
|
logger.info("purpose.md: %s", "已加载" if purpose else "未找到")
|
||||||
|
logger.info("index.md: %s", "已加载" if index else "未找到")
|
||||||
|
|
||||||
|
# 收集 raw 目录下的 .md 文件
|
||||||
|
md_files = sorted(raw_dir.rglob("*.md"))
|
||||||
|
if not md_files:
|
||||||
|
logger.info("未找到 %s 下的 .md 文件", raw_dir)
|
||||||
|
sys.exit(0)
|
||||||
|
|
||||||
|
logger.info("共发现 %d 个源文件", len(md_files))
|
||||||
|
|
||||||
|
for fp in md_files:
|
||||||
|
source_content = fp.read_text(encoding="utf-8")
|
||||||
|
logger.info("=" * 60)
|
||||||
|
logger.info("分析: %s", fp.relative_to(raw_dir.parent))
|
||||||
|
logger.info("=" * 60)
|
||||||
|
|
||||||
|
try:
|
||||||
|
result = query_analysis(
|
||||||
|
purpose=purpose,
|
||||||
|
index=index,
|
||||||
|
source_content=source_content,
|
||||||
|
)
|
||||||
|
logger.info("分析完成,结果长度: %d 字符", len(result))
|
||||||
|
logger.info("结果: %s", result)
|
||||||
|
except Exception as e:
|
||||||
|
logger.error("分析失败: %s", e, exc_info=True)
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
108
src/prompt.py
Normal file
108
src/prompt.py
Normal file
@@ -0,0 +1,108 @@
|
|||||||
|
"""Analysis prompt builder."""
|
||||||
|
|
||||||
|
|
||||||
|
def language_rule(source_content: str, configured_language: str | None = None) -> str:
|
||||||
|
"""
|
||||||
|
生成语言规则提示。
|
||||||
|
|
||||||
|
参数
|
||||||
|
----------
|
||||||
|
source_content : str
|
||||||
|
用于语言检测回退的源文档文本。
|
||||||
|
configured_language : str | None
|
||||||
|
明确配置的输出语言(如 "Chinese"、"English"、"auto")。
|
||||||
|
传入 ``None`` 或 ``"auto"`` 以自动检测。
|
||||||
|
|
||||||
|
返回
|
||||||
|
-------
|
||||||
|
str
|
||||||
|
语言规则提示字符串。
|
||||||
|
"""
|
||||||
|
if configured_language and configured_language.lower() != "auto":
|
||||||
|
return f"使用 {configured_language} 语言输出分析报告。"
|
||||||
|
|
||||||
|
# 简单的语言检测回退
|
||||||
|
if source_content and any("\u4e00" <= char <= "\u9fff" for char in source_content[:500]):
|
||||||
|
return "源文档包含中文内容,使用简体中文输出分析报告。"
|
||||||
|
|
||||||
|
return "使用与源文档相同的语言输出分析报告。"
|
||||||
|
|
||||||
|
|
||||||
|
def build_analysis_prompt(
|
||||||
|
purpose: str,
|
||||||
|
index: str,
|
||||||
|
source_content: str = "",
|
||||||
|
configured_language: str | None = None,
|
||||||
|
) -> str:
|
||||||
|
"""
|
||||||
|
第一步提示:AI 阅读源文档并生成结构化分析报告。
|
||||||
|
这是「讨论」步骤——AI 在撰写 wiki 页面之前先对源文档进行推理。
|
||||||
|
|
||||||
|
参数
|
||||||
|
----------
|
||||||
|
purpose : str
|
||||||
|
wiki/purpose.md 的内容(可能为空)。
|
||||||
|
index : str
|
||||||
|
wiki/index.md 的内容(可能为空)。
|
||||||
|
source_content : str
|
||||||
|
用于语言检测回退的源文档文本。
|
||||||
|
configured_language : str | None
|
||||||
|
显式覆盖语言设置。传入 ``None`` 时从配置文件读取。
|
||||||
|
|
||||||
|
返回
|
||||||
|
-------
|
||||||
|
str
|
||||||
|
完整的系统提示字符串。
|
||||||
|
"""
|
||||||
|
from config import get_language
|
||||||
|
|
||||||
|
if configured_language is None:
|
||||||
|
configured_language = get_language()
|
||||||
|
parts: list[str] = [
|
||||||
|
"你是一位专业研究分析师。阅读源文档并产出一份结构化的分析报告。",
|
||||||
|
"不要输出思维链、隐藏推理或思考过程。在内部进行推理,只撰写简洁的最终分析。",
|
||||||
|
"",
|
||||||
|
language_rule(source_content, configured_language),
|
||||||
|
"",
|
||||||
|
"你的分析需要包含:",
|
||||||
|
"",
|
||||||
|
"## 关键实体",
|
||||||
|
"列出文档中提到的人物、组织、产品、数据集、工具、芯片、算法等。对于每一项:",
|
||||||
|
"- 名称和类型",
|
||||||
|
"- 在源文档中的角色(核心还是边缘)",
|
||||||
|
"- 是否可能已存在于 wiki 中(查阅索引)",
|
||||||
|
"",
|
||||||
|
"## 关键概念",
|
||||||
|
"列出理论、方法、技术、现象。对于每一项:",
|
||||||
|
"- 名称和简要定义",
|
||||||
|
"- 为什么在源文档中重要",
|
||||||
|
"- 是否可能已存在于 wiki 中",
|
||||||
|
"",
|
||||||
|
"## 主要论点与发现",
|
||||||
|
"- 核心主张或结果是什么?",
|
||||||
|
"- 有什么证据支持这些主张?",
|
||||||
|
"- 证据的强度如何?",
|
||||||
|
"",
|
||||||
|
"## 与现有 Wiki 的关联",
|
||||||
|
"- 源文档与哪些现有页面相关?",
|
||||||
|
"- 它是否强化、挑战或扩展了现有知识?",
|
||||||
|
"",
|
||||||
|
"## 矛盾与张力",
|
||||||
|
"- 源文档中是否有任何内容与现有 wiki 内容冲突?",
|
||||||
|
"- 是否存在内部矛盾或注意事项?",
|
||||||
|
"",
|
||||||
|
"## 建议",
|
||||||
|
"- 应该创建或更新哪些 wiki 页面?",
|
||||||
|
"- 应该强调什么,弱化什么?",
|
||||||
|
"- 是否有任何需要向用户指出的开放性问题?",
|
||||||
|
"",
|
||||||
|
"分析要全面但简洁,专注于真正重要的内容。",
|
||||||
|
"",
|
||||||
|
"如果提供了文件夹上下文,请将其作为分类的提示——文件夹结构通常反映了用户的组织意图(例如,'papers/energy' 表示该文件是与能源相关的论文)。",
|
||||||
|
"",
|
||||||
|
f"## Wiki 目的(供参考)\n{purpose}" if purpose else "",
|
||||||
|
(f"## 当前 Wiki 索引(用于检查现有内容)\n{index}" if index else ""),
|
||||||
|
]
|
||||||
|
|
||||||
|
# 过滤空字符串并用换行符连接
|
||||||
|
return "\n".join(part for part in parts if part)
|
||||||
42
tests/test_analysis_prompt.py
Normal file
42
tests/test_analysis_prompt.py
Normal file
@@ -0,0 +1,42 @@
|
|||||||
|
"""分析提示构建函数测试。"""
|
||||||
|
|
||||||
|
import sys
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||||||
|
|
||||||
|
from prompt import build_analysis_prompt
|
||||||
|
|
||||||
|
|
||||||
|
class TestBuildAnalysisPrompt:
|
||||||
|
"""测试分析提示构建函数。"""
|
||||||
|
|
||||||
|
def test_basic_prompt(self) -> None:
|
||||||
|
"""测试基础提示词包含核心结构。"""
|
||||||
|
prompt = build_analysis_prompt(purpose="", index="")
|
||||||
|
assert "专业研究分析师" in prompt
|
||||||
|
assert "关键实体" in prompt
|
||||||
|
|
||||||
|
def test_with_purpose(self) -> None:
|
||||||
|
"""测试传入目的时,提示词包含 Wiki 目的部分。"""
|
||||||
|
prompt = build_analysis_prompt(purpose="测试目的", index="")
|
||||||
|
assert "Wiki 目的" in prompt
|
||||||
|
assert "测试目的" in prompt
|
||||||
|
|
||||||
|
def test_with_index(self) -> None:
|
||||||
|
"""测试传入索引时,提示词包含 Wiki 索引部分。"""
|
||||||
|
prompt = build_analysis_prompt(purpose="", index="测试索引")
|
||||||
|
assert "当前 Wiki 索引" in prompt
|
||||||
|
assert "测试索引" in prompt
|
||||||
|
|
||||||
|
def test_empty_content(self) -> None:
|
||||||
|
"""测试空内容时,可选部分不出现在提示词中。"""
|
||||||
|
prompt = build_analysis_prompt(purpose="", index="")
|
||||||
|
assert "Wiki 目的" not in prompt
|
||||||
|
assert "当前 Wiki 索引" not in prompt
|
||||||
|
|
||||||
|
def test_language_from_config(self) -> None:
|
||||||
|
"""测试语言设置从配置文件读取。"""
|
||||||
|
prompt = build_analysis_prompt(purpose="", index="", configured_language=None)
|
||||||
|
assert "使用" in prompt
|
||||||
|
assert "语言输出分析报告" in prompt
|
||||||
23
tests/test_config.py
Normal file
23
tests/test_config.py
Normal file
@@ -0,0 +1,23 @@
|
|||||||
|
"""配置加载模块测试。"""
|
||||||
|
|
||||||
|
import sys
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||||||
|
|
||||||
|
from config import get_language, load_config
|
||||||
|
|
||||||
|
|
||||||
|
class TestConfig:
|
||||||
|
"""测试配置加载模块。"""
|
||||||
|
|
||||||
|
def test_load_config_returns_dict(self) -> None:
|
||||||
|
"""测试 load_config 返回字典且包含 language 键。"""
|
||||||
|
config = load_config()
|
||||||
|
assert isinstance(config, dict)
|
||||||
|
assert "language" in config
|
||||||
|
|
||||||
|
def test_get_language_from_config(self) -> None:
|
||||||
|
"""测试 get_language 能正常返回语言设置。"""
|
||||||
|
language = get_language()
|
||||||
|
assert language is not None
|
||||||
32
tests/test_language_rule.py
Normal file
32
tests/test_language_rule.py
Normal file
@@ -0,0 +1,32 @@
|
|||||||
|
"""语言规则生成函数测试。"""
|
||||||
|
|
||||||
|
import sys
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||||||
|
|
||||||
|
from prompt import language_rule
|
||||||
|
|
||||||
|
|
||||||
|
class TestLanguageRule:
|
||||||
|
"""测试语言规则生成函数。"""
|
||||||
|
|
||||||
|
def test_explicit_language(self) -> None:
|
||||||
|
"""测试显式指定语言时,提示中包含该语言名称。"""
|
||||||
|
result = language_rule("test content", "Chinese")
|
||||||
|
assert "Chinese" in result
|
||||||
|
|
||||||
|
def test_auto_with_chinese_content(self) -> None:
|
||||||
|
"""测试中文内容自动检测。"""
|
||||||
|
result = language_rule("这是一段中文内容", None)
|
||||||
|
assert "简体中文" in result
|
||||||
|
|
||||||
|
def test_auto_with_english_content(self) -> None:
|
||||||
|
"""测试英文内容自动检测。"""
|
||||||
|
result = language_rule("This is English content", None)
|
||||||
|
assert "源文档" in result
|
||||||
|
|
||||||
|
def test_auto_detection_string(self) -> None:
|
||||||
|
"""测试传入 "auto" 字符串时触发自动检测。"""
|
||||||
|
result = language_rule("test", "auto")
|
||||||
|
assert "源文档" in result
|
||||||
128
tests/test_llm.py
Normal file
128
tests/test_llm.py
Normal file
@@ -0,0 +1,128 @@
|
|||||||
|
"""LLM 模块测试。"""
|
||||||
|
|
||||||
|
from unittest.mock import MagicMock, patch
|
||||||
|
|
||||||
|
import sys
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||||||
|
|
||||||
|
|
||||||
|
class TestCreateLLM:
|
||||||
|
"""测试 LLM 实例创建。"""
|
||||||
|
|
||||||
|
@patch("config.get_llm_config")
|
||||||
|
@patch("llama_index.llms.openai_like.OpenAILike")
|
||||||
|
def test_create_llm_uses_config(self, mock_openai_like, mock_get_config):
|
||||||
|
"""测试 _create_llm 使用配置文件中的参数。"""
|
||||||
|
mock_get_config.return_value = {
|
||||||
|
"model": "qwen/qwen3.6-27b",
|
||||||
|
"api_base": "http://100.123.83.113:1234/v1",
|
||||||
|
"api_key": "123456",
|
||||||
|
}
|
||||||
|
|
||||||
|
from llm import _create_llm
|
||||||
|
|
||||||
|
_create_llm()
|
||||||
|
|
||||||
|
mock_openai_like.assert_called_once_with(
|
||||||
|
model="qwen/qwen3.6-27b",
|
||||||
|
api_base="http://100.123.83.113:1234/v1",
|
||||||
|
api_key="123456",
|
||||||
|
temperature=0.1,
|
||||||
|
max_tokens=128000,
|
||||||
|
request_timeout=300.0,
|
||||||
|
is_chat_model=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestQueryAnalysis:
|
||||||
|
"""测试 query_analysis 端到端流程。"""
|
||||||
|
|
||||||
|
@patch("llm._create_llm")
|
||||||
|
@patch("llm.build_analysis_prompt")
|
||||||
|
def test_query_analysis_calls_llm(self, mock_prompt, mock_create_llm):
|
||||||
|
"""测试 query_analysis 正确调用 prompt 构建和 LLM stream_chat。"""
|
||||||
|
mock_prompt.return_value = "test prompt"
|
||||||
|
mock_llm = MagicMock()
|
||||||
|
mock_resp = MagicMock()
|
||||||
|
mock_resp.delta = "analysis result"
|
||||||
|
mock_llm.stream_chat.return_value = [mock_resp]
|
||||||
|
mock_create_llm.return_value = mock_llm
|
||||||
|
|
||||||
|
from llm import query_analysis
|
||||||
|
|
||||||
|
result = query_analysis(purpose="test purpose", index="test index")
|
||||||
|
|
||||||
|
mock_prompt.assert_called_once_with(
|
||||||
|
"test purpose",
|
||||||
|
"test index",
|
||||||
|
source_content="",
|
||||||
|
configured_language=None,
|
||||||
|
)
|
||||||
|
assert mock_llm.stream_chat.called
|
||||||
|
assert result == "analysis result"
|
||||||
|
|
||||||
|
@patch("llm._create_llm")
|
||||||
|
@patch("llm.build_analysis_prompt")
|
||||||
|
def test_query_analysis_passes_source_content(self, mock_prompt, mock_create_llm):
|
||||||
|
"""测试 source_content 正确传递。"""
|
||||||
|
mock_prompt.return_value = "test prompt"
|
||||||
|
mock_llm = MagicMock()
|
||||||
|
mock_resp = MagicMock()
|
||||||
|
mock_resp.delta = "result"
|
||||||
|
mock_llm.stream_chat.return_value = [mock_resp]
|
||||||
|
mock_create_llm.return_value = mock_llm
|
||||||
|
|
||||||
|
from llm import query_analysis
|
||||||
|
|
||||||
|
query_analysis(
|
||||||
|
purpose="",
|
||||||
|
index="",
|
||||||
|
source_content="some source",
|
||||||
|
configured_language="English",
|
||||||
|
)
|
||||||
|
|
||||||
|
mock_prompt.assert_called_once_with(
|
||||||
|
"",
|
||||||
|
"",
|
||||||
|
source_content="some source",
|
||||||
|
configured_language="English",
|
||||||
|
)
|
||||||
|
|
||||||
|
@patch("llm._create_llm")
|
||||||
|
@patch("llm.build_analysis_prompt")
|
||||||
|
def test_query_analysis_returns_text(self, mock_prompt, mock_create_llm):
|
||||||
|
"""测试返回值是纯文本字符串。"""
|
||||||
|
mock_prompt.return_value = "prompt"
|
||||||
|
mock_llm = MagicMock()
|
||||||
|
mock_resp1 = MagicMock()
|
||||||
|
mock_resp1.delta = "## 关键实体\n"
|
||||||
|
mock_resp2 = MagicMock()
|
||||||
|
mock_resp2.delta = "- 测试实体"
|
||||||
|
mock_llm.stream_chat.return_value = [mock_resp1, mock_resp2]
|
||||||
|
mock_create_llm.return_value = mock_llm
|
||||||
|
|
||||||
|
from llm import query_analysis
|
||||||
|
|
||||||
|
result = query_analysis(purpose="", index="")
|
||||||
|
|
||||||
|
assert isinstance(result, str)
|
||||||
|
assert "关键实体" in result
|
||||||
|
|
||||||
|
@patch("llm._create_llm")
|
||||||
|
@patch("llm.build_analysis_prompt")
|
||||||
|
def test_query_analysis_empty_response(self, mock_prompt, mock_create_llm):
|
||||||
|
"""测试 LLM 返回空字符串时不会崩溃。"""
|
||||||
|
mock_prompt.return_value = "prompt"
|
||||||
|
mock_llm = MagicMock()
|
||||||
|
mock_resp = MagicMock()
|
||||||
|
mock_resp.delta = ""
|
||||||
|
mock_llm.stream_chat.return_value = [mock_resp]
|
||||||
|
mock_create_llm.return_value = mock_llm
|
||||||
|
|
||||||
|
from llm import query_analysis
|
||||||
|
|
||||||
|
result = query_analysis(purpose="", index="")
|
||||||
|
|
||||||
|
assert result == ""
|
||||||
@@ -1,145 +0,0 @@
|
|||||||
"""主模块测试套件。
|
|
||||||
|
|
||||||
本文件包含三个测试类,分别覆盖:
|
|
||||||
1. 配置加载功能(从 config.json 读取项目设置)
|
|
||||||
2. 语言规则生成(根据配置或自动检测确定输出语言)
|
|
||||||
3. 分析提示构建(组装完整的 AI 分析提示词)
|
|
||||||
"""
|
|
||||||
|
|
||||||
import sys
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
# 将 src 目录加入 Python 路径,使测试可以导入项目模块
|
|
||||||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
||||||
|
|
||||||
from config import get_language, load_config
|
|
||||||
from main import build_analysis_prompt, language_rule
|
|
||||||
|
|
||||||
|
|
||||||
class TestConfig:
|
|
||||||
"""测试配置加载模块。
|
|
||||||
|
|
||||||
验证 config.json 能否被正确读取,以及语言设置是否能正常获取。
|
|
||||||
"""
|
|
||||||
|
|
||||||
def test_load_config_returns_dict(self) -> None:
|
|
||||||
"""测试 load_config 返回字典且包含 language 键。
|
|
||||||
|
|
||||||
验证配置文件的解析结果是一个字典结构,
|
|
||||||
并且其中包含必需的 language 字段。
|
|
||||||
"""
|
|
||||||
config = load_config()
|
|
||||||
assert isinstance(config, dict)
|
|
||||||
assert "language" in config
|
|
||||||
|
|
||||||
def test_get_language_from_config(self) -> None:
|
|
||||||
"""测试 get_language 能正常返回语言设置。
|
|
||||||
|
|
||||||
当 config.json 中 language 不为 "auto" 时,
|
|
||||||
get_language 应返回非 None 值。
|
|
||||||
"""
|
|
||||||
language = get_language()
|
|
||||||
assert language is not None
|
|
||||||
|
|
||||||
|
|
||||||
class TestLanguageRule:
|
|
||||||
"""测试语言规则生成函数。
|
|
||||||
|
|
||||||
language_rule 负责根据配置或源文档内容,
|
|
||||||
生成指导 AI 输出语言的提示文本。
|
|
||||||
"""
|
|
||||||
|
|
||||||
def test_explicit_language(self) -> None:
|
|
||||||
"""测试显式指定语言时,提示中包含该语言名称。
|
|
||||||
|
|
||||||
当 configured_language 为 "Chinese" 时,
|
|
||||||
返回的提示应包含 "Chinese" 字样。
|
|
||||||
"""
|
|
||||||
result = language_rule("test content", "Chinese")
|
|
||||||
assert "Chinese" in result
|
|
||||||
|
|
||||||
def test_auto_with_chinese_content(self) -> None:
|
|
||||||
"""测试中文内容自动检测。
|
|
||||||
|
|
||||||
当源文档包含中文字符且未指定语言时,
|
|
||||||
应自动选择简体中文作为输出语言。
|
|
||||||
"""
|
|
||||||
result = language_rule("这是一段中文内容", None)
|
|
||||||
assert "简体中文" in result
|
|
||||||
|
|
||||||
def test_auto_with_english_content(self) -> None:
|
|
||||||
"""测试英文内容自动检测。
|
|
||||||
|
|
||||||
当源文档为纯英文且未指定语言时,
|
|
||||||
应返回跟随源文档语言的提示。
|
|
||||||
"""
|
|
||||||
result = language_rule("This is English content", None)
|
|
||||||
assert "源文档" in result
|
|
||||||
|
|
||||||
def test_auto_detection_string(self) -> None:
|
|
||||||
"""测试传入 "auto" 字符串时触发自动检测。
|
|
||||||
|
|
||||||
"auto" 和 None 等效,都表示启用自动语言检测。
|
|
||||||
"""
|
|
||||||
result = language_rule("test", "auto")
|
|
||||||
assert "源文档" in result
|
|
||||||
|
|
||||||
|
|
||||||
class TestBuildAnalysisPrompt:
|
|
||||||
"""测试分析提示构建函数。
|
|
||||||
|
|
||||||
build_analysis_prompt 负责将 wiki 目的、索引、
|
|
||||||
源文档内容和语言设置组装成完整的 AI 提示词。
|
|
||||||
"""
|
|
||||||
|
|
||||||
def test_basic_prompt(self) -> None:
|
|
||||||
"""测试基础提示词包含核心结构。
|
|
||||||
|
|
||||||
即使目的和索引都为空,提示词仍应包含
|
|
||||||
角色定义(专业研究分析师)和分析框架(关键实体)。
|
|
||||||
"""
|
|
||||||
prompt = build_analysis_prompt(purpose="", index="")
|
|
||||||
assert "专业研究分析师" in prompt
|
|
||||||
assert "关键实体" in prompt
|
|
||||||
|
|
||||||
def test_with_purpose(self) -> None:
|
|
||||||
"""测试传入目的时,提示词包含 Wiki 目的部分。
|
|
||||||
|
|
||||||
当 purpose 参数非空时,生成的提示词应包含
|
|
||||||
"Wiki 目的" 标题和传入的目的内容。
|
|
||||||
"""
|
|
||||||
prompt = build_analysis_prompt(purpose="测试目的", index="")
|
|
||||||
assert "Wiki 目的" in prompt
|
|
||||||
assert "测试目的" in prompt
|
|
||||||
|
|
||||||
def test_with_index(self) -> None:
|
|
||||||
"""测试传入索引时,提示词包含 Wiki 索引部分。
|
|
||||||
|
|
||||||
当 index 参数非空时,生成的提示词应包含
|
|
||||||
"当前 Wiki 索引" 标题和传入的索引内容。
|
|
||||||
"""
|
|
||||||
prompt = build_analysis_prompt(purpose="", index="测试索引")
|
|
||||||
assert "当前 Wiki 索引" in prompt
|
|
||||||
assert "测试索引" in prompt
|
|
||||||
|
|
||||||
def test_empty_content(self) -> None:
|
|
||||||
"""测试空内容时,可选部分不出现在提示词中。
|
|
||||||
|
|
||||||
当 purpose 和 index 都为空时,
|
|
||||||
提示词不应包含 "Wiki 目的" 和 "当前 Wiki 索引" 部分,
|
|
||||||
以保持提示词简洁。
|
|
||||||
"""
|
|
||||||
prompt = build_analysis_prompt(purpose="", index="")
|
|
||||||
assert "Wiki 目的" not in prompt
|
|
||||||
assert "当前 Wiki 索引" not in prompt
|
|
||||||
|
|
||||||
def test_language_from_config(self) -> None:
|
|
||||||
"""测试语言设置从配置文件读取。
|
|
||||||
|
|
||||||
当 configured_language 为 None 时,
|
|
||||||
build_analysis_prompt 会从 config.json 读取语言设置,
|
|
||||||
生成的提示词应包含语言规则。
|
|
||||||
"""
|
|
||||||
prompt = build_analysis_prompt(purpose="", index="", configured_language=None)
|
|
||||||
assert "使用" in prompt
|
|
||||||
assert "语言输出分析报告" in prompt
|
|
||||||
Reference in New Issue
Block a user