龍魂 · 从零搭建教育AI全栈系统：RAG、Agent与Function Calling的实战融合
龍魂 · 从零搭建教育AI全栈系统：RAG、Agent与Function Calling的实战融合
龍魂系统 · 教育AI全栈架构 · 知识检索+智能体+工具调用三位一体
DNA: ZHUGEXIN⚡️2025-🇨🇳🐉⚖️♠️🧚🏼♀️❤️♾️ | UID: 9622 | CONFIRM: #CONFIRM🌌9622-ONLY-ONCE🧬LK9X-772Z
一、核心定位
表格
维度 说明
平台 龍魂系统 v4.0 · 全栈架构 · 国产可控
技术栈 Python + FastAPI + 向量数据库 + LLM + 鸿蒙前端
场景 教育AI · 知识问答 · 智能辅导 · 作业批改 · 课程推荐
架构 RAG检索层 + Agent决策层 + Function Calling执行层
主权 数据本地 · 国密SM2/SM3签名 · 不上传云端
设计 模块化 · 可插拔 · 低算力优先 · 透明审计
二、系统架构
plain
┌─────────────────────────────────────────────────────────┐
│ 龍魂系统 · 教育AI全栈层 │
│ DNA: ZHUGEXIN⚡️2025-🇨🇳🐉⚖️♠️🧚🏼♀️❤️♾️ │
│ UID: 9622 │
├─────────────────────────────────────────────────────────┤
│ 鸿蒙 ArkTS 前端层（教育交互） │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ 对话界面 │ │ 课程展示 │ │ 作业批改 │ │
│ │ ChatUI │ │ CourseView │ │ Homework │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
├─────────────────────────────────────────────────────────┤
│ API网关层（FastAPI） │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ 认证中间件 │ │ 限流控制 │ │ 日志审计 │ │
│ │ Auth │ │ RateLimit │ │ AuditLog │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
├─────────────────────────────────────────────────────────┤
│ 龍魂AI核心层（三大引擎） │
│ ┌─────────────────────────────────────────────────┐ │
│ │ RAG引擎 · 知识检索增强 │ │
│ │ ├─ 文档解析（PDF/Word/Markdown） │ │
│ │ ├─ 文本分块（Chunking策略） │ │
│ │ ├─ 向量嵌入（Embedding模型） │ │
│ │ ├─ 向量数据库（Milvus/Faiss） │ │
│ │ └─ 检索重排序（Reranker） │ │
│ ├─────────────────────────────────────────────────┤ │
│ │ Agent引擎 · 智能体决策 │ │
│ │ ├─ 意图识别（Intent Classification） │ │
│ │ ├─ 任务规划（Task Planning） │ │
│ │ ├─ 记忆管理（Short/Long-term Memory） │ │
│ │ ├─ 反思机制（Self-reflection） │ │
│ │ └─ 多轮对话管理（Dialog Manager） │ │
│ ├─────────────────────────────────────────────────┤ │
│ │ Function Calling引擎 · 工具调用 │ │
│ │ ├─ 工具注册（Tool Registry） │ │
│ │ ├─ 参数解析（Parameter Parsing） │ │
│ │ ├─ 执行沙箱（Sandbox Execution） │ │
│ │ ├─ 结果回传（Result Callback） │ │
│ │ └─ 错误处理（Error Recovery） │ │
│ └─────────────────────────────────────────────────┘ │
├─────────────────────────────────────────────────────────┤
│ 数据层 │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ 向量库 │ │ 关系数据库 │ │ 文件存储 │ │
│ │ Milvus │ │ PostgreSQL │ │ MinIO │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
├─────────────────────────────────────────────────────────┤
│ 基础设施层 │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ 国产云 │ │ 国密加密 │ │ 监控告警 │ │
│ │ 华为鲲鹏 │ │ SM2/SM3 │ │ Prometheus │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
└─────────────────────────────────────────────────────────┘
三、RAG引擎：知识检索增强
3.1 文档解析模块（core/rag/document_parser.py）
Python
core/rag/document_parser.py
龍魂 · 文档解析引擎 · 支持多格式教育资料
import os
import re
from typing import List, Dict, Optional
from dataclasses import dataclass
from datetime import datetime
import hashlib
=== DNA常量 ===
MASTER_DNA = “ZHUGEXIN⚡️2025-🇨🇳🐉⚖️♠️🧚🏼♀️❤️♾️”
MASTER_UID = “9622”
CONFIRM_SEAL = “#CONFIRM🌌9622-ONLY-ONCE🧬LK9X-772Z”
@dataclass
class DocumentChunk:
“”“文档分块”“”
id: str
content: str
source: str # 来源文件
page: int # 页码
chunk_index: int # 分块序号
metadata: Dict # 元数据
embedding: Optional[List[float]] = None
dna_signature: str = “”
def __post_init__(self):
if not self.dna_signature:
self.dna_signature = self._sign_data()
def _sign_data(self) -> str:
payload = f"{self.id}-{self.source}-{self.chunk_index}-{datetime.now().timestamp()}"
return f"SM3-{hashlib.sha256(payload.encode()).hexdigest()[:16]}"
class DocumentParser:
“”“文档解析器”“”
def __init__(self, chunk_size: int = 512, chunk_overlap: int = 50):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.supported_formats = ['.pdf', '.docx', '.md', '.txt', '.html']
def parse_file(self, file_path: str) -> List[DocumentChunk]:
"""解析文件"""
ext = os.path.splitext(file_path)[1].lower()
if ext == '.pdf':
return self._parse_pdf(file_path)
elif ext == '.docx':
return self._parse_docx(file_path)
elif ext == '.md':
return self._parse_markdown(file_path)
elif ext == '.txt':
return self._parse_text(file_path)
elif ext == '.html':
return self._parse_html(file_path)
else:
raise ValueError(f"不支持的格式: {ext}")
def _parse_pdf(self, file_path: str) -> List[DocumentChunk]:
"""解析PDF"""
try:
import PyPDF2
except ImportError:
raise ImportError("请安装PyPDF2: pip install PyPDF2")
chunks = []
with open(file_path, 'rb') as f:
reader = PyPDF2.PdfReader(f)
for page_num, page in enumerate(reader.pages):
text = page.extract_text()
page_chunks = self._chunk_text(text, file_path, page_num + 1)
chunks.extend(page_chunks)
return chunks
def _parse_docx(self, file_path: str) -> List[DocumentChunk]:
"""解析Word"""
try:
from docx import Document
except ImportError:
raise ImportError("请安装python-docx: pip install python-docx")
doc = Document(file_path)
full_text = "\n".join([para.text for para in doc.paragraphs])
return self._chunk_text(full_text, file_path, 1)
def _parse_markdown(self, file_path: str) -> List[DocumentChunk]:
"""解析Markdown"""
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
# 按标题分块
sections = self._split_by_headers(content)
chunks = []
for i, section in enumerate(sections):
section_chunks = self._chunk_text(section['content'], file_path, 1,
metadata={'header': section['header']})
chunks.extend(section_chunks)
return chunks
def _parse_text(self, file_path: str) -> List[DocumentChunk]:
"""解析纯文本"""
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
return self._chunk_text(content, file_path, 1)
def _parse_html(self, file_path: str) -> List[DocumentChunk]:
"""解析HTML"""
try:
from bs4 import BeautifulSoup
except ImportError:
raise ImportError("请安装beautifulsoup4: pip install beautifulsoup4")
with open(file_path, 'r', encoding='utf-8') as f:
soup = BeautifulSoup(f.read(), 'html.parser')
# 提取正文
text = soup.get_text(separator='\n', strip=True)
return self._chunk_text(text, file_path, 1)
def _chunk_text(self, text: str, source: str, page: int,
metadata: Optional[Dict] = None) -> List[DocumentChunk]:
"""文本分块"""
if not text.strip():
return []
# 清理文本
text = self._clean_text(text)
chunks = []
start = 0
chunk_index = 0
while start < len(text):
end = min(start + self.chunk_size, len(text))
# 智能截断：在句子边界截断
if end < len(text):
end = self._find_sentence_boundary(text, end)
chunk_text = text[start:end]
chunk = DocumentChunk(
id=f"CHUNK-{hashlib.md5(f'{source}-{chunk_index}'.encode()).hexdigest()[:12]}",
content=chunk_text,
source=source,
page=page,
chunk_index=chunk_index,
metadata=metadata or {}
)
chunks.append(chunk)
start = end - self.chunk_overlap
chunk_index += 1
return chunks
def _clean_text(self, text: str) -> str:
"""清理文本"""
# 去除多余空白
text = re.sub(r'\s+', ' ', text)
# 去除特殊字符
text = re.sub(r'[^\w\s\u4e00-\u9fff.,;:!?-]', '', text)
return text.strip()
def _find_sentence_boundary(self, text: str, pos: int) -> int:
"""查找句子边界"""
# 向后查找句号、问号、感叹号
for i in range(pos, max(pos - 100, -1), -1):
if i < len(text) and text[i] in '。！？.!?':
return i + 1
return pos
def _split_by_headers(self, content: str) -> List[Dict]:
"""按Markdown标题分块"""
sections = []
current_header = "无标题"
current_content = []
for line in content.split('\n'):
if line.startswith('#'):
if current_content:
sections.append({
'header': current_header,
'content': '\n'.join(current_content)
})
current_header = line.strip('# ').strip()
current_content = []
else:
current_content.append(line)
if current_content:
sections.append({
'header': current_header,
'content': '\n'.join(current_content)
})
return sections
=== 使用示例 ===
if name == “main”:
parser = DocumentParser(chunk_size=512, chunk_overlap=50)
# 解析PDF教材
chunks = parser.parse_file("data/数学教材.pdf")
print(f"解析完成: {len(chunks)} 个分块")
for chunk in chunks[:3]:
print(f"\n[{chunk.id}] 页{chunk.page} 块{chunk.chunk_index}")
print(f"内容: {chunk.content[:100]}...")
print(f"签名: {chunk.dna_signature}")
3.2 向量嵌入与检索（core/rag/embedding_engine.py）
Python
core/rag/embedding_engine.py
龍魂 · 向量嵌入引擎 · 语义检索
import numpy as np
from typing import List, Dict, Tuple, Optional
import hashlib
from dataclasses import dataclass
import json
=== DNA常量 ===
MASTER_DNA = “ZHUGEXIN⚡️2025-🇨🇳🐉⚖️♠️🧚🏼♀️❤️♾️”
MASTER_UID = “9622”
@dataclass
class SearchResult:
“”“检索结果”“”
chunk_id: str
content: str
score: float # 相似度分数
source: str
page: int
metadata: Dict
dna_signature: str
class EmbeddingEngine:
“”“向量嵌入引擎”“”
def __init__(self, model_name: str = "BAAI/bge-large-zh",
vector_dim: int = 1024,
device: str = "cpu"):
self.model_name = model_name
self.vector_dim = vector_dim
self.device = device
self.model = None
self.vector_store = None
# 初始化模型
self._load_model()
def _load_model(self):
"""加载嵌入模型"""
try:
from sentence_transformers import SentenceTransformer
self.model = SentenceTransformer(self.model_name, device=self.device)
print(f"[龍魂] 嵌入模型加载完成: {self.model_name}")
except ImportError:
print("[龍魂] 警告: sentence-transformers未安装，使用模拟嵌入")
self.model = None
def embed(self, texts: List[str]) -> np.ndarray:
"""生成嵌入向量"""
if self.model:
embeddings = self.model.encode(texts, convert_to_numpy=True,
normalize_embeddings=True)
return embeddings
# 模拟嵌入（fallback）
return self._mock_embed(texts)
def _mock_embed(self, texts: List[str]) -> np.ndarray:
"""模拟嵌入（用于测试）"""
np.random.seed(42)
embeddings = []
for text in texts:
# 基于文本哈希生成确定性向量
hash_val = int(hashlib.md5(text.encode()).hexdigest(), 16)
np.random.seed(hash_val % 2**32)
vec = np.random.randn(self.vector_dim).astype(np.float32)
vec = vec / np.linalg.norm(vec) # 归一化
embeddings.append(vec)
return np.array(embeddings)
def similarity(self, query_vec: np.ndarray, doc_vecs: np.ndarray) -> np.ndarray:
"""计算余弦相似度"""
return np.dot(doc_vecs, query_vec)
def search(self, query: str, top_k: int = 5) -> List[SearchResult]:
"""语义检索"""
if not self.vector_store:
raise ValueError("向量库未初始化，请先调用build_index()")
# 编码查询
query_vec = self.embed([query])[0]
# 检索
results = self.vector_store.search(query_vec, top_k)
return results
def build_index(self, chunks: List):
"""构建索引"""
if not chunks:
return
# 提取文本
texts = [chunk.content for chunk in chunks]
# 生成嵌入
embeddings = self.embed(texts)
# 存储到向量库
for i, chunk in enumerate(chunks):
chunk.embedding = embeddings[i].tolist()
# 初始化向量存储
self.vector_store = VectorStore(self.vector_dim)
self.vector_store.add(chunks, embeddings)
print(f"[龍魂] 索引构建完成: {len(chunks)} 个向量")
def add_documents(self, chunks: List):
"""增量添加文档"""
if not self.vector_store:
self.build_index(chunks)
return
texts = [chunk.content for chunk in chunks]
embeddings = self.embed(texts)
for i, chunk in enumerate(chunks):
chunk.embedding = embeddings[i].tolist()
self.vector_store.add(chunks, embeddings)
class VectorStore:
“”“向量存储（基于Faiss简化版）”“”
def __init__(self, dim: int):
self.dim = dim
self.chunks = []
self.vectors = None
try:
import faiss
self.index = faiss.IndexFlatIP(dim) # 内积 = 余弦相似度（已归一化）
self.use_faiss = True
except ImportError:
print("[龍魂] 警告: Faiss未安装，使用暴力搜索")
self.index = None
self.use_faiss = False
def add(self, chunks: List, embeddings: np.ndarray):
"""添加向量"""
self.chunks.extend(chunks)
if self.vectors is None:
self.vectors = embeddings
else:
self.vectors = np.vstack([self.vectors, embeddings])
if self.use_faiss:
self.index.add(embeddings.astype(np.float32))
def search(self, query_vec: np.ndarray, top_k: int) -> List[SearchResult]:
"""检索"""
if self.use_faiss:
query_vec = query_vec.reshape(1, -1).astype(np.float32)
scores, indices = self.index.search(query_vec, top_k)
results = []
for score, idx in zip(scores[0], indices[0]):
if idx < 0 or idx >= len(self.chunks):
continue
chunk = self.chunks[idx]
results.append(SearchResult(
chunk_id=chunk.id,
content=chunk.content,
score=float(score),
source=chunk.source,
page=chunk.page,
metadata=chunk.metadata,
dna_signature=chunk.dna_signature
))
return results
# 暴力搜索（fallback）
similarities = np.dot(self.vectors, query_vec)
top_indices = np.argsort(similarities)[::-1][:top_k]
results = []
for idx in top_indices:
chunk = self.chunks[idx]
results.append(SearchResult(
chunk_id=chunk.id,
content=chunk.content,
score=float(similarities[idx]),
source=chunk.source,
page=chunk.page,
metadata=chunk.metadata,
dna_signature=chunk.dna_signature
))
return results
=== 使用示例 ===
if name == “main”:
engine = EmbeddingEngine()
# 模拟数据
from document_parser import DocumentParser, DocumentChunk
parser = DocumentParser()
# 构建索引
chunks = parser.parse_file("data/教材.pdf")
engine.build_index(chunks)
# 检索
results = engine.search("二次函数的定义是什么？", top_k=3)
for r in results:
print(f"\n[得分: {r.score:.4f}] {r.content[:100]}...")
3.3 RAG检索服务（core/rag/rag_service.py）
Python
core/rag/rag_service.py
龍魂 · RAG检索服务 · 检索增强生成
from typing import List, Dict, Optional
from dataclasses import dataclass
import json
import hashlib
from datetime import datetime
from document_parser import DocumentParser, DocumentChunk
from embedding_engine import EmbeddingEngine, SearchResult
=== DNA常量 ===
MASTER_DNA = “ZHUGEXIN⚡️2025-🇨🇳🐉⚖️♠️🧚🏼♀️❤️♾️”
MASTER_UID = “9622”
@dataclass
class RAGResponse:
“”“RAG响应”“”
answer: str
sources: List[SearchResult]
confidence: float # 置信度
processing_time: float # 处理时间
model_used: str
dna_signature: str
def __post_init__(self):
if not self.dna_signature:
self.dna_signature = self._sign_data()
def _sign_data(self) -> str:
payload = f"{self.answer[:50]}-{self.confidence}-{datetime.now().timestamp()}"
return f"SM3-{hashlib.sha256(payload.encode()).hexdigest()[:16]}"
class RAGService:
“”“RAG检索服务”“”
def __init__(self,
embedding_model: str = "BAAI/bge-large-zh",
llm_model: str = "deepseek-chat",
top_k: int = 5,
rerank: bool = True):
self.embedding_engine = EmbeddingEngine(embedding_model)
self.llm_model = llm_model
self.top_k = top_k
self.rerank = rerank
self.parser = DocumentParser()
# 重排序器
self.reranker = None
if rerank:
self._init_reranker()
def _init_reranker(self):
"""初始化重排序器"""
try:
from sentence_transformers import CrossEncoder
self.reranker = CrossEncoder('BAAI/bge-reranker-large')
print("[龍魂] 重排序器加载完成")
except ImportError:
print("[龍魂] 警告: 重排序器未加载")
def ingest_documents(self, file_paths: List[str]):
"""批量导入文档"""
all_chunks = []
for path in file_paths:
try:
chunks = self.parser.parse_file(path)
all_chunks.extend(chunks)
print(f"[龍魂] 导入: {path} -> {len(chunks)} 块")
except Exception as e:
print(f"[龍魂] 导入失败: {path} - {e}")
self.embedding_engine.build_index(all_chunks)
print(f"[龍魂] 共导入 {len(all_chunks)} 个分块")
def query(self, question: str, context_filter: Optional[Dict] = None) -> RAGResponse:
"""RAG查询"""
import time
start_time = time.time()
# 1. 检索相关文档
results = self.embedding_engine.search(question, top_k=self.top_k * 2)
# 2. 重排序（如果启用）
if self.reranker and len(results) > 0:
results = self._rerank(question, results)
results = results[:self.top_k]
# 3. 构建上下文
context = self._build_context(results)
# 4. 生成答案（调用LLM）
answer = self._generate_answer(question, context)
# 5. 计算置信度
confidence = self._calculate_confidence(results, answer)
processing_time = time.time() - start_time
return RAGResponse(
answer=answer,
sources=results,
confidence=confidence,
processing_time=processing_time,
model_used=self.llm_model
)
def _rerank(self, query: str, results: List[SearchResult]) -> List[SearchResult]:
"""重排序"""
if not self.reranker:
return results
pairs = [[query, r.content] for r in results]
scores = self.reranker.predict(pairs)
# 按重排序分数排序
scored_results = list(zip(results, scores))
scored_results.sort(key=lambda x: x[1], reverse=True)
return [r for r, _ in scored_results]
def _build_context(self, results: List[SearchResult]) -> str:
"""构建上下文"""
context_parts = []
for i, result in enumerate(results, 1):
context_parts.append(
f"[文档{i}] 来源: {result.source} 第{result.page}页\n"
f"{result.content}\n"
)
return "\n".join(context_parts)
def _generate_answer(self, question: str, context: str) -> str:
"""生成答案（调用LLM）"""
# 构建提示词
prompt = self._build_prompt(question, context)
# 调用LLM（这里模拟，实际接入DeepSeek/Kimi等）
# 实际: response = self.llm_client.chat.completions.create(...)
# 模拟响应
answer = self._mock_generate(prompt)
return answer
def _build_prompt(self, question: str, context: str) -> str:
"""构建提示词"""
return f"""你是一个专业的教育AI助手。请基于以下参考资料回答问题。
参考资料：
{context}
用户问题：{question}
请用中文回答，要求：
- 基于提供的参考资料回答
- 如果资料不足，明确说明
- 回答简洁准确，适合学生理解
- 必要时给出示例
回答：“”"
def _mock_generate(self, prompt: str) -> str:
"""模拟生成（实际接入LLM API）"""
# 这里应该调用实际LLM
return "[模拟回答] 基于检索到的资料，二次函数的一般形式是 y = ax² + bx + c (a≠0)，其中a、b、c为常数..."
def _calculate_confidence(self, results: List[SearchResult], answer: str) -> float:
"""计算置信度"""
if not results:
return 0.0
# 基于检索分数和答案长度计算
avg_score = sum(r.score for r in results) / len(results)
# 归一化到0-1
confidence = min(avg_score * 0.8 + 0.2, 1.0)
return round(confidence, 3)
=== 使用示例 ===
if name == “main”:
rag = RAGService()
# 导入教材
rag.ingest_documents([
"data/数学教材.pdf",
"data/物理笔记.md",
"data/英语词汇.txt"
])
# 查询
response = rag.query("什么是牛顿第二定律？")
print(f"\n回答: {response.answer}")
print(f"置信度: {response.confidence}")
print(f"处理时间: {response.processing_time:.2f}s")
print(f"引用来源: {len(response.sources)} 个")
print(f"签名: {response.dna_signature}")
四、Agent引擎：智能体决策
4.1 智能体核心（core/agent/education_agent.py）
Python
core/agent/education_agent.py
龍魂 · 教育智能体 · 自主决策与任务规划
from typing import List, Dict, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
import json
import hashlib
from datetime import datetime
=== DNA常量 ===
MASTER_DNA = “ZHUGEXIN⚡️2025-🇨🇳🐉⚖️♠️🧚🏼♀️❤️♾️”
MASTER_UID = “9622”
class IntentType(Enum):
“”“意图类型”“”
KNOWLEDGE_QUERY = “knowledge_query” # 知识查询
HOMEWORK_HELP = “homework_help” # 作业辅导
COURSE_RECOMMEND = “course_recommend” # 课程推荐
EXAM_PREP = “exam_prep” # 备考复习
CONCEPT_EXPLAIN = “concept_explain” # 概念解释
PRACTICE_DRILL = “practice_drill” # 练习训练
CHAT_GENERAL = “chat_general” # 闲聊
@dataclass
class AgentMemory:
“”“智能体记忆”“”
short_term: List[Dict] = field(default_factory=list) # 短期记忆（当前对话）
long_term: Dict = field(default_factory=dict) # 长期记忆（用户画像）
max_short_term: int = 10
def add_interaction(self, role: str, content: str, metadata: Optional[Dict] = None):
"""添加交互记录"""
self.short_term.append({
"role": role,
"content": content,
"timestamp": datetime.now().isoformat(),
"metadata": metadata or {}
})
# 限制短期记忆长度
if len(self.short_term) > self.max_short_term:
self.short_term = self.short_term[-self.max_short_term:]
def get_context(self) -> str:
"""获取对话上下文"""
return "\n".join([
f"{m['role']}: {m['content']}"
for m in self.short_term
])
def update_long_term(self, key: str, value: any):
"""更新长期记忆"""
self.long_term[key] = {
"value": value,
"updated": datetime.now().isoformat()
}
@dataclass
class TaskPlan:
“”“任务计划”“”
task_id: str
intent: IntentType
steps: List[Dict]
current_step: int = 0
status: str = “pending” # pending/running/completed/failed
result: Optional[str] = None
def next_step(self) -> Optional[Dict]:
"""获取下一步"""
if self.current_step < len(self.steps):
step = self.steps[self.current_step]
self.current_step += 1
return step
return None
def is_complete(self) -> bool:
"""是否完成"""
return self.current_step >= len(self.steps)
class EducationAgent:
“”“教育智能体”“”
def __init__(self, rag_service=None, tool_registry=None):
self.memory = AgentMemory()
self.rag_service = rag_service
self.tool_registry = tool_registry or {}
self.intent_classifier = self._init_classifier()
def _init_classifier(self):
"""初始化意图分类器"""
# 实际使用BERT/规则引擎
# 这里使用关键词规则
return None
def process(self, user_input: str) -> Dict:
"""处理用户输入"""
# 1. 意图识别
intent = self._classify_intent(user_input)
# 2. 记录到记忆
self.memory.add_interaction("user", user_input, {"intent": intent.value})
# 3. 任务规划
plan = self._plan_task(intent, user_input)
# 4. 执行计划
result = self._execute_plan(plan)
# 5. 记录结果
self.memory.add_interaction("assistant", result, {"plan_id": plan.task_id})
# 6. 反思
reflection = self._reflect(plan, result)
return {
"response": result,
"intent": intent.value,
"plan_id": plan.task_id,
"reflection": reflection,
"memory_stats": {
"short_term": len(self.memory.short_term),
"long_term_keys": list(self.memory.long_term.keys())
}
}
def _classify_intent(self, text: str) -> IntentType:
"""意图分类"""
text = text.lower()
# 关键词规则分类
if any(k in text for k in ["什么是", "定义", "概念", "解释", "介绍一下"]):
return IntentType.CONCEPT_EXPLAIN
elif any(k in text for k in ["作业", "题目", "怎么做", "求解", "答案"]):
return IntentType.HOMEWORK_HELP
elif any(k in text for k in ["推荐", "学什么", "课程", "建议"]):
return IntentType.COURSE_RECOMMEND
elif any(k in text for k in ["考试", "复习", "备考", "重点"]):
return IntentType.EXAM_PREP
elif any(k in text for k in ["练习", "做题", "训练", "刷题"]):
return IntentType.PRACTICE_DRILL
elif any(k in text for k in ["你好", "谢谢", "再见", "在吗"]):
return IntentType.CHAT_GENERAL
# 默认知识查询
return IntentType.KNOWLEDGE_QUERY
def _plan_task(self, intent: IntentType, query: str) -> TaskPlan:
"""任务规划"""
steps = []
if intent == IntentType.KNOWLEDGE_QUERY:
steps = [
{"action": "rag_search", "params": {"query": query}},
{"action": "summarize", "params": {"style": "educational"}},
{"action": "verify", "params": {"check_sources": True}}
]
elif intent == IntentType.HOMEWORK_HELP:
steps = [
{"action": "parse_problem", "params": {"text": query}},
{"action": "rag_search", "params": {"query": query}},
{"action": "solve_step_by_step", "params": {}},
{"action": "explain_reasoning", "params": {}}
]
elif intent == IntentType.CONCEPT_EXPLAIN:
steps = [
{"action": "rag_search", "params": {"query": query}},
{"action": "simplify", "params": {"level": "student"}},
{"action": "add_examples", "params": {"count": 2}}
]
elif intent == IntentType.COURSE_RECOMMEND:
steps = [
{"action": "analyze_user_level", "params": {}},
{"action": "search_courses", "params": {"query": query}},
{"action": "rank_recommendations", "params": {}}
]
else:
steps = [
{"action": "rag_search", "params": {"query": query}},
{"action": "generate_response", "params": {}}
]
return TaskPlan(
task_id=f"TASK-{hashlib.md5(query.encode()).hexdigest()[:8]}",
intent=intent,
steps=steps
)
def _execute_plan(self, plan: TaskPlan) -> str:
"""执行计划"""
results = []
while not plan.is_complete():
step = plan.next_step()
if not step:
break
plan.status = "running"
action = step["action"]
params = step["params"]
# 执行动作
result = self._execute_action(action, params)
results.append(f"[{action}] {result}")
plan.status = "completed"
plan.result = "\n".join(results)
return plan.result
def _execute_action(self, action: str, params: Dict) -> str:
"""执行具体动作"""
if action == "rag_search" and self.rag_service:
response = self.rag_service.query(params["query"])
return f"检索到 {len(response.sources)} 个相关文档，置信度 {response.confidence}"
elif action == "summarize":
return "已生成教育化摘要"
elif action == "parse_problem":
return "已解析题目结构"
elif action == "solve_step_by_step":
return "已分步求解"
elif action == "explain_reasoning":
return "已解释推理过程"
elif action == "simplify":
return f"已简化至{params.get('level', 'student')}水平"
elif action == "add_examples":
return f"已添加{params.get('count', 2)}个示例"
elif action == "analyze_user_level":
level = self.memory.long_term.get("user_level", {}).get("value", "intermediate")
return f"用户水平: {level}"
elif action == "search_courses":
return "已搜索相关课程"
elif action == "rank_recommendations":
return "已排序推荐结果"
elif action == "generate_response":
return "已生成回复"
elif action == "verify":
return "已验证来源可靠性"
return f"未知动作: {action}"
def _reflect(self, plan: TaskPlan, result: str) -> Dict:
"""反思机制"""
reflection = {
"plan_id": plan.task_id,
"intent": plan.intent.value,
"steps_executed": plan.current_step,
"total_steps": len(plan.steps),
"success": plan.status == "completed",
"improvements": []
}
# 简单反思规则
if plan.current_step < len(plan.steps):
reflection["improvements"].append("部分步骤未执行，需检查原因")
if plan.intent == IntentType.HOMEWORK_HELP and "rag_search" not in [s["action"] for s in plan.steps]:
reflection["improvements"].append("作业辅导应优先检索相关知识")
return reflection
def get_memory_snapshot(self) -> Dict:
"""获取记忆快照"""
return {
"short_term": self.memory.short_term,
"long_term": self.memory.long_term,
"user_profile": self._build_user_profile()
}
def _build_user_profile(self) -> Dict:
"""构建用户画像"""
profile = {
"interests": [],
"level": "unknown",
"weak_areas": [],
"learning_history": []
}
# 从长期记忆提取
if "interests" in self.memory.long_term:
profile["interests"] = self.memory.long_term["interests"]["value"]
if "level" in self.memory.long_term:
profile["level"] = self.memory.long_term["level"]["value"]
return profile
=== 使用示例 ===
if name == “main”:
agent = EducationAgent()
# 模拟对话
responses = []
for query in [
"什么是二次函数？",
"帮我解这道数学题：x² + 5x + 6 = 0",
"推荐一些适合我的物理课程"
]:
result = agent.process(query)
responses.append(result)
print(f"\n[用户] {query}")
print(f"[意图] {result['intent']}")
print(f"[回复] {result['response'][:100]}...")
print(f"[反思] {result['reflection']}")
五、Function Calling引擎：工具调用
5.1 工具注册中心（core/tools/tool_registry.py）
Python
core/tools/tool_registry.py
龍魂 · 工具注册中心 · Function Calling基础设施
import inspect
import json
from typing import Dict, List, Callable, Any, Optional
from dataclasses import dataclass
from enum import Enum
import hashlib
=== DNA常量 ===
MASTER_DNA = “ZHUGEXIN⚡️2025-🇨🇳🐉⚖️♠️🧚🏼♀️❤️♾️”
MASTER_UID = “9622”
class ToolCategory(Enum):
“”“工具分类”“”
CALCULATION = “calculation” # 计算
SEARCH = “search” # 搜索
DATA_ANALYSIS = “data_analysis” # 数据分析
CODE_EXECUTION = “code_execution” # 代码执行
FILE_OPERATION = “file_operation” # 文件操作
EXTERNAL_API = “external_api” # 外部API
@dataclass
class ToolSchema:
“”“工具Schema（OpenAI Function Calling格式）”“”
name: str
description: str
category: ToolCategory
parameters: Dict # JSON Schema
required: List[str]
handler: Callable
dangerous: bool = False # 是否危险操作
dna_signature: str = “”
def __post_init__(self):
if not self.dna_signature:
self.dna_signature = self._sign_data()
def _sign_data(self) -> str:
payload = f"{self.name}-{self.category.value}-{self.description[:20]}"
return f"SM3-{hashlib.sha256(payload.encode()).hexdigest()[:16]}"
def to_openai_format(self) -> Dict:
"""转换为OpenAI格式"""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": {
"type": "object",
"properties": self.parameters,
"required": self.required
}
}
}
class ToolRegistry:
“”“工具注册中心”“”
def __init__(self):
self.tools: Dict[str, ToolSchema] = {}
self.categories: Dict[ToolCategory, List[str]] = {cat: [] for cat in ToolCategory}
def register(self,
name: str,
description: str,
category: ToolCategory,
parameters: Dict,
required: List[str],
dangerous: bool = False):
"""注册工具装饰器"""
def decorator(func: Callable):
tool = ToolSchema(
name=name,
description=description,
category=category,
parameters=parameters,
required=required,
handler=func,
dangerous=dangerous
)
self.tools[name] = tool
self.categories[category].append(name)
return func
return decorator
def get_tool(self, name: str) -> Optional[ToolSchema]:
"""获取工具"""
return self.tools.get(name)
def list_tools(self, category: Optional[ToolCategory] = None) -> List[ToolSchema]:
"""列出工具"""
if category:
return [self.tools[name] for name in self.categories[category]]
return list(self.tools.values())
def get_schemas(self) -> List[Dict]:
"""获取所有Schema（用于LLM）"""
return [tool.to_openai_format() for tool in self.tools.values()]
def execute(self, name: str, arguments: Dict) -> Dict:
"""执行工具"""
tool = self.tools.get(name)
if not tool:
return {
"error": f"工具未找到: {name}",
"status": "failed"
}
# 参数验证
validation = self._validate_params(tool, arguments)
if not validation["valid"]:
return {
"error": validation["error"],
"status": "failed"
}
# 危险操作确认
if tool.dangerous:
# 记录审计日志
print(f"[龍魂·审计] 危险操作: {name}({arguments})")
try:
# 执行
result = tool.handler(**arguments)
return {
"result": result,
"status": "success",
"tool": name,
"dna": tool.dna_signature
}
except Exception as e:
return {
"error": str(e),
"status": "failed",
"tool": name
}
def _validate_params(self, tool: ToolSchema, arguments: Dict) -> Dict:
"""参数验证"""
# 检查必填参数
for req in tool.required:
if req not in arguments:
return {
"valid": False,
"error": f"缺少必填参数: {req}"
}
# 类型检查（简化版）
for key, value in arguments.items():
if key in tool.parameters:
expected_type = tool.parameters[key].get("type", "string")
# 简化类型检查
if expected_type == "number" and not isinstance(value, (int, float)):
return {
"valid": False,
"error": f"参数 {key} 应为数字类型"
}
return {"valid": True}
=== 全局注册中心 ===
registry = ToolRegistry()
=== 工具定义 ===
@registry.register(
name=“calculate_math”,
description=“执行数学计算，支持基本运算和函数”,
category=ToolCategory.CALCULATION,
parameters={
“expression”: {
“type”: “string”,
“description”: “数学表达式，如 ‘2 + 2 * 3’ 或 ‘sin(30)’”
}
},
required=[“expression”]
)
def calculate_math(expression: str) -> str:
“”“数学计算工具”“”
import math
import ast
# 安全评估：只允许数学运算
allowed_names = {
"abs": abs, "max": max, "min": min,
"pow": pow, "sqrt": math.sqrt,
"sin": math.sin, "cos": math.cos, "tan": math.tan,
"log": math.log, "log10": math.log10,
"pi": math.pi, "e": math.e
}
try:
# 解析表达式
tree = ast.parse(expression, mode='eval')
# 安全检查
for node in ast.walk(tree):
if isinstance(node, ast.Call):
if not isinstance(node.func, ast.Name) or node.func.id not in allowed_names:
raise ValueError(f"不允许的函数调用: {node.func}")
result = eval(compile(tree, '', 'eval'), {"__builtins__": {}}, allowed_names)
return f"计算结果: {result}"
except Exception as e:
return f"计算错误: {str(e)}"
@registry.register(
name=“solve_equation”,
description=“求解方程，支持一元二次方程等”,
category=ToolCategory.CALCULATION,
parameters={
“equation”: {
“type”: “string”,
“description”: “方程表达式，如 ‘x^2 + 5x + 6 = 0’”
},
“variable”: {
“type”: “string”,
“description”: “变量名，如 ‘x’”
}
},
required=[“equation”, “variable”]
)
def solve_equation(equation: str, variable: str) -> str:
“”“方程求解工具”“”
try:
from sympy import symbols, solve, Eq, parse_expr
var = symbols(variable)
# 解析方程
if '=' in equation:
left, right = equation.split('=')
eq = Eq(parse_expr(left), parse_expr(right))
else:
eq = parse_expr(equation)
solutions = solve(eq, var)
if not solutions:
return "方程无解"
result = f"方程 {equation} 的解:\n"
for i, sol in enumerate(solutions, 1):
result += f" x{i} = {sol}\n"
return result
except ImportError:
return "错误: 请安装sympy库"
except Exception as e:
return f"求解错误: {str(e)}"
@registry.register(
name=“search_knowledge”,
description=“搜索教育知识库中的相关内容”,
category=ToolCategory.SEARCH,
parameters={
“query”: {
“type”: “string”,
“description”: “搜索查询”
},
“top_k”: {
“type”: “number”,
“description”: “返回结果数量”,
“default”: 5
}
},
required=[“query”]
)
def search_knowledge(query: str, top_k: int = 5) -> str:
“”“知识搜索工具”“”
# 这里接入RAG服务
return f"搜索 ‘{query}’ 的结果（模拟）:\n1. 相关知识点A\n2. 相关知识点B\n3. 相关知识点C"
@registry.register(
name=“generate_quiz”,
description=“生成练习题”,
category=ToolCategory.DATA_ANALYSIS,
parameters={
“topic”: {
“type”: “string”,
“description”: “题目主题”
},
“difficulty”: {
“type”: “string”,
“description”: “难度: easy/medium/hard”,
“enum”: [“easy”, “medium”, “hard”]
},
“count”: {
“type”: “number”,
“description”: “题目数量”,
“default”: 5
}
},
required=[“topic”, “difficulty”]
)
def generate_quiz(topic: str, difficulty: str, count: int = 5) -> str:
“”“生成练习题”“”
questions = []
for i in range(1, count + 1):
questions.append(f"{i}. [{difficulty}] {topic} 相关练习题（模拟）")
return “\n”.join(questions)
@registry.register(
name=“execute_python”,
description=“执行Python代码（沙箱环境）”,
category=ToolCategory.CODE_EXECUTION,
parameters={
“code”: {
“type”: “string”,
“description”: “Python代码”
}
},
required=[“code”],
dangerous=True
)
def execute_python(code: str) -> str:
“”“Python代码执行（沙箱）”“”
# 实际应使用更严格的沙箱
import io
import sys
# 输出捕获
old_stdout = sys.stdout
sys.stdout = buffer = io.StringIO()
try:
# 安全限制：执行前检查
forbidden = ["import os", "import sys", "open(", "eval(", "exec("]
for f in forbidden:
if f in code:
return f"错误: 包含禁止的操作 '{f}'"
exec(code, {"__builtins__": {}}, {})
output = buffer.getvalue()
return output or "执行完成（无输出）"
except Exception as e:
return f"执行错误: {str(e)}"
finally:
sys.stdout = old_stdout
=== 使用示例 ===
if name == “main”:
# 列出所有工具
print(“=== 龍魂工具注册中心 ===”)
for tool in registry.list_tools():
print(f"[{tool.category.value}] {tool.name}: {tool.description}“)
print(f” 参数: {list(tool.parameters.keys())}“)
print(f” 签名: {tool.dna_signature}\n")
# 执行工具
print("=== 工具执行测试 ===")
# 数学计算
result = registry.execute("calculate_math", {"expression": "sqrt(16) + pow(2, 3)"})
print(f"calculate_math: {result}")
# 方程求解
result = registry.execute("solve_equation", {"equation": "x^2 + 5*x + 6 = 0", "variable": "x"})
print(f"\nsolve_equation: {result}")
# 生成练习题
result = registry.execute("generate_quiz", {"topic": "二次函数", "difficulty": "medium", "count": 3})
print(f"\ngenerate_quiz:\n{result}")
# 获取OpenAI格式Schema
print("\n=== OpenAI Function Schema ===")
schemas = registry.get_schemas()
print(json.dumps(schemas, indent=2, ensure_ascii=False))
六、FastAPI服务层
6.1 主服务（api/main.py）
Python
api/main.py
龍魂 · FastAPI主服务 · 教育AI全栈入口
import os
import sys
import time
import hashlib
from datetime import datetime
from typing import List, Dict, Optional
from fastapi import FastAPI, HTTPException, Depends, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field
import uvicorn
添加核心模块路径
sys.path.append(os.path.join(os.path.dirname(file), ‘…’, ‘core’))
from rag.rag_service import RAGService
from agent.education_agent import EducationAgent
from tools.tool_registry import registry
=== DNA常量 ===
MASTER_DNA = “ZHUGEXIN⚡️2025-🇨🇳🐉⚖️♠️🧚🏼♀️❤️♾️”
MASTER_UID = “9622”
CONFIRM_SEAL = “#CONFIRM🌌9622-ONLY-ONCE🧬LK9X-772Z”
=== 初始化服务 ===
rag_service = RAGService()
agent = EducationAgent(rag_service=rag_service)
app = FastAPI(
title=“龍魂教育AI系统”,
description=“RAG + Agent + Function Calling 全栈融合”,
version=“4.0.0”,
docs_url=“/docs”,
redoc_url=“/redoc”
)
CORS
app.add_middleware(
CORSMiddleware,
allow_origins=[““],
allow_credentials=True,
allow_methods=[””],
allow_headers=[“*”],
)
安全
security = HTTPBearer()
=== 数据模型 ===
class ChatRequest(BaseModel):
“”“聊天请求”“”
message: str = Field(…, description=“用户消息”)
session_id: Optional[str] = Field(None, description=“会话ID”)
context: Optional[Dict] = Field(None, description=“上下文信息”)
class ChatResponse(BaseModel):
“”“聊天响应”“”
response: str
intent: str
sources: List[Dict]
confidence: float
tools_used: List[str]
processing_time: float
dna_signature: str
class RAGQueryRequest(BaseModel):
“”“RAG查询请求”“”
query: str = Field(…, description=“查询问题”)
top_k: int = Field(5, description=“返回数量”)
filter: Optional[Dict] = Field(None, description=“过滤条件”)
class RAGQueryResponse(BaseModel):
“”“RAG查询响应”“”
answer: str
sources: List[Dict]
confidence: float
processing_time: float
class ToolExecuteRequest(BaseModel):
“”“工具执行请求”“”
tool_name: str
parameters: Dict
class ToolExecuteResponse(BaseModel):
“”“工具执行响应”“”
result: str
status: str
tool: str
class DocumentUploadRequest(BaseModel):
“”“文档上传请求”“”
file_path: str
metadata: Optional[Dict] = None
=== 中间件 ===
@app.middleware(“http”)
async def audit_log(request: Request, call_next):
“”“审计日志中间件”“”
start_time = time.time()
response = await call_next(request)
duration = time.time() - start_time
# 记录审计日志
log_entry = {
"timestamp": datetime.now().isoformat(),
"method": request.method,
"path": request.url.path,
"duration": round(duration, 3),
"status": response.status_code,
"client": request.client.host if request.client else "unknown"
}
# 实际应写入日志文件/数据库
print(f"[龍魂·审计] {log_entry}")
return response
=== API路由 ===
@app.get(“/”)
async def root():
“”“根路径”“”
return {
“name”: “龍魂教育AI系统”,
“version”: “4.0.0”,
“dna”: MASTER_DNA,
“uid”: MASTER_UID,
“status”: “running”,
“modules”: [“RAG”, “Agent”, “FunctionCalling”]
}
@app.get(“/health”)
async def health_check():
“”“健康检查”“”
return {
“status”: “healthy”,
“timestamp”: datetime.now().isoformat(),
“services”: {
“rag”: rag_service is not None,
“agent”: agent is not None,
“tools”: len(registry.list_tools())
}
}
@app.post(“/api/v1/chat”, response_model=ChatResponse)
async def chat(request: ChatRequest):
“”“智能对话”“”
start_time = time.time()
# Agent处理
result = agent.process(request.message)
# 提取工具使用
tools_used = []
if "tools" in result:
tools_used = result["tools"]
# 构建响应
processing_time = time.time() - start_time
return ChatResponse(
response=result["response"],
intent=result["intent"],
sources=[], # 从RAG获取
confidence=0.9, # 从Agent获取
tools_used=tools_used,
processing_time=processing_time,
dna_signature=f"SM3-{hashlib.sha256(request.message.encode()).hexdigest()[:16]}"
)
@app.post(“/api/v1/rag/query”, response_model=RAGQueryResponse)
async def rag_query(request: RAGQueryRequest):
“”“RAG检索”“”
start_time = time.time()
response = rag_service.query(request.query, request.filter)
processing_time = time.time() - start_time
return RAGQueryResponse(
answer=response.answer,
sources=[{
"content": s.content,
"source": s.source,
"page": s.page,
"score": s.score
} for s in response.sources],
confidence=response.confidence,
processing_time=processing_time
)
@app.post(“/api/v1/tools/execute”, response_model=ToolExecuteResponse)
async def execute_tool(request: ToolExecuteRequest):
“”“工具执行”“”
result = registry.execute(request.tool_name, request.parameters)
return ToolExecuteResponse(
result=str(result.get("result", result.get("error", "未知"))),
status=result.get("status", "unknown"),
tool=request.tool_name
)
@app.get(“/api/v1/tools/list”)
async def list_tools():
“”“列出工具”“”
return {
“tools”: [
{
“name”: t.name,
“description”: t.description,
“category”: t.category.value,
“parameters”: t.parameters,
“required”: t.required
}
for t in registry.list_tools()
]
}
@app.post(“/api/v1/documents/ingest”)
async def ingest_document(request: DocumentUploadRequest):
“”“导入文档”“”
try:
rag_service.ingest_documents([request.file_path])
return {
“status”: “success”,
“message”: f"文档导入完成: {request.file_path}“,
“dna”: f"SM3-{hashlib.sha256(request.file_path.encode()).hexdigest()[:16]}”
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get(“/api/v1/agent/memory”)
async def get_agent_memory():
“”“获取Agent记忆”“”
return agent.get_memory_snapshot()
@app.post(“/api/v1/agent/reset”)
async def reset_agent_memory():
“”“重置Agent记忆”“”
agent.memory.short_term = []
return {“status”: “success”, “message”: “记忆已重置”}
=== 启动 ===
if name == “main”:
uvicorn.run(
“main:app”,
host=“0.0.0.0”,
port=8000,
reload=True,
log_level=“info”
)
七、鸿蒙前端对接
7.1 教育AI客户端（entry/src/main/ets/pages/EducationAI.ets）
TypeScript
// entry/src/main/ets/pages/EducationAI.ets
// 龍魂 · 教育AI鸿蒙客户端 · 对接全栈后端
import { http } from ‘@kit.NetworkKit’;
const API_BASE = “http://localhost:8000/api/v1”;
const MASTER_UID = “9622”;
interface ChatMessage {
id: string;
role: ‘user’ | ‘assistant’;
content: string;
intent?: string;
sources?: SourceItem[];
confidence?: number;
toolsUsed?: string[];
timestamp: string;
}
interface SourceItem {
content: string;
source: string;
page: number;
score: number;
}
@Entry
@Component
struct EducationAIPage {
@State private messages: ChatMessage[] = [];
@State private inputText: string = ‘’;
@State private isLoading: boolean = false;
@State private sessionId: string = SESSION-${Date.now()};
@State private showSources: boolean = false;
@State private selectedMessage: ChatMessage | null = null;
aboutToAppear() {
// 欢迎消息
this.messages.push({
id: WELCOME-${Date.now()},
role: ‘assistant’,
content: ‘🐉 龍魂教育AI已就绪\n\n我可以帮你：\n• 查询知识点\n• 解答作业题目\n• 推荐学习课程\n• 生成练习题\n\n请直接输入你的问题！’,
timestamp: new Date().toISOString()
});
}
build() {
Column() {
this.HeaderBuilder()
this.ChatAreaBuilder()
this.InputAreaBuilder()
}
.width(‘100%’)
.height(‘100%’)
.backgroundColor(‘#0a0a0a’)
}
@Builder
HeaderBuilder() {
Row() {
Text(‘🐉’).fontSize(28).margin({ right: 8 })
Column() {
Text(‘龍魂教育AI’).fontSize(20).fontWeight(FontWeight.Bold).fontColor(‘#c41e3a’)
Text(UID:${MASTER_UID} · RAG+Agent+Tools).fontSize(12).fontColor(‘#666’)
}
.alignItems(HorizontalAlign.Start)
.layoutWeight(1)
Button('🛠️')
.fontSize(14)
.fontColor('#fff')
.backgroundColor('#333')
.borderRadius(8)
.onClick(() => {
this.showToolsDialog();
})
}
.width('100%')
.height(56)
.padding({ left: 16, right: 16 })
.backgroundColor('#1a1a1a')
.border({ width: { bottom: 1 }, color: '#333' })
}
@Builder
ChatAreaBuilder() {
List() {
ForEach(this.messages, (msg: ChatMessage) => {
ListItem() {
this.MessageBubbleBuilder(msg)
}
})
}
.width(‘100%’)
.layoutWeight(1)
.padding({ left: 12, right: 12, top: 8, bottom: 8 })
.scrollBar(BarState.Auto)
}
@Builder
MessageBubbleBuilder(msg: ChatMessage) {
Column() {
// 头像+名称
Row() {
if (msg.role === ‘assistant’) {
Text(‘🐉’).fontSize(20)
Text(‘龍魂AI’).fontSize(12).fontColor(‘#c41e3a’).margin({ left: 4 })
} else {
Text(‘👤’).fontSize(20)
Text(‘我’).fontSize(12).fontColor(‘#00ff00’).margin({ left: 4 })
}
}
.width(‘100%’)
.justifyContent(msg.role === ‘assistant’ ? FlexAlign.Start : FlexAlign.End)
.margin({ bottom: 4 })
// 内容气泡
Column() {
Text(msg.content)
.fontSize(14)
.fontColor('#fff')
.maxLines(100)
.textOverflow({ overflow: TextOverflow.Ellipsis })
// 元信息（AI消息）
if (msg.role === 'assistant' && msg.intent) {
Row() {
Text(`意图: ${msg.intent}`).fontSize(10).fontColor('#666')
if (msg.confidence) {
Text(`置信度: ${(msg.confidence * 100).toFixed(1)}%`)
.fontSize(10)
.fontColor(msg.confidence > 0.8 ? '#00ff00' : '#ffcc00')
.margin({ left: 8 })
}
}
.width('100%')
.margin({ top: 8 })
// 工具使用
if (msg.toolsUsed && msg.toolsUsed.length > 0) {
Row() {
Text('🔧').fontSize(10)
Text(msg.toolsUsed.join(', ')).fontSize(10).fontColor('#888').margin({ left: 4 })
}
.width('100%')
.margin({ top: 4 })
}
// 来源引用
if (msg.sources && msg.sources.length > 0) {
Button('查看来源').fontSize(11).fontColor('#c41e3a').backgroundColor('transparent').onClick(() => {
this.selectedMessage = msg;
this.showSources = true;
})
}
}
}
.width('85%')
.padding(12)
.backgroundColor(msg.role === 'assistant' ? '#1a1a1a' : '#0d2d0d')
.borderRadius(12)
.alignSelf(msg.role === 'assistant' ? ItemAlign.Start : ItemAlign.End)
.margin({ bottom: 8 })
}
.width('100%')
.alignItems(msg.role === 'assistant' ? HorizontalAlign.Start : HorizontalAlign.End)
}
@Builder
InputAreaBuilder() {
Row() {
TextInput({ placeholder: ‘输入问题…’, text: $$this.inputText })
.width(‘75%’)
.height(44)
.fontSize(14)
.fontColor(‘#fff’)
.placeholderColor(‘#666’)
.backgroundColor(‘#1a1a1a’)
.borderRadius(8)
.padding({ left: 12, right: 12 })
.onSubmit(() => {
this.sendMessage();
})
Button(this.isLoading ? '⏳' : '发送')
.width('20%')
.height(44)
.fontSize(14)
.fontColor('#fff')
.backgroundColor(this.isLoading ? '#333' : '#c41e3a')
.borderRadius(8)
.enabled(!this.isLoading)
.onClick(() => {
this.sendMessage();
})
}
.width('100%')
.height(60)
.padding({ left: 12, right: 12, top: 8, bottom: 8 })
.backgroundColor('#1a1a1a')
.border({ width: { top: 1 }, color: '#333' })
}
// === 发送消息 ===
async sendMessage() {
if (!this.inputText.trim() || this.isLoading) return;
const userMsg: ChatMessage = {
id: `USER-${Date.now()}`,
role: 'user',
content: this.inputText,
timestamp: new Date().toISOString()
};
this.messages.push(userMsg);
const question = this.inputText;
this.inputText = '';
this.isLoading = true;
try {
// 调用后端API
const response = await http.createHttp().request(`${API_BASE}/chat`, {
method: http.RequestMethod.POST,
header: { 'Content-Type': 'application/json' },
extraData: JSON.stringify({
message: question,
session_id: this.sessionId
})
});
const result = JSON.parse(response.result.toString());
const aiMsg: ChatMessage = {
id: `AI-${Date.now()}`,
role: 'assistant',
content: result.response,
intent: result.intent,
sources: result.sources,
confidence: result.confidence,
toolsUsed: result.tools_used,
timestamp: new Date().toISOString()
};
this.messages.push(aiMsg);
} catch (e) {
this.messages.push({
id: `ERROR-${Date.now()}`,
role: 'assistant',
content: `❌ 请求失败: ${e.message}`,
timestamp: new Date().toISOString()
});
} finally {
this.isLoading = false;
}
}
// === 工具对话框 ===
showToolsDialog() {
// 显示可用工具列表
// 实际实现：调用 /api/v1/tools/list
}
}
// 来源详情弹窗（简化）
@CustomDialog
struct SourcesDialog {
@Prop message: ChatMessage;
controller: CustomDialogController;
build() {
Column() {
Text(‘📚 引用来源’).fontSize(18).fontWeight(FontWeight.Bold).fontColor(‘#fff’).margin({ bottom: 16 })
if (this.message.sources) {
List() {
ForEach(this.message.sources, (source: SourceItem, index: number) => {
ListItem() {
Column() {
Text(`[${index + 1}] ${source.source} 第${source.page}页`)
.fontSize(12)
.fontColor('#c41e3a')
Text(source.content.substring(0, 200) + '...')
.fontSize(12)
.fontColor('#ccc')
.margin({ top: 4 })
Text(`相关度: ${(source.score * 100).toFixed(1)}%`)
.fontSize(11)
.fontColor('#888')
.margin({ top: 4 })
}
.width('100%')
.padding(12)
.backgroundColor('#1a1a1a')
.borderRadius(8)
.margin({ bottom: 8 })
}
})
}
.width('100%')
.layoutWeight(1)
}
Button('关闭')
.width('100%')
.height(44)
.fontColor('#fff')
.backgroundColor('#c41e3a')
.borderRadius(8)
.margin({ top: 16 })
.onClick(() => {
this.controller.close();
})
}
.width('90%')
.height('70%')
.padding(16)
.backgroundColor('#0a0a0a')
.borderRadius(16)
}
}
八、部署脚本
8.1 Docker Compose（docker-compose.yml）
yaml
docker-compose.yml
龍魂 · 教育AI全栈部署
version: ‘3.8’
services:
=== API服务 ===
longhun-api:
build: ./api
container_name: longhun-api
ports:
- “8000:8000”
environment:
- DNA=ZHUGEXIN⚡️2025-🇨🇳🐉⚖️♠️🧚🏼♀️❤️♾️
- UID=9622
- LLM_MODEL=deepseek-chat
- EMBEDDING_MODEL=BAAI/bge-large-zh
- VECTOR_DB=milvus
volumes:
- ./data:/app/data
- ./models:/app/models
depends_on:
- milvus
- postgres
networks:
- longhun-net
=== 向量数据库 ===
milvus:
image: milvusdb/milvus:v2.3.3
container_name: longhun-milvus
ports:
- “19530:19530”
- “9091:9091”
environment:
- ETCD_ENDPOINTS=etcd:2379
- MINIO_ADDRESS=minio:9000
volumes:
- milvus-data:/var/lib/milvus
networks:
- longhun-net
etcd:
image: quay.io/coreos/etcd:v3.5.5
container_name: longhun-etcd
environment:
- ETCD_AUTO_COMPACTION_MODE=revision
- ETCD_AUTO_COMPACTION_RETENTION=1000
networks:
- longhun-net
minio:
image: minio/minio:RELEASE.2023-03-20T20-16-18Z
container_name: longhun-minio
ports:
- “9000:9000”
- “9001:9001”
environment:
- MINIO_ACCESS_KEY=longhun9622
- MINIO_SECRET_KEY=ZHUGEXIN2025
command: server /data --console-address “:9001”
volumes:
- minio-data:/data
networks:
- longhun-net
=== 关系数据库 ===
postgres:
image: postgres:15
container_name: longhun-postgres
ports:
- “5432:5432”
environment:
- POSTGRES_USER=longhun
- POSTGRES_PASSWORD=ZHUGEXIN2025
- POSTGRES_DB=education
volumes:
- postgres-data:/var/lib/postgresql/data
networks:
- longhun-net
=== 监控 ===
prometheus:
image: prom/prometheus:v2.47.0
container_name: longhun-prometheus
ports:
- “9090:9090”
volumes:
- ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
- prometheus-data:/prometheus
networks:
- longhun-net
networks:
longhun-net:
driver: bridge
volumes:
milvus-data:
minio-data:
postgres-data:
prometheus-data:
九、组件清单
表格
文件 路径 说明
文档解析 core/rag/document_parser.py PDF/Word/Markdown/TXT/HTML解析
基于龍魂价值观对联/文言文/国学/历史/藏头诗/历史人物/人性知识/道德经/河图洛书/五行八卦/风水学/二十八星宿生成器-基于鸿蒙的AI对联创作应用开发实践
复制
分享
龍魂 · 鸿蒙AI国学创作应用：对联/文言文/藏头诗/历史人物/道德经/河图洛书/五行八卦/二十八星宿生成器
龍魂系统 · 鸿蒙原生国学创作层 · 传统文化数字化引擎
DNA: ZHUGEXIN⚡️2025-🇨🇳🐉⚖️♠️🧚🏼♀️❤️♾️ | UID: 9622 | CONFIRM: #CONFIRM🌌9622-ONLY-ONCE🧬LK9X-772Z
一、核心定位
表格
维度 说明
平台 鸿蒙 HarmonyOS NEXT · API 24+ · 纯血鸿蒙
语言 ArkTS · 声明式UI · 传统文化引擎
场景 对联创作 · 文言文生成 · 藏头诗 · 历史人物问答 · 道德经解读 · 河图洛书 · 五行八卦 · 二十八星宿 · 风水学
架构 龍魂国学引擎 → 规则模板 + AI生成 → 鸿蒙交互层
主权 数据本地 · 国密SM2/SM3签名 · 不上传云端
设计 模板优先 · 规则引擎 · 低算力 · 可复用 · 结构清晰
二、系统架构
plain
┌─────────────────────────────────────────┐
│ 龍魂系统 · 国学创作层 │
│ DNA: ZHUGEXIN⚡️2025-🇨🇳🐉⚖️♠️🧚🏼♀️❤️♾️ │
│ UID: 9622 │
├─────────────────────────────────────────┤
│ 鸿蒙 ArkTS 国学创作引擎 │
│ │
│ 创作流水线 │
│ ├── 对联生成器（平仄对仗引擎） │
│ ├── 文言文生成器（句式模板库） │
│ ├── 藏头诗生成器（首字约束引擎） │
│ ├── 历史人物问答（知识图谱检索） │
│ ├── 道德经解读（章节索引+释义） │
│ ├── 河图洛书（数字矩阵计算） │
│ ├── 五行八卦（生克推演引擎） │
│ ├── 二十八星宿（天文历法计算） │
│ ├── 风水学（方位吉凶推演） │
│ └── 国密签名验证（CryptoVerifier） │
├─────────────────────────────────────────┤
│ 龍魂国学数据层 │
│ │
│ 对联库 · 文言文模板 · 历史人物库 · 道德经全文 │
│ 河图洛书矩阵 · 五行生克表 · 八卦卦象 · 星宿表 │
│ 风水罗盘数据 · 国密加密存储 │
└─────────────────────────────────────────┘
三、国学创作引擎数据模型
3.1 核心数据模型（entry/src/main/ets/models/GuoxueModel.ets）
TypeScript
// entry/src/main/ets/models/GuoxueModel.ets
// 龍魂 · 国学创作数据模型
// === DNA常量 ===
const MASTER_DNA = “ZHUGEXIN⚡️2025-🇨🇳🐉⚖️♠️🧚🏼♀️❤️♾️”;
const MASTER_UID = “9622”;
const CONFIRM_SEAL = “#CONFIRM🌌9622-ONLY-ONCE🧬LK9X-772Z”;
// === 创作类型 ===
export enum CreationType {
COUPLET = ‘couplet’, // 对联
CLASSICAL_PROSE = ‘prose’, // 文言文
ACROSTIC_POEM = ‘acrostic’, // 藏头诗
HISTORICAL_FIGURE = ‘figure’, // 历史人物
DAO_DE_JING = ‘daodejing’, // 道德经
HE_TU = ‘hetu’, // 河图
LUO_SHU = ‘luoshu’, // 洛书
WU_XING = ‘wuxing’, // 五行
BA_GUA = ‘bagua’, // 八卦
XING_XIU = ‘xingxiu’, // 二十八星宿
FENG_SHUI = ‘fengshui’ // 风水
}
// === 对联数据 ===
export class CoupletData {
id: string;
upper: string; // 上联
lower: string; // 下联
horizontal: string; // 横批
theme: string; // 主题
tonePattern: string; // 平仄格式
wordCount: number; // 字数
dnaSignature: string;
constructor(data: Partial = {}) {
this.id = data.id || COUPLET-${Date.now()};
this.upper = data.upper || ‘’;
this.lower = data.lower || ‘’;
this.horizontal = data.horizontal || ‘’;
this.theme = data.theme || ‘’;
this.tonePattern = data.tonePattern || ‘’;
this.wordCount = data.wordCount || 7;
this.dnaSignature = this.signData();
}
private signData(): string {
const payload = ${this.id}-${this.upper}-${this.lower}-${Date.now()};
return SM3-${this.hashCode(payload)};
}
private hashCode(str: string): string {
let hash = 0;
for (let i = 0; i < str.length; i++) {
hash = ((hash << 5) - hash) + str.charCodeAt(i);
hash = hash & hash;
}
return Math.abs(hash).toString(16).substring(0, 16);
}
}
// === 文言文数据 ===
export class ClassicalProseData {
id: string;
title: string;
content: string;
style: string; // 风格：骈文/散文/赋
era: string; // 朝代
theme: string;
dnaSignature: string;
constructor(data: Partial = {}) {
this.id = data.id || PROSE-${Date.now()};
this.title = data.title || ‘’;
this.content = data.content || ‘’;
this.style = data.style || ‘散文’;
this.era = data.era || ‘先秦’;
this.theme = data.theme || ‘’;
this.dnaSignature = this.signData();
}
private signData(): string {
const payload = ${this.id}-${this.title}-${Date.now()};
return SM3-${this.hashCode(payload)};
}
private hashCode(str: string): string {
let hash = 0;
for (let i = 0; i < str.length; i++) {
hash = ((hash << 5) - hash) + str.charCodeAt(i);
hash = hash & hash;
}
return Math.abs(hash).toString(16).substring(0, 16);
}
}
// === 藏头诗数据 ===
export class AcrosticPoemData {
id: string;
headWord: string; // 藏头字
lines: string[]; // 诗句
theme: string;
style: string; // 五言/七言
dnaSignature: string;
constructor(data: Partial = {}) {
this.id = data.id || ACROSTIC-${Date.now()};
this.headWord = data.headWord || ‘’;
this.lines = data.lines || [];
this.theme = data.theme || ‘’;
this.style = data.style || ‘七言’;
this.dnaSignature = this.signData();
}
private signData(): string {
const payload = ${this.id}-${this.headWord}-${this.lines.join('')}-${Date.now()};
return SM3-${this.hashCode(payload)};
}
private hashCode(str: string): string {
let hash = 0;
for (let i = 0; i < str.length; i++) {
hash = ((hash << 5) - hash) + str.charCodeAt(i);
hash = hash & hash;
}
return Math.abs(hash).toString(16).substring(0, 16);
}
}
// === 历史人物数据 ===
export class HistoricalFigure {
id: string;
name: string;
dynasty: string;
era: string;
achievements: string[];
personality: string;
famousQuotes: string[];
relatedEvents: string[];
dnaSignature: string;
constructor(data: Partial = {}) {
this.id = data.id || FIGURE-${Date.now()};
this.name = data.name || ‘’;
this.dynasty = data.dynasty || ‘’;
this.era = data.era || ‘’;
this.achievements = data.achievements || [];
this.personality = data.personality || ‘’;
this.famousQuotes = data.famousQuotes || [];
this.relatedEvents = data.relatedEvents || [];
this.dnaSignature = this.signData();
}
private signData(): string {
const payload = ${this.id}-${this.name}-${this.dynasty}-${Date.now()};
return SM3-${this.hashCode(payload)};
}
private hashCode(str: string): string {
let hash = 0;
for (let i = 0; i < str.length; i++) {
hash = ((hash << 5) - hash) + str.charCodeAt(i);
hash = hash & hash;
}
return Math.abs(hash).toString(16).substring(0, 16);
}
}
// === 道德经章节 ===
export class DaoDeJingChapter {
id: string;
chapter: number;
title: string;
original: string;
translation: string;
interpretation: string;
keywords: string[];
dnaSignature: string;
constructor(data: Partial = {}) {
this.id = data.id || DAO-${Date.now()};
this.chapter = data.chapter || 1;
this.title = data.title || ‘’;
this.original = data.original || ‘’;
this.translation = data.translation || ‘’;
this.interpretation = data.interpretation || ‘’;
this.keywords = data.keywords || [];
this.dnaSignature = this.signData();
}
private signData(): string {
const payload = ${this.id}-${this.chapter}-${this.title}-${Date.now()};
return SM3-${this.hashCode(payload)};
}
private hashCode(str: string): string {
let hash = 0;
for (let i = 0; i < str.length; i++) {
hash = ((hash << 5) - hash) + str.charCodeAt(i);
hash = hash & hash;
}
return Math.abs(hash).toString(16).substring(0, 16);
}
}
// === 五行数据 ===
export class WuXingData {
id: string;
element: string; // 金木水火土
generation: string; // 生
restriction: string; // 克
color: string;
direction: string;
season: string;
organ: string;
emotion: string;
dnaSignature: string;
constructor(data: Partial = {}) {
this.id = data.id || WUXING-${Date.now()};
this.element = data.element || ‘’;
this.generation = data.generation || ‘’;
this.restriction = data.restriction || ‘’;
this.color = data.color || ‘’;
this.direction = data.direction || ‘’;
this.season = data.season || ‘’;
this.organ = data.organ || ‘’;
this.emotion = data.emotion || ‘’;
this.dnaSignature = this.signData();
}
private signData(): string {
const payload = ${this.id}-${this.element}-${Date.now()};
return SM3-${this.hashCode(payload)};
}
private hashCode(str: string): string {
let hash = 0;
for (let i = 0; i < str.length; i++) {
hash = ((hash << 5) - hash) + str.charCodeAt(i);
hash = hash & hash;
}
return Math.abs(hash).toString(16).substring(0, 16);
}
}
// === 八卦数据 ===
export class BaGuaData {
id: string;
name: string; // 乾/坤/震/巽/坎/离/艮/兑
symbol: string; // ☰/☷/☳/☴/☵/☲/☶/☱
nature: string; // 天/地/雷/风/水/火/山/泽
attribute: string; // 健/顺/动/入/陷/丽/止/悦
family: string; // 父/母/长男/长女/中男/中女/少男/少女
direction: string;
number: number;
dnaSignature: string;
constructor(data: Partial = {}) {
this.id = data.id || BAGUA-${Date.now()};
this.name = data.name || ‘’;
this.symbol = data.symbol || ‘’;
this.nature = data.nature || ‘’;
this.attribute = data.attribute || ‘’;
this.family = data.family || ‘’;
this.direction = data.direction || ‘’;
this.number = data.number || 1;
this.dnaSignature = this.signData();
}
private signData(): string {
const payload = ${this.id}-${this.name}-${this.symbol}-${Date.now()};
return SM3-${this.hashCode(payload)};
}
private hashCode(str: string): string {
let hash = 0;
for (let i = 0; i < str.length; i++) {
hash = ((hash << 5) - hash) + str.charCodeAt(i);
hash = hash & hash;
}
return Math.abs(hash).toString(16).substring(0, 16);
}
}
// === 二十八星宿数据 ===
export class XingXiuData {
id: string;
name: string; // 角/亢/氐/房/心/尾/箕…
direction: string; // 东方青龙/北方玄武/西方白虎/南方朱雀
animal: string; // 苍龙/玄武/白虎/朱雀
constellation: string; // 对应星座
auspicious: boolean; // 吉凶
meaning: string; // 含义
dnaSignature: string;
constructor(data: Partial = {}) {
this.id = data.id || XINGXIU-${Date.now()};
this.name = data.name || ‘’;
this.direction = data.direction || ‘’;
this.animal = data.animal || ‘’;
this.constellation = data.constellation || ‘’;
this.auspicious = data.auspicious ?? true;
this.meaning = data.meaning || ‘’;
this.dnaSignature = this.signData();
}
private signData(): string {
const payload = ${this.id}-${this.name}-${this.direction}-${Date.now()};
return SM3-${this.hashCode(payload)};
}
private hashCode(str: string): string {
let hash = 0;
for (let i = 0; i < str.length; i++) {
hash = ((hash << 5) - hash) + str.charCodeAt(i);
hash = hash & hash;
}
return Math.abs(hash).toString(16).substring(0, 16);
}
}
// === 创作状态 ===
@Observed
export class GuoxueState {
// 各类型创作记录
couplets: CoupletData[] = [];
proseList: ClassicalProseData[] = [];
acrostics: AcrosticPoemData[] = [];
figures: HistoricalFigure[] = [];
daoChapters: DaoDeJingChapter[] = [];
wuXingList: WuXingData[] = [];
baGuaList: BaGuaData[] = [];
xingXiuList: XingXiuData[] = [];
// 当前选中
currentType: CreationType = CreationType.COUPLET;
currentResult: string = ‘’;
// 统计
get statistics(): GuoxueStats {
return {
totalCouplets: this.couplets.length,
totalProse: this.proseList.length,
totalAcrostics: this.acrostics.length,
totalFigures: this.figures.length,
totalDaoChapters: this.daoChapters.length,
totalWuXing: this.wuXingList.length,
totalBaGua: this.baGuaList.length,
totalXingXiu: this.xingXiuList.length
};
}
persist(): void {
AppStorage.setOrCreate(‘guoxue_couplets’, JSON.stringify(this.couplets.slice(-50)));
AppStorage.setOrCreate(‘guoxue_prose’, JSON.stringify(this.proseList.slice(-50)));
AppStorage.setOrCreate(‘guoxue_acrostics’, JSON.stringify(this.acrostics.slice(-50)));
}
}
export interface GuoxueStats {
totalCouplets: number;
totalProse: number;
totalAcrostics: number;
totalFigures: number;
totalDaoChapters: number;
totalWuXing: number;
totalBaGua: number;
totalXingXiu: number;
}
export const guoxueState = new GuoxueState();
AppStorage.setOrCreate(‘guoxueState’, guoxueState);
四、国学创作引擎
4.1 对联生成器（entry/src/main/ets/engines/CoupletEngine.ets）
TypeScript
// entry/src/main/ets/engines/CoupletEngine.ets
// 龍魂 · 对联生成引擎 · 平仄对仗规则
import { CoupletData } from ‘…/models/GuoxueModel’;
export class CoupletEngine {
// 平仄映射（简化版）
private pingZeMap: Map<string, string> = new Map([
// 平声
[‘天’, ‘平’], [‘地’, ‘仄’], [‘山’, ‘平’], [‘水’, ‘仄’],
[‘风’, ‘平’], [‘雨’, ‘仄’], [‘春’, ‘平’], [‘秋’, ‘仄’],
[‘花’, ‘平’], [‘月’, ‘仄’], [‘日’, ‘仄’], [‘云’, ‘平’],
[‘龙’, ‘平’], [‘虎’, ‘仄’], [‘人’, ‘平’], [‘国’, ‘仄’],
[‘福’, ‘平’], [‘寿’, ‘仄’], [‘喜’, ‘仄’], [‘财’, ‘平’],
[‘红’, ‘平’], [‘绿’, ‘仄’], [‘白’, ‘仄’], [‘黑’, ‘平’],
[‘大’, ‘仄’], [‘小’, ‘仄’], [‘高’, ‘平’], [‘低’, ‘平’],
[‘长’, ‘平’], [‘短’, ‘仄’], [‘宽’, ‘平’], [‘窄’, ‘仄’],
[‘新’, ‘平’], [‘旧’, ‘仄’], [‘老’, ‘仄’], [‘少’, ‘仄’],
[‘来’, ‘平’], [‘去’, ‘仄’], [‘开’, ‘平’], [‘关’, ‘平’],
[‘上’, ‘仄’], [‘下’, ‘仄’], [‘左’, ‘仄’], [‘右’, ‘仄’],
[‘东’, ‘平’], [‘西’, ‘平’], [‘南’, ‘平’], [‘北’, ‘仄’],
[‘金’, ‘平’], [‘木’, ‘仄’], [‘水’, ‘仄’], [‘火’, ‘仄’], [‘土’, ‘仄’],
[‘仁’, ‘平’], [‘义’, ‘仄’], [‘礼’, ‘仄’], [‘智’, ‘仄’], [‘信’, ‘仄’]
]);
// 对仗词库
private antithesisMap: Map<string, string> = new Map([
[‘天’, ‘地’], [‘地’, ‘天’], [‘山’, ‘水’], [‘水’, ‘山’],
[‘风’, ‘雨’], [‘雨’, ‘风’], [‘春’, ‘秋’], [‘秋’, ‘春’],
[‘花’, ‘月’], [‘月’, ‘花’], [‘日’, ‘月’], [‘月’, ‘日’],
[‘龙’, ‘虎’], [‘虎’, ‘龙’], [‘人’, ‘我’], [‘我’, ‘人’],
[‘福’, ‘寿’], [‘寿’, ‘福’], [‘喜’, ‘乐’], [‘乐’, ‘喜’],
[‘来’, ‘去’], [‘去’, ‘来’], [‘开’, ‘合’], [‘合’, ‘开’],
[‘新’, ‘旧’], [‘旧’, ‘新’], [‘大’, ‘小’], [‘小’, ‘大’],
[‘高’, ‘低’], [‘低’, ‘高’], [‘长’, ‘短’], [‘短’, ‘长’],
[‘东’, ‘西’], [‘西’, ‘东’], [‘南’, ‘北’], [‘北’, ‘南’],
[‘金’, ‘木’], [‘木’, ‘金’], [‘水’, ‘火’], [‘火’, ‘水’],
[‘仁’, ‘义’], [‘义’, ‘仁’], [‘礼’, ‘智’], [‘智’, ‘礼’]
]);
// 主题模板库
private themeTemplates: Map<string, string[][]> = new Map([
[‘春节’, [
[‘春风送暖入屠苏’, ‘爆竹声中一岁除’],
[‘天增岁月人增寿’, ‘春满乾坤福满门’],
[‘一元复始呈兴旺’, ‘万象更新谱华章’]
]],
[‘龙年’, [
[‘龙腾虎跃闹新春’, ‘凤舞鸾歌庆丰年’],
[‘龙行龘龘前程朤朤’, ‘凤鸣喈喈后路迢迢’],
[‘金龙献瑞千家乐’, ‘玉兔呈祥万户春’]
]],
[‘通用’, [
[‘忠厚传家久’, ‘诗书继世长’],
[‘海纳百川有容乃大’, ‘壁立千仞无欲则刚’],
[‘书山有路勤为径’, ‘学海无涯苦作舟’]
]],
[‘励志’, [
[‘宝剑锋从磨砺出’, ‘梅花香自苦寒来’],
[‘业精于勤荒于嬉’, ‘行成于思毁于随’],
[‘长风破浪会有时’, ‘直挂云帆济沧海’]
]],
[‘爱国’, [
[‘苟利国家生死以’, ‘岂因祸福避趋之’],
[‘人生自古谁无死’, ‘留取丹心照汗青’],
[‘天下兴亡匹夫有责’, ‘国家危难壮士当先’]
]]
]);
// 生成对联
generate(theme: string, wordCount: number = 7): CoupletData {
const templates = this.themeTemplates.get(theme) || this.themeTemplates.get(‘通用’)!;
// 随机选择模板
const template = templates[Math.floor(Math.random() * templates.length)];
let upper: string;
let lower: string;
if (wordCount === 5) {
// 五言
upper = this.extractFiveWords(template[0]);
lower = this.extractFiveWords(template[1]);
} else if (wordCount === 7) {
// 七言
upper = template[0];
lower = template[1];
} else {
// 默认
upper = template[0];
lower = template[1];
}
// 生成横批
const horizontal = this.generateHorizontal(theme);
// 验证平仄
const tonePattern = this.analyzeTone(upper, lower);
return new CoupletData({
upper,
lower,
horizontal,
theme,
tonePattern,
wordCount
});
}
// 提取五言
private extractFiveWords(sentence: string): string {
// 简化处理：取前5个字或后5个字
if (sentence.length <= 5) return sentence;
return sentence.substring(0, 5);
}
// 生成横批
private generateHorizontal(theme: string): string {
const horizontals: Map<string, string[]> = new Map([
[‘春节’, [‘万象更新’, ‘春回大地’, ‘福满人间’]],
[‘龙年’, [‘龙年大吉’, ‘龙腾盛世’, ‘龙凤呈祥’]],
[‘通用’, [‘厚德载物’, ‘自强不息’, ‘宁静致远’]],
[‘励志’, [‘志存高远’, ‘鹏程万里’, ‘奋发图强’]],
[‘爱国’, [‘精忠报国’, ‘振兴中华’, ‘国泰民安’]]
]);
const options = horizontals.get(theme) || horizontals.get('通用')!;
return options[Math.floor(Math.random() * options.length)];
}
// 分析平仄
analyzeTone(upper: string, lower: string): string {
let upperTone = ‘’;
let lowerTone = ‘’;
for (let i = 0; i < upper.length; i++) {
const char = upper[i];
const tone = this.pingZeMap.get(char) || '中';
upperTone += tone === '平' ? '平' : (tone === '仄' ? '仄' : '中');
}
for (let i = 0; i < lower.length; i++) {
const char = lower[i];
const tone = this.pingZeMap.get(char) || '中';
lowerTone += tone === '平' ? '平' : (tone === '仄' ? '仄' : '中');
}
return `上联: ${upperTone} | 下联: ${lowerTone}`;
}
// 验证对仗
validateAntithesis(upper: string, lower: string): boolean {
if (upper.length !== lower.length) return false;
for (let i = 0; i < upper.length; i++) {
const u = upper[i];
const l = lower[i];
const expected = this.antithesisMap.get(u);
if (expected && expected !== l) {
// 允许部分不对仗（宽对）
continue;
}
}
return true;
}
// 获取所有主题
getThemes(): string[] {
return Array.from(this.themeTemplates.keys());
}
}
export const coupletEngine = new CoupletEngine();
4.2 藏头诗生成器（entry/src/main/ets/engines/AcrosticEngine.ets）
TypeScript
// entry/src/main/ets/engines/AcrosticEngine.ets
// 龍魂 · 藏头诗生成引擎
import { AcrosticPoemData } from ‘…/models/GuoxueModel’;
export class Acrostic
本月额度已用完
作为“人工智能6S店”的官方数字引擎，为AI开发者与企业提供一个覆盖软硬件全栈、一站式门户。
更多推荐
0
0- 0
已为社区贡献42条内容
所有评论(0)