从零开始写RAG - OneRAG （四）
"Role": ["你是一个多领域智能查询优化引擎，通过多步流程动态重构用户请求"], "Background": "用户弱信息查询", "Preferences": "强结构化输出与问题重述", "Goals": "实现认知增强：将普通查询升级为专家级问题框架", "Constrains": "Template的格式与要求", "Skills": "语义解构 | 领域适配引擎| 抗模糊处理 |
OneRAG系列 [求Star ⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐]
OneRAG， Github传送链接：https://github.com/Hlufies/OneRAG.git
目录
默默更新自己的小demo，向高阶demo出发！根据自己构想，更改了RAG框架。
Query
直接上代码先：
# app.py
query="西红柿炒蛋怎么做的？"
user_query, task_dict, final_dict = Query(user_query=query, config=config)
# Tools.Query
# 查询优化
def Query(user_query, config=None):
assert config is not None
config = config['query']
with open(os.path.join(current_dir,config["template_query"]), "r") as f:
query_template = json.load(f)
query_template["Initialization"] += user_query
user_dict = format_template(query_template)[1:-1]
task_dict = QUERY_MODEL_MAPPING[config["query_model"]](user_dict)
# 分析task流
task_graph = analyze_workflow(task_dict)
task_funcs = {k: TASK_FUNC_MAPPING[config["task_func"]] for k, _ in task_dict.items()}
task_llms = {k: TASK_LLM_MAPPING[config["task_llm"]] for k, _ in task_dict.items()}
# 解析分析结果
print_workflow_results(task_graph)
try:
final_dict = asyncio.run(
run(
user_query=user_query, # 用户查询
task_dict=task_dict, # 任务字典
task_graph=task_graph, # 任务图
task_funcs=task_funcs, # 任务方法
task_llms=task_llms, # 任务模型
template=config["template_workflow"]
)
)
except Exception as e:
print("Query error: ", e)
return user_query, task_dict, task_dict
save_dict("query_before.json", task_dict)
save_dict("query_after.json", final_dict)
return user_query, task_dict, final_dict在当前项目中，我们通常根据用户输入的原始查询（query）进行上下文检索，并将检索结果拼接为提示词（prompt）输入大模型生成回答。然而，我们发现用户提问往往存在意图模糊或表述不明确的问题，导致检索结果与用户真实需求存在偏差。为提升系统响应准确性，对用户查询进行语义重写是必要的优化方向。 根据搜索资料，重写技术（为什么需要查询重写的原因）能针对性解决以下问题：
-
简单场景：同义词替换、纠错
-
复杂意图：Step-Back抽象化、Query分解
-
低置信查询：HyDE生成假设文档引导检索
-
强化问题定位：将“模糊的”整合为 精准描述用户查询的典型问题。
-
点明核心矛盾：增加 “导致检索结果与用户真实需求存在偏差”，直接关联模糊查询与结果失效间的因果关系。
-
术语专业化：使用 “语义重写” 替代口语化的“重写”，呼应业界对查询优化的标准表述（如Query Rewriting）。
-
意图歧义：如用户问“苹果的生产地”，可能指水果或手机品牌，需通过实体消歧明确意图。
-
语义鸿沟：用户口语化表达（如“咋退款”）与知识库专业术语（“退货流程”）需通过同义替换对齐。
-
检索效率：重写可精简冗余词（如“请问你能告诉我…”→核心关键词），提升向量检索精度。
-
长尾查询：对模糊问法（如“那个新技术”）采用HyDE生成假设文档，扩充语义召回空间。
-
价值导向：通过 “提升系统响应准确性” 强调重写操作的业务价值，而非单纯描述动作本身。
技术细节
Template
首先，我讲用户查询改写定一个为任务分解问题，通过预先设置template的格式，来指导模型。其中Template有机个关键的地方：
1. 整体的架构采用一下模式，这里联合role play，背景，能力偏好，目标，约束，强化能力，few-shot，初始化等已经验证的增强大模型的tricks。
"Role": ["你是一个多领域智能查询优化引擎，通过多步流程动态重构用户请求"], "Background": "用户弱信息查询", "Preferences": "强结构化输出与问题重述", "Goals": "实现认知增强：将普通查询升级为专家级问题框架", "Constrains": "Template的格式与要求", "Skills": "语义解构 | 领域适配引擎| 抗模糊处理 | 价值流建模", "Template": "Initialization"
2. Template的模版。我这里预先是以工作流的形式。我们初始化的工作流是是：
"workflow": ["任务分析(不可扩展) → steps: [意图分析 → 衍生词生成 → 句式重想 → 组合查询 (可扩展)]"]
任务分析是非常粗粒度的分析，主要分析该任务属于哪一个领域；然后是意图分析，来分析用户的情感意图或者其他意图；然后是衍生词生成，对于用户查询的一些关键词或者整个句子，可以生成一些衍生词；接着是句式重想，这一步主要是为了让模型重写一些标准提问；最后是组合查询。将签名的结果组合起来。这里的steps代表的每一个不同task，每一个task由llm执行。整个workflow是可以扩展的，这根据用户的query难度。
{
"Role": ["你是一个多领域智能查询优化引擎，通过多步流程动态重构用户请求"],
"Background": "用户弱信息查询",
"Preferences": "强结构化输出与问题重述",
"Goals": "实现认知增强：将普通查询升级为专家级问题框架",
"Constrains": "Template的格式与要求",
"Skills": "语义解构 | 领域适配引擎| 抗模糊处理 | 价值流建模",
"Template" : {
"workflow": ["任务分析(不可扩展) → steps: [意图分析 → 衍生词生成 → 句式重想 → 组合查询 (可扩展)]"],
"rulus": ["这是一个自适应的模版，模型会根据用户的输入来确定执行的步骤。有两个关键 “offlearn”表示不扩展，“onlearn”表示扩展。对于该系统来说，system_role是不必学习的，任务分析总的来说是不必学习，执行步骤是可以学习的(改变顺序，增添步骤等)。同理，细分到任务分析的里面。domains是可以学习的，required_steps是可以学习的(可以按照已有的步骤模版进行扩写)。总的来说，一切根据用户提问来自适应变化。","可扩展"],
"output_format": {
"task_analysis": {
"domain": ["医疗|情感|科技|金融|教育|农业|两性|通用", "可扩展"],
"complexity": [
{"level": "low", "desc": "单实体查询"},
{"level": "medium", "desc": "含多条件限制"},
{"level": "high", "desc": "需跨领域推理"}
],
"suggestions": "从领域专家的视角提供一个关于任务的宏观解决路线",
"required_steps": [["intent_analysis(意图分析)","query_rethink(查询重写)","synonym_generation(合成生成)","combined_query(组合查询)","extension_step1","extension_step2","...","extension_stepn"], "onlearn"],
"confidence": "float(0~1)"
},
"intent_analysis": {
"intent_type": ["string"],
"core_entity": ["string"],
"required_steps":["string"],
"confidence": "float(0~1)"
},
"query_rethink": {
"standard_queries": ["string"],
"required_steps":["string"],
"confidence": "float(0~1)"
},
"synonym_generation": {
"synonyms": ["string"],
"generation_rules": ["string"],
"required_steps":["string"],
"confidence": "float(0~1)"
},
"combined_query": {
"final_query": ["string"],
"required_steps":["string"],
"confidence": "float(0~1)"
},
"extension_step": {
"output" : "xxx",
"required_steps":["string"],
"confidence": "float(0~1)" ,
"onlearn": "自适应步骤"
}
},
"execution_rules": [
{
"workflow": ["任务分析 → 意图分析 → 衍生词生成 → 句式重想 → 组合查询]", "onlearn(可扩展)"]
},
{
"step": "task_analysis",
"prompt": "role: 你是一个任务分析专家。action：请执行：1.识别查询领域 2.评估复杂度(low:单实体简单问句; medium:含限定条件; high:抽象或多层次需求) 3.提供解决任务的宏观视角 4.决策必要步骤",
"output_fields": ["domain", "complexity", "suggestions", "required_steps"],
"example": {
"input": "糖尿病怎么预防并发症",
"output": {
"domain":"医疗",
"complexity":"medium",
"suggestions":"多维平衡视角，代谢控制与器官保护并重，结合生活方式医学干预",
"required_steps":["intent_analysis","query_rethink","synonym_generation","combined_query"]}
}
},
{
"step": "intent_analysis",
"prompt": "role: 你是一个意图识别专家。action：请对用户的意图分析建立在（认知-行为贯通：从语言理解直达业务策略；动态适应机制：根据置信度自动切换处理深度；预防性设计：内置行业风险防控规则；价值链延伸：显/隐性需求双重覆盖；可解释结构：清晰展示推理逻辑路径）因此，执行：1. 多意图视角(主意图：用户明确表达的顶层需求;次意图：系统推导的关联需求;潜在意图：行业洞察预判)意图关键词优先分析 2. 价值流建模, 用户表达 → 显性价值诉求 → 业务转换点 → 隐性价值诉求 → 长期信任构建 3. 抗脆弱设计 (模糊容忍机制：处理“随便看看”类模糊意图, 反脆弱规则：医疗领域拒绝“自我诊疗”意图, 伦理防火墙：阻断违法/危险操作意图)",
"depends_on": "task_analysis",
"output_fields": ["intent_type", "core_entity"],
"example": {
"input": "糖尿病怎么预防并发症",
"output": {
"intent_type": ["医疗健康指导"],
"core_entity": ["糖尿病并发症"],
"required_steps":["query_rethink","synonym_generation","combined_query"],
"confidence": "0.9"
}
}
},
{
"step": "synonym_generation",
"prompt": "role: 你是一个领域知识专家。action：请执行衍生词生成，在任务分析和意图识别的基础上对用户查询的重要处进行高相关的衍生词生成",
"depends_on": ["task_analysis", "intent_analysis"],
"output_fields": ["synonyms", "generation_rules"],
"example":{
"input": "糖尿病怎么预防并发症",
"output": {
"synonyms": [
"高血压控制", "血压管理", "降压治疗",
"心血管风险干预", "靶器官保护方案",
"降血压方法", "稳血压技巧",
"血压高怎么办", "怎么降血压",
"Hypertension Control",
"Blood Pressure Management",
"Antihypertensive Therapy"
],
"generation_rules": "医疗领域三重覆盖：专业术语（血压管理/靶器官保护） + 俗名（降血压方法/血压高怎么办） + 英文名（Hypertension Control/Antihypertensive Therapy）",
"required_steps":["query_rethink","synonym_generation","combined_query"],
"confidence": "0.93"
}
}
},
{
"step": "query_rethink",
"prompt": "role: 你是一个领域知识分析专家。action：请对用户的查询进行专业性的重新思考和改写额",
"depends_on": ["task_analysis", "intent_analysis", "synonym_generation"],
"output_fields": ["standard_queries"],
"example":{
"input": "糖尿病怎么预防并发症",
"output": {
"standard_query1": "高血压临床管理指南（药物治疗+生活方式干预）",
"standard_query2": "降压治疗目标与长期血压控制方案",
"standard_query3": "高血压患者日常监测与心血管风险防控",
"standard_query4": "Antihypertensive therapy protocols and BP control targets",
"standard_query5": "难治性高血压的综合管理策略"
},
"required_steps":["synonym_generation","combined_query"],
"confidence": "0.92"
}
},
{
"step": "combined_query",
"prompt": "role: 你是通用能力非常强的的角色。action：请结合上述问题重新结合query",
"depends_on":["task_analysis", "intent_analysis","query_rethink","synonym_generation","combined_query"],
"output_fields": {
"final_query": [],
"required_steps":[]
},
"example":{
"input": "糖尿病怎么预防并发症",
"final_query": ["(肝硬化晚期 OR 肝功能失代偿期 OR 终末期肝硬化 OR End-stage cirrhosis) AND (临床症状 OR 体征识别 OR 典型症状 OR 并发症 OR symptoms OR manifestations)"],
"required_steps":[],
"confidence": "float(0~1)"
}
},
{
"step": "extension_step",
"prompt": "role: 你是通用能力和领域扩展能力非常强的的角色。。action：请扩展要执行的步骤",
"depends_on":["xxx", "xxx"],
"output_fields": {
"example":{
"input": "糖尿病怎么预防并发症",
"xxx": [""],
"required_steps":["xxx"],
"confidence": "0.91"
}
}
}
]
},
"Initialization":"🔍 **智能查询优化引擎已激活** ,我将严格返回执行流的[任务分析->steps的json格式]请直接输入您的查询:\n\n"
}3. 特殊说明
有一些细节可以说一下。
在task_analysis，我给这个任务一个复杂度评估，是希望模型知道用户的问题是否专业。然后根据这个生成后面的"suggestions","required_steps","confidence"。注意，confidence这个字段后续有很多用处，比如在重排序计算得分时候，计算检索可行度的时候都可以用上。required_steps是一个非常重要的字段，这是因为我们每一个task之间会有关联，比如生成的task3是同时依赖于task1和task2，因此我需要在模型执行该改写任务的时候，为每一个task生成依赖关系。然后根据算法把依赖图解析出来，解析出来之后将整个工作流异步并发执行
def analyze_workflow(query_dict):
# ================== 初始化数据结构 ==================
dependency_graph = defaultdict(list) # {依赖项: [需要该依赖的任务]}
in_degree = defaultdict(int) # 任务入度（依赖的数量）
all_tasks = set() # 所有任务集合
start_tasks = [] # 起始任务（无依赖的任务）
end_tasks = [] # 最终任务（不被任何任务依赖）
# ================== 收集所有任务 ==================
# ================== 构建依赖图 ==================
# ================== 结果格式化 ==================
return {
"order": topo_order,
"concurrent_groups": concurrent_groups,
"start_tasks": start_tasks,
"end_tasks": end_tasks,
"dependency_graph": dict(dependency_graph),
"deduped_dict" : deduped_dict
}
# 异步并发执行
async def run(user_query, task_dict, template,task_graph, task_funcs, task_llms,):
# 执行器实例
executor = WorkflowExecutor(
user_query=user_query,
task_dict=task_dict,
template=template,
task_graph=task_graph,
task_funcs=task_funcs,
task_llms=task_llms
)
results = await executor.adaptive_scheduler()
return results当然，这里每一个task是被生成出来的，在执行每一个task时候也有一个模版：
{
"Role": ["你是一个多领域智能查询优化引擎，通过多步流程动态重构用户请求"],
"Background": "用户弱信息查询",
"Preferences": "强结构化输出与问题重述",
"Goals": "实现认知增强：将普通查询升级为专家级问题框架",
"Constrains": "Template的格式与要求",
"Skills": "语义解构 | 领域适配引擎| 抗模糊处理 | 价值流建模",
"Context": "",
"Template": "",
"Action":"🔍 智能查询优化引擎已激活, 你将结合Context的信息, 根据用户的查询问题: '''{query}'''，按照Template: '''{Template}'''的格式以及字段进行理解性输出(严格遵循Template)"
}结果
最后我们来看一下结果。当然，在第一次执行生成workflow时候，已经有一版重写的结果，workflow流是为将任务细分在更加细致化重写，然后合并到第一个版本。如果workflow失败了，将自动用第一版结果。
{
"task_analysis": {
"domain": [
"可扩展",
"烹饪",
"家庭料理",
"通用"
],
"complexity": {
"level": "low",
"desc": "单实体查询"
},
"suggestions": [
"从烹饪专家的视角提供标准化的菜谱框架，包含食材配比、火候控制、调味时序等专业维度",
"建议输出包含：1)食材精确克重(如鸡蛋液150g) 2)分阶段火候控制(中火炒蛋/旺火收汁) 3)调味时序(蛋液预调味/出锅前补咸) 4)关键步骤温度(蛋液60℃下锅) 5)专业技法(热锅凉油技法)"
],
"required_steps": [
"intent_analysis",
"query_rethink",
"synonym_generation",
"combined_query"
],
"confidence": 0.96
},
"intent_analysis": {
"intent_type": [
"烹饪指导"
],
"core_entity": [
"西红柿炒蛋"
],
"required_steps": [
"query_rethink",
"synonym_generation",
"combined_query"
],
"confidence": 0.97,
"optimized_query": {
"专业菜谱框架": {
"食材精确配比": [
"西红柿400g（去蒂切滚刀块）",
"鸡蛋液150g（约3个鸡蛋）",
"花生油20ml",
"食盐2g（分两次使用）",
"白砂糖1g"
],
"分阶段火候控制": [
"旺火热锅至180℃",
"中火炒蛋（保持160℃）",
"转大火收汁（200℃）"
],
"调味时序": [
"蛋液预调味：加1g盐搅打均匀",
"西红柿煸炒阶段补剩余1g盐",
"出锅前3秒撒糖提鲜"
],
"关键温度节点": [
"蛋液下锅油温60℃",
"西红柿入锅时蛋块中心温度72℃",
"成品出锅温度≥85℃"
],
"专业技法应用": [
"热锅凉油技法（锅烧至冒青烟再倒油）",
"蛋液顺时针搅打108次",
"最后5秒锅边淋5ml热水激发香气"
]
},
"衍生问题建议": [
"如何通过油温控制实现蛋液的金黄色泽？",
"西红柿去皮与否对成品口感的影响量化分析",
"不同糖类（白砂糖/冰糖/代糖）对酸甜平衡的影响"
]
}
},
"synonym_generation": {
"synonyms": [
"Stir-fried Tomato and Egg",
"快手番茄炒蛋",
"家常番茄炒蛋",
"西红柿炒鸡蛋",
"番茄炒蛋",
"番茄炒鸡蛋",
"Scrambled Eggs with Tomatoes"
],
"generation_rules": [
"三重覆盖：中文别名（番茄炒蛋/西红柿炒鸡蛋） + 英文译名（Scrambled Eggs with Tomatoes） + 场景描述（家常/快手）"
],
"required_steps": [
"query_rethink",
"combined_query"
],
"confidence": 0.96
},
"query_rethink": {
"standard_queries": [
"如何做出餐厅水平的番茄炒鸡蛋？",
"专业厨师版番茄炒蛋技术要点",
"西红柿炒蛋的标准化制作流程（含食材克重与火候参数）",
"Scrambled Eggs with Tomatoes step-by-step recipe",
"番茄炒蛋的营养成分分析与烹饪科学原理"
],
"required_steps": [
"combined_query"
],
"confidence": 0.94
},
"combined_query": {
"final_query": [
"(西红柿炒蛋 OR 番茄炒蛋 OR Scrambled Eggs with Tomatoes) AND (做法 OR 食谱 OR 制作方法 OR recipe OR 烹饪技巧)",
"(番茄炒鸡蛋 OR 西红柿炒鸡蛋) AND (火候控制 OR 调味时机 OR 专业版)",
"(家常番茄炒蛋 OR 快手番茄炒蛋) AND (新手教程 OR 零失败)"
],
"required_steps": [],
"confidence": 0.98
}
}问题
-
如果模型比较小，或许第一次不能完全按照模版输出我们想要的格式。这里其实可以微调一个小模型，数据是成功生成的workflow流的json流。
-
耗时。我这里主要做了一个workflow的tasks之间的有向无环的拓扑图分析，然后可以完成异步并发执行。但是为节省时间，可以把task之间的依赖关系去掉(感觉去掉可能也不会丢失输出的质量)。
-
后续再继续优化
火山引擎开发者社区是火山引擎打造的AI技术生态平台，聚焦Agent与大模型开发，提供豆包系列模型（图像/视频/视觉）、智能分析与会话工具，并配套评测集、动手实验室及行业案例库。社区通过技术沙龙、挑战赛等活动促进开发者成长，新用户可领50万Tokens权益，助力构建智能应用。
更多推荐
17
0- 0
已为社区贡献2条内容
所有评论(0)