华为MetaERP 提供一个更详尽、更落地的Oracle EBS FA资产模块AI建模方案。这个方案不仅包含架构设计和代码实现，还深入讲解了AI大模型如何结合知识图谱进行意图理解、参数提取和SQL动态-育师

提供一个更详尽、更落地的Oracle EBS FA资产模块AI建模方案。这个方案不仅包含架构设计和代码实现，还深入讲解了AI大模型如何结合知识图谱进行意图理解、参数提取和SQL动态生成，并通过多个完整示例展示工作流程。

一、整体架构：四层智能体设计

┌─────────────────────────────────────────────────────┐ │ 用户自然语言输入 │ └──────────────────────────┬──────────────────────────┘ ▼ ┌─────────────────────────────────────────────────────┐ │ 1. 意图识别与参数提取层（LLM + 规则引擎） │ │ · 使用DeepSeek等大模型进行语义理解 │ │ · 结合正则提取结构化参数（资产ID、日期、账簿等） │ │ · 输出：意图标签 + 参数字典 │ └──────────────────────────┬──────────────────────────┘ ▼ ┌─────────────────────────────────────────────────────┐ │ 2. 领域知识图谱层（图数据库 / 内存字典） │ │ · 存储表、字段、关联关系、业务规则 │ │ · 支持多跳路径查找（如：资产→事务→SLA→GL） │ │ · 输出：查询所需的最优表路径 │ └──────────────────────────┬──────────────────────────┘ ▼ ┌─────────────────────────────────────────────────────┐ │ 3. SQL动态生成引擎（模板 + AST组装） │ │ · 基于意图选择基模板 │ │ · 根据参数和表路径动态添加JOIN、WHERE、GROUP BY │ │ · 输出：可执行的SQL语句 │ └──────────────────────────┬──────────────────────────┘ ▼ ┌─────────────────────────────────────────────────────┐ │ 4. 结果解释与反馈层 │ │ · 返回SQL + 自然语言解释 │ │ · 可选：执行SQL并返回表格 │ └─────────────────────────────────────────────────────┘

二、知识图谱构建（关键步骤）

2.1 表节点与字段节点

将Excel中的每个表抽象为一个节点，每个字段作为属性。同时建立业务概念节点（如“资产新增”、“折旧计提”），并关联到对应的表路径。

示例（使用JSON表示）：

{ "nodes": { "FA_TRANSACTION_HEADERS": { "label": "事务头表", "fields": { "TRANSACTION_HEADER_ID": { "type": "NUMBER", "desc": "事务唯一ID" }, "ASSET_ID": { "type": "NUMBER", "desc": "资产ID" }, "TRANSACTION_TYPE_CODE": { "type": "VARCHAR2", "desc": "事务类型代码" }, "TRANSACTION_DATE": { "type": "DATE", "desc": "事务日期" } } }, "FA_BOOKS": { "label": "账簿价值快照", "fields": { "TRANSACTION_HEADER_ID_IN": { "type": "NUMBER", "desc": "生效事务ID" }, "COST": { "type": "NUMBER", "desc": "当前原值" }, "NBV": { "type": "NUMBER", "desc": "账面净值" } } } }, "edges": [ { "from": "FA_TRANSACTION_HEADERS", "to": "FA_BOOKS", "on": "TRANSACTION_HEADER_ID = TRANSACTION_HEADER_ID_IN" } ], "business_concepts": { "资产新增": { "description": "查询新增资产的信息", "path": ["FA_ADDITIONS_B", "FA_TRANSACTION_HEADERS", "FA_BOOKS"], "conditions": { "FA_TRANSACTION_HEADERS.TRANSACTION_TYPE_CODE": "'ADDITION'" } }, "折旧明细": { "description": "查询折旧明细", "path": ["FA_DISTRIBUTION_HISTORY", "FA_DEPRN_DETAIL"], "default_fields": ["ASSET_ID", "PERIOD_NAME", "DEPRN_AMOUNT"] }, "SLA追溯": { "description": "从FA事务追溯到GL凭证", "path": ["FA_TRANSACTION_HEADERS", "XLA_EVENTS", "XLA_AE_HEADERS", "XLA_AE_LINES"], "conditions": { "XLA_EVENTS.APPLICATION_ID": 140 } } } }

2.2 业务规则编码

将常见业务逻辑（如“本月折旧 = 上月折旧 + 本期调整”）不直接硬编码，而是作为提示词注入LLM，辅助生成正确SQL。例如：

“当月折旧金额”对应FA_DEPRN_DETAIL.DEPRN_AMOUNT
“累计折旧”对应FA_DEPRN_DETAIL.DEPRN_RESERVE
“账面净值” =COST - ACCUMULATED_DEPRECIATION（在FA_BOOKS中有NBV字段）

三、意图识别与参数提取（LLM驱动）

3.1 提示词设计

我们使用一个结构化的Prompt，让LLM输出JSON格式的意图和参数。以下是一个示例（实际使用时可根据模型调整）：

你是一个Oracle EBS FA资产模块的专家。请根据用户的问题，提取以下信息并以JSON格式返回： - intent: 查询意图，可选值：query_addition, query_depreciation, query_retirement, trace_sla, query_transfer, query_revaluation, query_impairment, query_cip_capitalization, query_mass, general_query - params: 包含提取出的参数，如asset_id, start_date, end_date, period_name, book_type_code, transaction_header_id, category_id, location_id, retirement_type等。如果没有则为null。 用户问题：{user_input} 请只返回JSON，不要其他内容。

3.2 参数提取增强

对于日期、期间等模糊表述，我们在LLM输出后再用规则补充。例如：

“上月” → 计算当前月份的上一个月首尾日期。
“最近三个月” → 当前日期往前推90天。
“JUN-2025” → 保持原样，用于PERIOD_NAME字段。

3.3 多轮对话支持

如果需要多轮追问（如用户先问“查一下资产”，再问“看看它的折旧”），我们维护一个会话上下文，记录上一次提到的资产ID或事务ID。

四、SQL动态生成引擎（核心代码）

下面提供一个更完整的Python实现，包含：

知识图谱类（Graph）
LLM接口模拟（实际可替换为DeepSeek API）
SQL生成器（支持多表JOIN、条件拼接、排序）

import json import re from datetime import datetime, timedelta from typing import Dict, List, Optional # ---------- 1. 知识图谱类 ---------- class FAGraph: def __init__(self, schema_file: str = "fa_schema.json"): with open(schema_file, 'r', encoding='utf-8') as f: self.data = json.load(f) self.tables = self.data["tables"] self.edges = self.data["edges"] self.concepts = self.data.get("business_concepts", {}) def get_path_for_concept(self, concept: str) -> List[str]: """根据业务概念获取表路径""" if concept in self.concepts: return self.concepts[concept]["path"] return [] def get_joins_for_path(self, path: List[str]) -> List[str]: """根据路径生成JOIN条件""" joins = [] for i in range(len(path)-1): left_table = path[i] right_table = path[i+1] # 查找边 edge = next((e for e in self.edges if e["from"]==left_table and e["to"]==right_table), None) if edge: joins.append(f"LEFT JOIN {right_table} ON {left_table}.{edge['on'].split('=')[0].strip()} = {right_table}.{edge['on'].split('=')[1].strip()}") else: # 尝试反向查找 edge = next((e for e in self.edges if e["from"]==right_table and e["to"]==left_table), None) if edge: joins.append(f"LEFT JOIN {right_table} ON {left_table}.{edge['on'].split('=')[1].strip()} = {right_table}.{edge['on'].split('=')[0].strip()}") return joins def get_default_fields(self, concept: str) -> List[str]: return self.concepts.get(concept, {}).get("default_fields", ["*"]) # ---------- 2. LLM接口模拟（实际使用DeepSeek API） ---------- class LLMInterface: @staticmethod def analyze_query(user_input: str) -> Dict: """ 模拟LLM返回意图和参数。 实际应调用外部API。 """ # 这里为了演示，使用简单规则模拟 input_lower = user_input.lower() intent = "general_query" params = {} if any(kw in input_lower for kw in ["新增","添加","购买"]): intent = "query_addition" elif any(kw in input_lower for kw in ["折旧","计提"]): intent = "query_depreciation" elif any(kw in input_lower for kw in ["报废","处置","出售"]): intent = "query_retirement" elif any(kw in input_lower for kw in ["追溯","sla","凭证","总账"]): intent = "trace_sla" # 提取资产ID m = re.search(r'资产[Ii][Dd]\s*[:：]?\s*(\d+)', user_input) if m: params['asset_id'] = int(m.group(1)) # 提取事务头ID m = re.search(r'事务[头]?[Ii][Dd]\s*[:：]?\s*(\d+)', user_input) if m: params['transaction_header_id'] = int(m.group(1)) # 提取日期 m = re.search(r'(\d{4}-\d{2}-\d{2})\s*(?:到|至|~)\s*(\d{4}-\d{2}-\d{2})', user_input) if m: params['start_date'] = m.group(1) params['end_date'] = m.group(2) elif '上月' in input_lower: today = datetime.now() first = today.replace(day=1) last_month_end = first - timedelta(days=1) last_month_start = last_month_end.replace(day=1) params['start_date'] = last_month_start.strftime('%Y-%m-%d') params['end_date'] = last_month_end.strftime('%Y-%m-%d') # 提取期间 m = re.search(r'期间\s*[:：]?\s*([A-Za-z]+-\d{4})', user_input) if m: params['period_name'] = m.group(1) # 提取账簿 m = re.search(r'(主账簿|税务账簿|CIP)\s*[:：]?\s*(\w+)', user_input) if m: params['book_type_code'] = m.group(2) # 提取类别 m = re.search(r'类别\s*[:：]?\s*(\d+)', user_input) if m: params['category_id'] = int(m.group(1)) return {"intent": intent, "params": params} # ---------- 3. SQL生成器 ---------- class SQLGenerator: def __init__(self, graph: FAGraph): self.graph = graph self.base_templates = { "query_addition": { "select_from": "FA_ADDITIONS_B fa", "joins": ["JOIN FA_TRANSACTION_HEADERS fth ON fa.asset_id = fth.asset_id AND fth.transaction_type_code = 'ADDITION'", "JOIN FA_BOOKS fb ON fth.transaction_header_id = fb.transaction_header_id_in"], "where_clauses": [], "order_by": "fth.transaction_date DESC" }, "query_depreciation": { "select_from": "FA_DEPRN_DETAIL fdd", "joins": ["JOIN FA_DISTRIBUTION_HISTORY fdh ON fdd.distribution_id = fdh.distribution_id"], "where_clauses": [], "order_by": "fdd.period_counter" }, "query_retirement": { "select_from": "FA_RETIREMENTS fr", "joins": ["JOIN FA_TRANSACTION_HEADERS fth ON fr.transaction_header_id = fth.transaction_header_id"], "where_clauses": [], "order_by": "fr.date_retired DESC" }, "trace_sla": { "select_from": "FA_TRANSACTION_HEADERS fth", "joins": ["JOIN XLA_EVENTS xev ON fth.transaction_header_id = xev.source_id_int_1 AND xev.application_id = 140", "JOIN XLA_AE_HEADERS xah ON xev.event_id = xah.event_id", "JOIN XLA_AE_LINES xal ON xah.ae_header_id = xal.ae_header_id"], "where_clauses": [], "order_by": "fth.transaction_header_id, xal.ae_line_num" } } def generate(self, intent: str, params: Dict) -> str: if intent not in self.base_templates: return f"-- 未知意图: {intent}" tmpl = self.base_templates[intent].copy() # 动态添加WHERE条件 where_list = [] if 'asset_id' in params: # 根据意图决定哪个表的asset_id table_alias = "fa" if intent=="query_addition" else "fth" if intent=="query_depreciation": table_alias = "fdd" elif intent=="query_retirement": table_alias = "fr" elif intent=="trace_sla": table_alias = "fth" where_list.append(f"{table_alias}.asset_id = {params['asset_id']}") if 'transaction_header_id' in params: where_list.append(f"fth.transaction_header_id = {params['transaction_header_id']}") if 'start_date' in params and 'end_date' in params: # 根据意图选择日期字段 date_field = "fth.transaction_date" if intent=="query_retirement": date_field = "fr.date_retired" where_list.append(f"{date_field} >= TO_DATE('{params['start_date']}', 'YYYY-MM-DD')") where_list.append(f"{date_field} <= TO_DATE('{params['end_date']}', 'YYYY-MM-DD')") if 'period_name' in params: where_list.append(f"fdd.period_name = '{params['period_name']}'") if 'book_type_code' in params: where_list.append(f"fdd.book_type_code = '{params['book_type_code']}'") if 'category_id' in params: where_list.append(f"fa.category_id = {params['category_id']}") # 构建SQL select_part = "SELECT " # 默认字段，可根据需要扩展 default_fields = { "query_addition": "fa.asset_id, fa.segment1 AS asset_number, fa.description, fth.transaction_date, fb.cost, fb.nbv", "query_depreciation": "fdd.asset_id, fdd.period_name, fdd.deprn_amount, fdd.deprn_reserve, fdd.nbv", "query_retirement": "fr.asset_id, fr.date_retired, fr.cost_retired, fr.accum_deprn_retired, fr.nbv_retired, fr.sale_proceeds, fr.gain_loss_amount, fr.retirement_type", "trace_sla": "fth.transaction_header_id, fth.transaction_type_code, xev.event_type_code, xah.ae_header_id, xal.accounting_class_code, xal.code_combination_id, xal.accounted_dr, xal.accounted_cr, xal.gl_sl_link_id" } select_part += default_fields.get(intent, "*") from_part = f"FROM {tmpl['select_from']}" join_part = "\n".join(tmpl['joins']) where_part = " AND ".join(where_list) if where_list else "1=1" order_part = f"ORDER BY {tmpl['order_by']}" sql = f"{select_part}\n{from_part}\n{join_part}\nWHERE {where_part}\n{order_part}" return sql # ---------- 4. 主控制器 ---------- class FAAssistant: def __init__(self): self.graph = FAGraph("fa_schema.json") # 需要提前准备好fa_schema.json self.llm = LLMInterface() self.sql_gen = SQLGenerator(self.graph) def process(self, user_input: str) -> str: # Step1: 意图识别 analysis = self.llm.analyze_query(user_input) intent = analysis["intent"] params = analysis["params"] # Step2: 生成SQL sql = self.sql_gen.generate(intent, params) # Step3: 返回解释 explanation = f"""**用户问题**: {user_input} **识别意图**: {intent} **提取参数**: {json.dumps(params, ensure_ascii=False)} **生成的SQL**:

sql

**说明**: 该SQL将查询您所需的FA数据。如果您需要执行，请复制到数据库客户端运行。如需进一步过滤或调整，请告诉我。""" return explanation # ---------- 5. 测试 ---------- if __name__ == "__main__": assistant = FAAssistant() test_questions = [ "查询上月新增的所有资产", "查一下资产ID 500123的折旧明细，期间JUN-2025", "追溯资产ID 500123的报废信息", "追溯事务头ID 1002345的SLA凭证到总账", "查一下类别为203的新增资产" ] for q in test_questions: print("\n" + "="*70) print(assistant.process(q))

五、AI大模型工作结果示例（完整输出）

假设用户输入：“查一下资产ID 500123的折旧明细，期间JUN-2025”

模型内部处理：

LLM分析 → 输出JSON：

{"intent": "query_depreciation", "params": {"asset_id": 500123, "period_name": "JUN-2025"}}

知识图谱检索 → 找到概念“折旧明细”的表路径：FA_DISTRIBUTION_HISTORY→FA_DEPRN_DETAIL，并确认关联键为DISTRIBUTION_ID。

SQL生成 → 根据模板和参数生成：

SELECT fdd.asset_id, fdd.period_name, fdd.deprn_amount, fdd.deprn_reserve, fdd.nbv FROM FA_DEPRN_DETAIL fdd JOIN FA_DISTRIBUTION_HISTORY fdh ON fdd.distribution_id = fdh.distribution_id WHERE fdd.asset_id = 500123 AND fdd.period_name = 'JUN-2025' ORDER BY fdd.period_counter

返回结果（带解释）：

**用户问题**: 查一下资产ID 500123的折旧明细，期间JUN-2025 **识别意图**: query_depreciation **提取参数**: {"asset_id": 500123, "period_name": "JUN-2025"} **生成的SQL**:

sql

SELECT fdd.asset_id, fdd.period_name, fdd.deprn_amount, fdd.deprn_reserve, fdd.nbv

FROM FA_DEPRN_DETAIL fdd

JOIN FA_DISTRIBUTION_HISTORY fdh ON fdd.distribution_id = fdh.distribution_id

WHERE fdd.asset_id = 500123 AND fdd.period_name = 'JUN-2025'

ORDER BY fdd.period_counter

**说明**: 该SQL将查询资产500123在JUN-2025期间的折旧金额、累计折旧余额和账面净值。如果您想查看更详细的分配信息，可以添加`fdh.*`字段。

六、进阶功能与优化

6.1 多表复杂查询支持

对于用户问题如“查询上月所有报废的资产，并显示处置损益”，模型会自动组合query_retirement模板，并额外关联FA_BOOKS获取原值和累计折旧。这可以通过在知识图谱中定义复合概念来实现。

6.2 聚合与分组

用户可能问：“按部门统计本月折旧总额”。此时需要：

识别意图：aggregate_depreciation_by_dept
表路径：FA_DEPRN_DETAIL→FA_DISTRIBUTION_HISTORY→GL_CODE_COMBINATIONS（需额外关联科目段）
SQL生成：SELECT ... SUM(deprn_amount) GROUP BY segment1