3 小时搞定 5 万字段！LLM 自动生成数据字典实战指南

# 安装: pip install sqlalchemy psycopg2-binary (以PostgreSQL为例)
from sqlalchemy import create_engine, inspect

# 替换为你的数据库连接字符串
DATABASE_URI = "postgresql://user:password@host:port/dbname"
engine = create_engine(DATABASE_URI)
inspector = inspect(engine)

schema_info = {}
table_names = inspector.get_table_names() # 获取所有表名

for table_name in table_names:
    columns_data = []
    columns = inspector.get_columns(table_name) # 获取表的列信息
    pk_constraint = inspector.get_pk_constraint(table_name) # 获取主键
    pk_columns = pk_constraint.get('constrained_columns', []) if pk_constraint else []
    
    # 获取表注释 (不同数据库方言获取方式可能略有差异)
    table_comment = inspector.get_table_comment(table_name).get('text') if inspector.get_table_comment(table_name) else""

    for col in columns:
        columns_data.append({
            "name": col['name'],
            "type": str(col['type']),
            "nullable": col['nullable'],
            "default": col.get('default'),
            "comment": col.get('comment', ''), # 已有注释
            "is_primary_key": col['name'] in pk_columns
        })
    schema_info[table_name] = {"columns": columns_data, "table_comment": table_comment}

# schema_info 现在包含了所有表的元数据
# print(schema_info['your_table_name'])

数据库表名: {table_name}
已有表注释: {existing_table_comment}
包含以下字段:
{column_list_summary} 

请基于以上信息，用一段简洁的中文描述此表的主要业务用途和上下文。
如果已有表注释清晰，请优先参考。

数据库表名: {table_name}
表的主要用途: {table_description_from_llm_or_manual}

字段名: {column_name}
数据类型: {column_type}
是否可空: {is_nullable}
默认值: {default_value}
是否主键: {is_primary_key}
已有字段注释: {existing_column_comment}

请基于以上信息，提供以下内容 (使用中文，JSON格式输出):
1. **business_meaning**: 此字段在业务上下文中的清晰、简洁解释。
2. **data_characteristics**: 描述其典型格式 (如日期YYYY-MM-DD)，潜在约束或取值范围 (如果可从名称、类型、已有注释推断)。
3. **example_values**: 提供1-2个真实且有代表性的示例值 (如果能合理推断)。
4. **sensitivity_analysis**: 判断此字段是否可能包含敏感数据 (如PII, 财务数据)，如果是，请说明理由并给出敏感等级建议 (如: 高、中、低)。

请重点参考已有字段注释。如果字段名或表用途已足够清晰，请确保解释精准。

# 安装: pip install openai
import openai
import json # 用于解析JSON输出

# openai.api_key = "YOUR_OPENAI_API_KEY"# 推荐使用环境变量

def generate_with_gpt4(prompt, is_json_output=False):
    try:
        response = openai.chat.completions.create(
            model="gpt-4o", # 或者 "gpt-4-turbo", "gpt-3.5-turbo"
            messages=[
                {"role": "system", "content": "你是一位资深的数据库架构师，精通数据字典的编写和业务含义的解读。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.2, # 低温确保输出更稳定和一致
            max_tokens=800 # 根据需要调整
        )
        content = response.choices[0].message.content.strip()
        if is_json_output:
            # 尝试去除Markdown代码块标记 (如果存在)
            if content.startswith("```json"):
                content = content[7:]
            if content.endswith("```"):
                content = content[:-3]
            return json.loads(content)
        return content
    except Exception as e:
        print(f"调用LLM出错: {e}")
        if is_json_output:
            return {"error": str(e)}
        return f"生成失败: {e}"

# --- 编排逻辑 (伪代码) ---
# data_dictionary = {}
#for table_name, table_data in schema_info.items():
#     # 1. 生成表描述
#     column_list_for_table_prompt = "\n".join([f"- {c['name']} ({c['type']})"for c in table_data['columns'][:10]]) # 示例：仅用前10个字段
#     table_prompt_text = TABLE_DESCRIPTION_PROMPT_TEMPLATE.format(
#         table_name=table_name,
#         existing_table_comment=table_data['table_comment'],
#         column_list_summary=column_list_for_table_prompt
#     )
#     generated_table_description = generate_with_gpt4(table_prompt_text)
#     data_dictionary[table_name] = {"table_description": generated_table_description, "columns": []}
    
#     # 2. 遍历列，生成列描述
#     for col_info in table_data['columns']:
#         column_prompt_text = COLUMN_DESCRIPTION_PROMPT_TEMPLATE.format(
#             table_name=table_name,
#             table_description_from_llm_or_manual=generated_table_description, # 将生成的表描述作为上下文
#             column_name=col_info['name'],
#             column_type=col_info['type'],
#             is_nullable=col_info['nullable'],
#             default_value=col_info['default'],
#             is_primary_key=col_info['is_primary_key'],
#             existing_column_comment=col_info['comment']
#         )
#         generated_column_details = generate_with_gpt4(column_prompt_text, is_json_output=True)
#         # 合并原始元数据和LLM生成的信息
#         final_col_data = {**col_info, **generated_column_details}
#         data_dictionary[table_name]["columns"].append(final_col_data)

#print(json.dumps(data_dictionary, indent=2, ensure_ascii=False))

defgenerate_markdown_output(data_dictionary):
    markdown_string = ""
    for table_name, table_data in data_dictionary.items():
        markdown_string += f"## 表名: {table_name}\n\n"
        markdown_string += f"**表描述**: {table_data.get('table_description', 'N/A')}\n\n"
        markdown_string += "| 字段名 | 数据类型 | 是否可空 | 主键 | 默认值 | 已有注释 | 业务含义 (LLM) | 数据特征 (LLM) | 示例值 (LLM) | 敏感性分析 (LLM) |\n"
        markdown_string += "|---|---|---|---|---|---|---|---|---|---|\n"
        for col in table_data.get('columns', []):
            markdown_string += f"| {col.get('name','')} | {col.get('type','')} | {col.get('nullable','')} | {col.get('is_primary_key','')} | {col.get('default','') if col.get('default') isnotNoneelse''} | {col.get('comment','')} | {col.get('business_meaning','')} | {col.get('data_characteristics','')} | {str(col.get('example_values',''))} | {str(col.get('sensitivity_analysis',''))} |\n"
        markdown_string += "\n"
    return markdown_string

# md_output = generate_markdown_output(data_dictionary_after_review)
# with open("data_dictionary.md", "w", encoding="utf-8") as f:
# f.write(md_output)