基于LLM与自定义规则的实体抽取：自然语言API数据库查询方案问询

核心工作流程 #

Agent需要严格按照以下逻辑执行：

• 先调用API获取数据库所有可用列名
• 逐个选取列名，调用API获取该列的所有唯一值
• 结合自然语言查询语句，匹配列名、对应运算符及该列的合法值，生成<列名, 运算符, 值>三元组
• 若当前列匹配失败，切换至下一列重复上述步骤，直到遍历完所有列

基于LangChain的具体实现 #

1. 封装API调用工具 #

先写两个工具函数，用来对接获取列名和列唯一值的API：

def fetch_all_db_columns():
    # 调用内部API返回数据库列名列表，示例返回：["gender", "age", "event name", "event timestamp"]
    return call_internal_api(endpoint="/db/columns")

def fetch_column_unique_vals(column_name):
    # 传入列名，调用API返回该列的所有唯一值，比如传入"gender"返回["male", "female"]
    return call_internal_api(endpoint="/db/unique-values", params={"column": column_name})

2. 配置LangChain Agent #

使用StructuredChatAgent搭配AgentExecutor，让Agent能自主调用工具并完成推理：

from langchain.agents import AgentExecutor, StructuredChatAgent
from langchain.tools import StructuredTool
from langchain.prompts import ChatPromptTemplate
from langchain.chat_models import ChatOpenAI

# 注册工具到LangChain
tools = [
    StructuredTool.from_function(
        func=fetch_all_db_columns,
        name="FetchAllDBColumns",
        description="获取数据库所有可用列名，不知道有哪些列时必须调用这个工具"
    ),
    StructuredTool.from_function(
        func=fetch_column_unique_vals,
        name="FetchColumnUniqueValues",
        description="传入列名，获取该列的所有合法唯一值，用来验证查询中的值是否属于该列"
    )
]

# 定义提示词，明确Agent的工作规则
prompt_template = ChatPromptTemplate.from_messages([
    ("system", """你要从用户的自然语言查询里提取符合要求的实体三元组，格式为<列名, 运算符, 值>。
必须严格按以下步骤来：
1. 先调用FetchAllDBColumns获取所有列名
2. 对每个列名，调用FetchColumnUniqueValues拿到它的唯一值集合
3. 对照用户的查询语句，判断该列是否匹配：
   - 列名要和查询里的描述对应（比如“年龄”对应列名“age”）
   - 查询里提到的值必须是该列的唯一值之一
   - 根据查询表述推断运算符：比如“超过20岁”用“greater than”，“是男性”用“equals”，“最近30天”用“more than”
4. 匹配成功就生成对应的三元组，失败就继续检查下一列
5. 最后把所有匹配到的三元组分行返回"""),
    ("user", "{input}"),
    ("assistant", "{agent_scratchpad}")
])

# 初始化Agent并运行
llm = ChatOpenAI(temperature=0)
agent = StructuredChatAgent.from_llm_and_tools(llm=llm, tools=tools, prompt=prompt_template)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 示例调用
sample_query = "How many more than 20 years old male users viewed a page or logged in in the last 30 days?"
output = agent_executor.run(sample_query)
print(output)

3. 关键规则说明 #

• 运算符映射：提前在提示词里明确常见表述对应的运算符，避免LLM出错
• 多值处理：当查询出现“或”逻辑（比如“查看页面或登录”），要为同一列生成多个三元组

非LangChain替代实现 #

如果不想用LangChain，直接用LLM的函数调用功能也能实现：

import openai

openai.api_key = "你的API密钥"

def extract_entities(query):
    # 第一步：获取所有列名
    columns = fetch_all_db_columns()
    
    # 第二步：让LLM判断需要检查哪些列的唯一值
    messages = [
        {"role": "system", f"可用数据库列名：{columns}，请根据用户查询确定需要获取哪些列的唯一值"},
        {"role": "user", query}
    ]
    resp = openai.ChatCompletion.create(
        model="gpt-3.5-turbo-0613",
        messages=messages,
        functions=[
            {
                "name": "fetch_column_unique_vals",
                "parameters": {
                    "type": "object",
                    "properties": {"column_name": {"type": "string"}}
                }
            }
        ],
        function_call="auto"
    )
    
    # 处理函数调用，获取列唯一值
    if resp.choices[0].finish_reason == "function_call":
        func_call = resp.choices[0].message.function_call
        if func_call.name == "fetch_column_unique_vals":
            col_name = func_call.arguments["column_name"]
            col_values = fetch_column_unique_vals(col_name)
            
            # 第三步：生成实体三元组
            messages.append(resp.choices[0].message)
            messages.append({"role": "function", "name": "fetch_column_unique_vals", "content": str(col_values)})
            final_resp = openai.ChatCompletion.create(
                model="gpt-3.5-turbo-0613",
                messages=messages + [{"role": "user", "请生成符合要求的实体三元组"}]
            )
            return final_resp.choices[0].message.content

# 示例调用
print(extract_entities("How many more than 20 years old male users viewed a page or logged in in the last 30 days?"))

内容的提问来源于stack exchange，提问作者theodre7