From ad0082d81109e6d492f20cab7fae401782cf027e Mon Sep 17 00:00:00 2001 From: chenjw28 <792430652@qq.com> Date: Thu, 26 Feb 2026 18:06:17 +0800 Subject: [PATCH] init --- README.md | 18 ++- _trial_temp/_trial_marker | 0 agent/conversation.py | 9 +- agent/graph.py | 8 +- agent/nodes.py | 49 +++++++- agent/tool.py | 9 +- agents/base_agent.py | 6 - agents/conversation_agent.py | 53 --------- agents/tool_agent.py | 5 - api/dependencies.py | 4 + api/endpoints.py | 49 +++++++- config/config.ini.example | 12 +- config/prompts.yaml | 7 ++ config/ragflow_templates/README.md | 4 - config/ragflow_templates/example_table.json | 8 -- .../README.md | 0 .../apbo_eta_ful.json | 0 .../example_table.json | 0 config/table_metadata_prompts/README.md | 4 - .../example_table_metadata.json | 20 ---- config/table_retrieval_prompts/README.md | 4 + config/table_retrieval_prompts/tables.json | 23 ++++ requirements.txt | 1 - scripts/sync_ragflow_templates.py | 62 +--------- scripts/sync_sql_gen_prompts.py | 11 ++ server.py | 3 + services/cache.py | 18 --- services/nacos_service.py | 7 +- services/prompt_manager.py | 11 ++ services/ragflow_client.py | 11 +- services/ragflow_sync.py | 106 ++++++++++++++++++ services/sql_prompt_manager.py | 26 +++++ services/template_matcher.py | 57 ++++------ 33 files changed, 356 insertions(+), 249 deletions(-) delete mode 100644 _trial_temp/_trial_marker delete mode 100644 agents/base_agent.py delete mode 100644 agents/conversation_agent.py delete mode 100644 agents/tool_agent.py delete mode 100644 config/ragflow_templates/README.md delete mode 100644 config/ragflow_templates/example_table.json rename config/{sql_prompts => sql_gen_prompts}/README.md (100%) rename config/{sql_prompts => sql_gen_prompts}/apbo_eta_ful.json (100%) rename config/{sql_prompts => sql_gen_prompts}/example_table.json (100%) delete mode 100644 config/table_metadata_prompts/README.md delete mode 100644 config/table_metadata_prompts/example_table_metadata.json create mode 100644 config/table_retrieval_prompts/README.md create mode 100644 config/table_retrieval_prompts/tables.json create mode 100644 scripts/sync_sql_gen_prompts.py create mode 100644 services/ragflow_sync.py create mode 100644 services/sql_prompt_manager.py diff --git a/README.md b/README.md index 879641a..06d8f7c 100644 --- a/README.md +++ b/README.md @@ -35,8 +35,7 @@ more_dots/ ├── config/ # 配置层 │ └── settings.py # 配置读取 │ └── prompts.yaml # 提示词配置 -│ └── ragflow_templates/ # RAGFlow 模板（表名 -> 模板列表） -│ └── table_metadata_prompts/ # 表模型元数据提示词 +│ └── table_retrieval_prompts/ # 表名检索提示词（表名 -> 模板列表） ├── tools/ # 工具模块 │ ├── calculator.py # 计算器工具 │ └── web_search.py # 网络搜索工具（占位符） @@ -148,14 +147,21 @@ heartbeat_interval = 5 ### RAGFlow 模板同步 -模板文件位于 `config/ragflow_templates`，每个 JSON 对应一个表名与模板列表。 +模板文件位于 `config/table_retrieval_prompts/tables.json`，单文件包含多个表名与模板列表。 同步脚本： ```bash python scripts/sync_ragflow_templates.py ``` -请在 `config/config.ini` 中配置 `ragflow.upload` 上传接口。 +请在 `config/config.ini` 中配置 `ragflow.upload` 上传接口，并分别设置： +`table_retrieval_dataset_id` 与 `sql_gen_dataset_id`。 + +默认使用覆盖更新模式（`ragflow.upload_mode = overwrite`）。 + +热更新接口： +- `POST /api/ragflow/table-retrieval/reload` +- `POST /api/ragflow/sql-gen/reload` # 使用对话工作流 result = manager.execute_workflow( @@ -191,7 +197,7 @@ class CustomTool(BaseTool): 创建新的代理类型： ```python -from agents.base_agent import BaseAgent +from agent.graph import BaseAgent class CustomAgent(BaseAgent): def _build_graph(self): @@ -215,7 +221,7 @@ class CustomAgent(BaseAgent): ### 添加新功能 1. **新工具**: 在 `tools/` 目录下创建新的工具类 -2. **新代理**: 在 `agents/` 目录下继承 `BaseAgent` 类 +2. **新代理**: 在 `agent/` 目录下继承 `BaseAgent` 类 3. **新工作流**: 在 `workflows/` 目录下扩展工作流管理器 ### 测试 diff --git a/_trial_temp/_trial_marker b/_trial_temp/_trial_marker deleted file mode 100644 index e69de29..0000000 diff --git a/agent/conversation.py b/agent/conversation.py index 0424d09..e1039ac 100644 --- a/agent/conversation.py +++ b/agent/conversation.py @@ -19,11 +19,13 @@ class ConversationAgent(BaseAgent): workflow.add_node("analyze_intent", self._analyze_intent) workflow.add_node("normalize_input", self._normalize_input) + workflow.add_node("generate_sql", self._generate_sql) workflow.add_node("generate_response", self._generate_response) workflow.add_node("update_context", self._update_context) workflow.add_edge("analyze_intent", "normalize_input") - workflow.add_edge("normalize_input", "generate_response") + workflow.add_edge("normalize_input", "generate_sql") + workflow.add_edge("generate_sql", "generate_response") workflow.add_edge("generate_response", "update_context") workflow.add_edge("update_context", END) @@ -61,6 +63,11 @@ class ConversationAgent(BaseAgent): state.current_step = "response_generated" return state + def _generate_sql(self, state: AgentState) -> AgentState: + """生成 SQL""" + from . import nodes + return nodes.generate_sql(state, self.model) + def _update_context(self, state: AgentState) -> AgentState: """更新对话上下文与历史""" for message in state.messages: diff --git a/agent/graph.py b/agent/graph.py index b73d08a..3be3d8f 100644 --- a/agent/graph.py +++ b/agent/graph.py @@ -20,10 +20,12 @@ class BaseAgent: workflow.add_node("process_input", nodes.process_input) workflow.add_node("normalize_input", self._normalize_input) + workflow.add_node("generate_sql", self._generate_sql) workflow.add_node("generate_response", self._generate_response) workflow.add_edge("process_input", "normalize_input") - workflow.add_edge("normalize_input", "generate_response") + workflow.add_edge("normalize_input", "generate_sql") + workflow.add_edge("generate_sql", "generate_response") workflow.add_edge("generate_response", END) workflow.set_entry_point("process_input") @@ -38,6 +40,10 @@ class BaseAgent: """规范化用户输入""" return nodes.normalize_input(state, self.model) + def _generate_sql(self, state: AgentState) -> AgentState: + """生成 SQL""" + return nodes.generate_sql(state, self.model) + def run(self, user_input: str, **kwargs) -> Dict[str, Any]: """运行代理并处理用户输入""" initial_state = AgentState( diff --git a/agent/nodes.py b/agent/nodes.py index 136f89a..7006a83 100644 --- a/agent/nodes.py +++ b/agent/nodes.py @@ -1,7 +1,11 @@ -from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage +import json + +from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage, AIMessage from .state import AgentState -from services.prompt_manager import PromptManager -from services.template_matcher import TemplateMatcher +from services.prompt_manager import get_prompt_manager +from services.template_matcher import get_template_matcher +from services.sql_prompt_manager import SqlPromptManager +from tools.sr_api_tool import SrApiQueryTool def process_input(state: AgentState) -> AgentState: @@ -12,6 +16,14 @@ def process_input(state: AgentState) -> AgentState: def generate_response(state: AgentState, model) -> AgentState: """使用 LLM 生成回复""" + sr_api_result = state.context.get("sr_api_result") + if sr_api_result: + state.messages.append(AIMessage(content=str(sr_api_result))) + return state + final_sql = state.context.get("final_sql") + if final_sql: + state.messages.append(AIMessage(content=final_sql)) + return state if state.messages: response = model.invoke(state.messages) state.messages.append(response) @@ -27,7 +39,7 @@ def normalize_input(state: AgentState, model) -> AgentState: if not isinstance(last_message, HumanMessage): return state - prompt_manager = PromptManager() + prompt_manager = get_prompt_manager() system_prompt = SystemMessage( content=prompt_manager.get("system", "english_normalizer") ) @@ -38,6 +50,33 @@ def normalize_input(state: AgentState, model) -> AgentState: state.context["original_input"] = last_message.content state.context["normalized_input"] = normalized - matcher = TemplateMatcher() + matcher = get_template_matcher() state.context["table_match"] = matcher.match(normalized) return state + + +def generate_sql(state: AgentState, model) -> AgentState: + """根据表名与提示词生成 SQL""" + table_match = state.context.get("table_match") or {} + table_name = table_match.get("table_name") + normalized = state.context.get("normalized_input") + + if not table_name or not normalized: + return state + + prompt_manager = SqlPromptManager() + prompt_data = prompt_manager.get_prompt(table_name) + if not prompt_data: + return state + + prompt_text = json.dumps(prompt_data, ensure_ascii=False, indent=2) + system_template = get_prompt_manager().get("system", "sql_mysql_select_only") + system_content = system_template.format(table_prompt_json=prompt_text) + user_content = f"User question (normalized English): {normalized}" + response = model.invoke([SystemMessage(content=system_content), HumanMessage(content=user_content)]) + sql_text = response.content if hasattr(response, "content") else str(response) + + state.context["final_sql"] = sql_text + tool = SrApiQueryTool() + state.context["sr_api_result"] = tool.run(json.dumps({"sql": sql_text}, ensure_ascii=False)) + return state diff --git a/agent/tool.py b/agent/tool.py index 06ea8e8..622f814 100644 --- a/agent/tool.py +++ b/agent/tool.py @@ -28,10 +28,12 @@ class ToolAgent(BaseAgent): workflow = StateGraph(AgentState) workflow.add_node("normalize_input", self._normalize_input) + workflow.add_node("generate_sql", self._generate_sql) workflow.add_node("agent", self._agent_node) workflow.add_node("tools", self.tool_node) - workflow.add_edge("normalize_input", "agent") + workflow.add_edge("normalize_input", "generate_sql") + workflow.add_edge("generate_sql", "agent") workflow.add_edge("tools", "agent") workflow.add_conditional_edges( @@ -65,6 +67,11 @@ class ToolAgent(BaseAgent): return state + def _generate_sql(self, state: AgentState) -> AgentState: + """生成 SQL""" + from . import nodes + return nodes.generate_sql(state, self.model) + def _should_use_tools(self, state: AgentState) -> str: """判断是否需要使用工具""" last_message = state.messages[-1] diff --git a/agents/base_agent.py b/agents/base_agent.py deleted file mode 100644 index c2acc4d..0000000 --- a/agents/base_agent.py +++ /dev/null @@ -1,6 +0,0 @@ -"""兼容导出：请优先使用 agent 包""" - -from agent.graph import BaseAgent -from agent.state import AgentState - -__all__ = ["BaseAgent", "AgentState"] \ No newline at end of file diff --git a/agents/conversation_agent.py b/agents/conversation_agent.py deleted file mode 100644 index 6de1eff..0000000 --- a/agents/conversation_agent.py +++ /dev/null @@ -1,53 +0,0 @@ -from typing import Dict, Any, List, Optional -from langchain_core.messages import BaseMessage, HumanMessage, AIMessage -from langgraph.graph import StateGraph, END -from .base_agent import BaseAgent, AgentState - - -class ConversationAgent(BaseAgent): - """处理多轮对话的代理""" - - def __init__(self, model_section: Optional[str] = None): - super().__init__(model_section) - self.conversation_history: List[BaseMessage] = [] - - def _build_graph(self) -> StateGraph: - """构建对话专用图""" - workflow = StateGraph(AgentState) - - # 添加节点 - workflow.add_node("analyze_intent", self._analyze_intent) - workflow.add_node("generate_response", self._generate_response) - workflow.add_node("update_context", self._update_context) - - # 定义边 - workflow.add_edge("analyze_intent", "generate_response") - workflow.add_edge("generate_response", "update_context") - workflow.add_edge("update_context", END) - - # 设置入口节点 - workflow.set_entry_point("analyze_intent") - - return workflow.compile() - - def _analyze_intent(self, state: AgentState) -> AgentState: - """分析用户意图与对话上下文""" - # 简单意图分析，可用更复杂逻辑增强 - user_message = state.messages[-1] if state.messages else None - - if user_message and isinstance(user_message, HumanMessage): - content = user_message.content.lower() - - # 基础意图识别 - if any(word in content for word in ["hello", "hi", "hey", "greetings"]): - state.context["intent"] = "greeting" - elif any(word in content for word in ["help", "assist", "support"]): - state.context["intent"] = "help" - elif "?" in content: - state.context["intent"] = "question" - else: - """兼容导出：请优先使用 agent 包""" - - from agent.conversation import ConversationAgent - - __all__ = ["ConversationAgent"] \ No newline at end of file diff --git a/agents/tool_agent.py b/agents/tool_agent.py deleted file mode 100644 index db73474..0000000 --- a/agents/tool_agent.py +++ /dev/null @@ -1,5 +0,0 @@ -"""兼容导出：请优先使用 agent 包""" - -from agent.tool import ToolAgent - -__all__ = ["ToolAgent"] \ No newline at end of file diff --git a/api/dependencies.py b/api/dependencies.py index c3c6f16..37fb8ae 100644 --- a/api/dependencies.py +++ b/api/dependencies.py @@ -15,3 +15,7 @@ def get_service_config(request: Request): def get_tool_router(request: Request): return request.app.state.tool_router + + +def get_prompt_manager(request: Request): + return request.app.state.prompt_manager diff --git a/api/endpoints.py b/api/endpoints.py index b371ea3..a4d02b8 100644 --- a/api/endpoints.py +++ b/api/endpoints.py @@ -6,7 +6,8 @@ from schemas.agent_output import AgentOutput from schemas.tool_input import ToolInput from schemas.tool_output import ToolOutput from workflows.workflow_manager import WorkflowType -from api.dependencies import get_workflow_manager, get_nacos_manager, get_service_config, get_tool_router +from api.dependencies import get_workflow_manager, get_nacos_manager, get_service_config, get_tool_router, get_prompt_manager +from services.ragflow_sync import RagflowSync router = APIRouter() @@ -63,12 +64,32 @@ def run_workflow_stream(payload: AgentInput, workflow_manager=Depends(get_workfl if workflow_type != WorkflowType.CONVERSATION: raise HTTPException(status_code=400, detail="仅支持对话工作流的流式输出") - agent = workflow_manager.get_workflow(workflow_type) + def _extract_output_text(result: dict) -> str: + context = (result.get("context") or {}) if isinstance(result, dict) else {} + if "sr_api_result" in context: + return str(context.get("sr_api_result") or "") + messages = result.get("messages") if isinstance(result, dict) else None + if messages: + last = messages[-1] + if hasattr(last, "content"): + return str(last.content or "") + return "" def event_stream(): try: - for token in agent.stream_run(payload.input): - yield f"data: {token}\n\n" + result = workflow_manager.execute_workflow( + workflow_type=workflow_type, + user_input=payload.input, + session_id=payload.session_id, + ) + text = _extract_output_text(result.get("result") or {}) + if not text: + yield "event: end\ndata: [DONE]\n\n" + return + chunk_size = 512 + for i in range(0, len(text), chunk_size): + chunk = text[i : i + chunk_size] + yield f"data: {chunk}\n\n" yield "event: end\ndata: [DONE]\n\n" except Exception as e: yield f"event: error\ndata: {str(e)}\n\n" @@ -80,3 +101,23 @@ def run_workflow_stream(payload: AgentInput, workflow_manager=Depends(get_workfl def run_tool(payload: ToolInput, tool_router=Depends(get_tool_router)): result = tool_router.call(payload.tool_name, payload.payload) return ToolOutput(**result) + + +@router.post("/api/prompts/reload") +def reload_prompts(prompt_manager=Depends(get_prompt_manager)): + prompt_manager.reload() + return {"ok": True} + + +@router.post("/api/ragflow/table-retrieval/reload") +def reload_table_retrieval(): + syncer = RagflowSync() + result = syncer.sync_table_retrieval() + return {"ok": True, "result": result} + + +@router.post("/api/ragflow/sql-gen/reload") +def reload_sql_gen(): + syncer = RagflowSync() + result = syncer.sync_sql_gen_prompts() + return {"ok": True, "result": result} diff --git a/config/config.ini.example b/config/config.ini.example index 0e03063..b89b59f 100644 --- a/config/config.ini.example +++ b/config/config.ini.example @@ -32,17 +32,13 @@ model_section = gpt-4o [ragflow] url = http://10.122.176.97:21020 -dataset_ids = f5b8b854d63a11f083230242c0a8e006 -document_ids = 819cf100f52611f0a7fa0242c0a8e006 api_key = ragflow-xxxxx retrieval = /api/v1/retrieval upload = /api/v1/documents -cache_ttl = 600 - -[redis] -enabled = false -url = redis://localhost:6379/0 -db = 0 +# 上传模式：overwrite（覆盖更新）或 append（追加） +upload_mode = overwrite +table_retrieval_dataset_id = +sql_gen_dataset_id = [nacos] # 是否启用 Nacos 注册 diff --git a/config/prompts.yaml b/config/prompts.yaml index 8185d6a..4bb211e 100644 --- a/config/prompts.yaml +++ b/config/prompts.yaml @@ -4,6 +4,13 @@ system: sql_generator: | You are an assistant that converts user intent into SQL. Ensure the SQL is correct, safe, and syntactically valid. + sql_mysql_select_only: | + You are an expert SQL generator. + Only output a single MySQL SELECT statement. + Do not use LIMIT. + Do not output any other text. + Use the following table prompt JSON to generate SQL: + {table_prompt_json} user: default: | diff --git a/config/ragflow_templates/README.md b/config/ragflow_templates/README.md deleted file mode 100644 index cb6dd91..0000000 --- a/config/ragflow_templates/README.md +++ /dev/null @@ -1,4 +0,0 @@ -本目录存放用于 RAGFlow 匹配的模板配置（JSON）。 - -约定：每个 JSON 文件对应一个表名（key），包含模板列表（value）。 -示例文件：orders.json、customers.json。 diff --git a/config/ragflow_templates/example_table.json b/config/ragflow_templates/example_table.json deleted file mode 100644 index b662350..0000000 --- a/config/ragflow_templates/example_table.json +++ /dev/null @@ -1,8 +0,0 @@ -{ - "table": "example_table", - "templates": [ - "example_table created in {date}", - "count of records in example_table", - "example_table where status = {status}" - ] -} diff --git a/config/sql_prompts/README.md b/config/sql_gen_prompts/README.md similarity index 100% rename from config/sql_prompts/README.md rename to config/sql_gen_prompts/README.md diff --git a/config/sql_prompts/apbo_eta_ful.json b/config/sql_gen_prompts/apbo_eta_ful.json similarity index 100% rename from config/sql_prompts/apbo_eta_ful.json rename to config/sql_gen_prompts/apbo_eta_ful.json diff --git a/config/sql_prompts/example_table.json b/config/sql_gen_prompts/example_table.json similarity index 100% rename from config/sql_prompts/example_table.json rename to config/sql_gen_prompts/example_table.json diff --git a/config/table_metadata_prompts/README.md b/config/table_metadata_prompts/README.md deleted file mode 100644 index c23619c..0000000 --- a/config/table_metadata_prompts/README.md +++ /dev/null @@ -1,4 +0,0 @@ -本目录存放业务表模型元数据提示词（JSON 格式）。 - -约定：每个 JSON 文件对应一个数据库表模型，描述表与字段元数据。 -示例文件：order_metadata.json、customer_metadata.json 等。 diff --git a/config/table_metadata_prompts/example_table_metadata.json b/config/table_metadata_prompts/example_table_metadata.json deleted file mode 100644 index ce922ca..0000000 --- a/config/table_metadata_prompts/example_table_metadata.json +++ /dev/null @@ -1,20 +0,0 @@ -{ - "table": "example_table", - "description": "示例表模型元数据提示词", - "fields": [ - { - "name": "id", - "type": "int", - "description": "主键" - }, - { - "name": "created_at", - "type": "datetime", - "description": "创建时间" - } - ], - "relationships": [], - "notes": [ - "字段含义与业务规则可在此补充" - ] -} diff --git a/config/table_retrieval_prompts/README.md b/config/table_retrieval_prompts/README.md new file mode 100644 index 0000000..dfc05fe --- /dev/null +++ b/config/table_retrieval_prompts/README.md @@ -0,0 +1,4 @@ +本目录存放用于 RAGFlow 匹配的表名检索提示词（JSON）。 + +约定：使用单一 JSON 文件维护多个表名及其模板列表。 +示例文件：tables.json。 diff --git a/config/table_retrieval_prompts/tables.json b/config/table_retrieval_prompts/tables.json new file mode 100644 index 0000000..65e0847 --- /dev/null +++ b/config/table_retrieval_prompts/tables.json @@ -0,0 +1,23 @@ +{ + "tables": { + "apbo_eta_ful": [ + "SO ETA INFO", + "BO list", + "CC", + "Commodity Code", + "BO QTY", + "open order", + "BO status", + "BO report", + "model", + "MTM SN", + "backlog", + "Premier", + "SN machine_sn", + "work order information", + "recovery ETA", + "history order", + "Warranty type" + ] + } +} diff --git a/requirements.txt b/requirements.txt index 09d67bb..491d7c1 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,4 +8,3 @@ uvicorn>=0.30.0 nacos-sdk-python>=2.0.9 httpx>=0.27.0 pyyaml>=6.0.1 -redis>=5.0.0 \ No newline at end of file diff --git a/scripts/sync_ragflow_templates.py b/scripts/sync_ragflow_templates.py index 49acb1c..5b0fb89 100644 --- a/scripts/sync_ragflow_templates.py +++ b/scripts/sync_ragflow_templates.py @@ -1,62 +1,8 @@ -import json -import os -from typing import Dict, List - -import httpx - -from config import Config - - -def load_templates(dir_path: str) -> List[Dict[str, any]]: - items = [] - for name in os.listdir(dir_path): - if not name.endswith(".json"): - continue - with open(os.path.join(dir_path, name), "r", encoding="utf-8") as f: - items.append(json.load(f)) - return items - - -def build_document(item: Dict[str, any]) -> str: - table = item.get("table", "") - templates = item.get("templates", []) - lines = [f"table: {table}"] - for t in templates: - lines.append(f"- {t}") - return "\n".join(lines) - - def main(): - cfg = Config.get_section("ragflow") - base_url = cfg.get("url", "").rstrip("/") - api_key = cfg.get("api_key", "") - dataset_ids = cfg.get("dataset_ids", "") - - upload_path = cfg.get("upload", "") - if not upload_path: - raise RuntimeError("未配置 ragflow.upload 上传接口，请在 config/config.ini 中设置") - - url = base_url + "/" + upload_path.lstrip("/") - headers = {"Authorization": f"Bearer {api_key}"} if api_key else {} - - root = os.path.dirname(os.path.dirname(__file__)) - templates_dir = os.path.join(root, "config", "ragflow_templates") - items = load_templates(templates_dir) - - payload = [] - for item in items: - payload.append( - { - "dataset_ids": dataset_ids, - "content": build_document(item), - "metadata": {"table": item.get("table")}, - } - ) - - with httpx.Client(timeout=60) as client: - response = client.post(url, json={"documents": payload}, headers=headers) - response.raise_for_status() - print("同步完成") + from services.ragflow_sync import RagflowSync + syncer = RagflowSync() + syncer.sync_table_retrieval() + print("表名检索模板同步完成") if __name__ == "__main__": diff --git a/scripts/sync_sql_gen_prompts.py b/scripts/sync_sql_gen_prompts.py new file mode 100644 index 0000000..604facf --- /dev/null +++ b/scripts/sync_sql_gen_prompts.py @@ -0,0 +1,11 @@ +from services.ragflow_sync import RagflowSync + + +def main(): + syncer = RagflowSync() + syncer.sync_sql_gen_prompts() + print("SQL 生成提示词同步完成") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/server.py b/server.py index 9a2e6f5..8ef27fa 100644 --- a/server.py +++ b/server.py @@ -8,6 +8,7 @@ from config import Config from workflows.workflow_manager import WorkflowManager from services.nacos_service import load_nacos_config, load_service_config, NacosManager from services.tool_router import ToolRouter +from services.prompt_manager import get_prompt_manager from api import endpoints logger = logging.getLogger(__name__) @@ -22,6 +23,7 @@ def create_app() -> FastAPI: workflow_manager = WorkflowManager(default_model_section=default_model_section) nacos_manager = NacosManager(nacos_config=nacos_config, service_config=service_config) tool_router = ToolRouter() + prompt_manager = get_prompt_manager() @asynccontextmanager async def lifespan(app: FastAPI): @@ -31,6 +33,7 @@ def create_app() -> FastAPI: app.state.nacos_manager = nacos_manager app.state.service_config = service_config app.state.tool_router = tool_router + app.state.prompt_manager = prompt_manager await nacos_manager.start() logger.info("✅ 服务准备就绪: %s on %s:%s", service_config.service_name, service_config.ip, service_config.port) diff --git a/services/cache.py b/services/cache.py index 185c61f..ce2fddb 100644 --- a/services/cache.py +++ b/services/cache.py @@ -2,11 +2,6 @@ from __future__ import annotations from typing import Optional -try: - import redis -except Exception: - redis = None - class CacheBase: """缓存接口""" @@ -28,16 +23,3 @@ class NoopCache(CacheBase): return None -class RedisCache(CacheBase): - """Redis 缓存""" - - def __init__(self, url: str, db: int = 0): - if redis is None: - raise ImportError("未安装 redis 依赖") - self._client = redis.Redis.from_url(url, db=db, decode_responses=True) - - def get(self, key: str) -> Optional[str]: - return self._client.get(key) - - def set(self, key: str, value: str, ttl: int) -> None: - self._client.set(key, value, ex=ttl) diff --git a/services/nacos_service.py b/services/nacos_service.py index 4e1004f..2b585c5 100644 --- a/services/nacos_service.py +++ b/services/nacos_service.py @@ -3,12 +3,7 @@ import logging import socket from dataclasses import dataclass from typing import Any, Dict, Optional - -try: - import nacos -except Exception: - nacos = None - +import nacos from config import Config logger = logging.getLogger(__name__) diff --git a/services/prompt_manager.py b/services/prompt_manager.py index 211f27a..b1db33d 100644 --- a/services/prompt_manager.py +++ b/services/prompt_manager.py @@ -29,3 +29,14 @@ class PromptManager: def list_prompts(self, group: str) -> list[str]: """列出分组内提示词""" return list(self._data.get(group, {}).keys()) + + +_GLOBAL_PROMPT_MANAGER: Optional[PromptManager] = None + + +def get_prompt_manager(config_path: Optional[str] = None) -> PromptManager: + """获取全局 PromptManager（单例）""" + global _GLOBAL_PROMPT_MANAGER + if _GLOBAL_PROMPT_MANAGER is None: + _GLOBAL_PROMPT_MANAGER = PromptManager(config_path=config_path) + return _GLOBAL_PROMPT_MANAGER diff --git a/services/ragflow_client.py b/services/ragflow_client.py index 51aaae9..f875a90 100644 --- a/services/ragflow_client.py +++ b/services/ragflow_client.py @@ -14,22 +14,27 @@ class RagflowClient: self._base_url = cfg.get("url", "") self._api_key = cfg.get("api_key", "") self._retrieval_path = cfg.get("retrieval", "/api/v1/retrieval") - self._dataset_ids = cfg.get("dataset_ids", "") + self._table_retrieval_dataset_id = cfg.get("table_retrieval_dataset_id", "") + self._sql_gen_dataset_id = cfg.get("sql_gen_dataset_id", "") def _build_url(self) -> str: return self._base_url.rstrip("/") + "/" + self._retrieval_path.lstrip("/") - def retrieve(self, query: str, top_k: int = 3) -> Dict[str, Any]: + def retrieve(self, query: str, top_k: int = 3, dataset_id: Optional[str] = None, document_ids: Optional[str] = None) -> Dict[str, Any]: """检索匹配文档""" if not self._base_url or not self._retrieval_path: raise RuntimeError("未配置 ragflow.url 或 ragflow.retrieval") + if not dataset_id: + raise ValueError("未提供 ragflow.dataset_id，无法进行检索") url = self._build_url() headers = {"Authorization": f"Bearer {self._api_key}"} if self._api_key else {} payload = { - "dataset_ids": self._dataset_ids, + "dataset_ids": dataset_id or "", "query": query, "top_k": top_k, } + if document_ids: + payload["document_ids"] = document_ids with httpx.Client(timeout=30) as client: response = client.post(url, json=payload, headers=headers) diff --git a/services/ragflow_sync.py b/services/ragflow_sync.py new file mode 100644 index 0000000..48dbd39 --- /dev/null +++ b/services/ragflow_sync.py @@ -0,0 +1,106 @@ +import json +import os +from typing import Dict, List, Optional + +import httpx + +from config import Config + + +def _build_document_for_table(table: str, templates: List[str]) -> str: + lines = [f"table: {table}"] + for t in templates: + lines.append(f"- {t}") + return "\n".join(lines) + + +def _build_sql_gen_document(table: str, prompt: Dict[str, any]) -> str: + system_prompt = prompt.get("system_prompt", "") + business_prompt = prompt.get("business_prompt", "") + constraints = prompt.get("constraints", []) + lines = [f"table: {table}"] + if system_prompt: + lines.append("[system] " + system_prompt) + if business_prompt: + lines.append("[business] " + business_prompt) + if constraints: + lines.append("[constraints]") + for c in constraints: + lines.append(f"- {c}") + return "\n".join(lines) + + +class RagflowSync: + """RAGFlow 同步工具""" + + def __init__(self): + cfg = Config.get_section("ragflow") + self._base_url = cfg.get("url", "").rstrip("/") + self._api_key = cfg.get("api_key", "") + self._table_retrieval_dataset_id = (cfg.get("table_retrieval_dataset_id") or "").strip() + self._sql_gen_dataset_id = (cfg.get("sql_gen_dataset_id") or "").strip() + self._upload_path = (cfg.get("upload") or "").strip() + self._upload_mode = (cfg.get("upload_mode") or "overwrite").strip().lower() + + def _validate_common(self) -> None: + if not self._base_url: + raise RuntimeError("未配置 ragflow.url") + if not self._upload_path: + raise RuntimeError("未配置 ragflow.upload 上传接口，请在 config/config.ini 中设置") + if self._upload_mode not in ("overwrite", "append"): + raise RuntimeError("ragflow.upload_mode 仅支持 overwrite 或 append") + + def _post(self, documents: List[Dict[str, any]]): + self._validate_common() + url = self._base_url + "/" + self._upload_path.lstrip("/") + headers = {"Authorization": f"Bearer {self._api_key}"} if self._api_key else {} + payload: Dict[str, any] = {"documents": documents} + if self._upload_mode == "overwrite": + payload["mode"] = "overwrite" + with httpx.Client(timeout=60) as client: + response = client.post(url, json=payload, headers=headers) + response.raise_for_status() + return response.json() + + def sync_table_retrieval(self) -> Dict[str, any]: + if not self._table_retrieval_dataset_id: + raise RuntimeError("未配置 ragflow.table_retrieval_dataset_id，无法同步表名检索模板") + root = os.path.dirname(os.path.dirname(__file__)) + tables_file = os.path.join(root, "config", "table_retrieval_prompts", "tables.json") + with open(tables_file, "r", encoding="utf-8") as f: + data = json.load(f) + + tables = data.get("tables", {}) + documents = [] + for table, templates in tables.items(): + doc = { + "dataset_ids": self._table_retrieval_dataset_id, + "content": _build_document_for_table(table, templates), + "metadata": {"table": table}, + } + documents.append(doc) + + return self._post(documents) + + def sync_sql_gen_prompts(self) -> Dict[str, any]: + if not self._sql_gen_dataset_id: + raise RuntimeError("未配置 ragflow.sql_gen_dataset_id，无法同步 SQL 生成提示词") + root = os.path.dirname(os.path.dirname(__file__)) + prompts_dir = os.path.join(root, "config", "sql_gen_prompts") + documents = [] + + for name in os.listdir(prompts_dir): + if not name.endswith(".json"): + continue + path = os.path.join(prompts_dir, name) + with open(path, "r", encoding="utf-8") as f: + prompt = json.load(f) + table = prompt.get("table") or os.path.splitext(name)[0] + doc = { + "dataset_ids": self._sql_gen_dataset_id, + "content": _build_sql_gen_document(table, prompt), + "metadata": {"table": table}, + } + documents.append(doc) + + return self._post(documents) diff --git a/services/sql_prompt_manager.py b/services/sql_prompt_manager.py new file mode 100644 index 0000000..3fb09af --- /dev/null +++ b/services/sql_prompt_manager.py @@ -0,0 +1,26 @@ +import json +import os +from typing import Any, Dict, Optional + + +class SqlPromptManager: + """按表名读取 SQL 提示词""" + + def __init__(self, base_dir: Optional[str] = None): + root_dir = os.path.dirname(os.path.dirname(__file__)) + self._base_dir = base_dir or os.path.join(root_dir, "config", "sql_gen_prompts") + + @staticmethod + def _safe_filename(name: str) -> str: + return name.replace("..", "").replace("/", "_").replace("\\", "_") + + def get_prompt(self, table_name: str) -> Optional[Dict[str, Any]]: + """读取指定表的提示词 JSON""" + if not table_name: + return None + filename = self._safe_filename(table_name) + ".json" + path = os.path.join(self._base_dir, filename) + if not os.path.exists(path): + return None + with open(path, "r", encoding="utf-8") as f: + return json.load(f) diff --git a/services/template_matcher.py b/services/template_matcher.py index 6b6c422..4054752 100644 --- a/services/template_matcher.py +++ b/services/template_matcher.py @@ -1,53 +1,28 @@ -import hashlib -import json -from typing import Any, Dict, Optional +from typing import Any, Dict from config import Config -from services.cache import NoopCache, RedisCache from services.ragflow_client import RagflowClient, extract_table_name class TemplateMatcher: - """模板匹配器：RAGFlow + Redis 缓存""" + """模板匹配器：RAGFlow 检索""" def __init__(self): self._ragflow = RagflowClient() - self._cache = self._init_cache() cfg = Config.get_section("ragflow") - self._cache_ttl = int(cfg.get("cache_ttl", 600)) + self._dataset_id = (cfg.get("table_retrieval_dataset_id") or "").strip() - def _init_cache(self): - cfg = Config.get_section("redis") - enabled = str(cfg.get("enabled", "false")).lower() in ("1", "true", "yes") - if not enabled: - return NoopCache() - - url = cfg.get("url") - db = int(cfg.get("db", 0)) - if not url: - return NoopCache() - - try: - return RedisCache(url=url, db=db) - except Exception: - return NoopCache() - - @staticmethod - def _cache_key(text: str) -> str: - return "ragflow:table:" + hashlib.sha256(text.encode("utf-8")).hexdigest() + def _validate(self) -> None: + if not self._dataset_id: + raise RuntimeError("未配置 ragflow.table_retrieval_dataset_id，无法进行表名检索") def match(self, normalized_text: str) -> Dict[str, Any]: """返回匹配的表名与原始响应""" - key = self._cache_key(normalized_text) - cached = self._cache.get(key) - if cached: - return json.loads(cached) + self._validate() try: - response = self._ragflow.retrieve(normalized_text, top_k=3) + response = self._ragflow.retrieve(normalized_text, top_k=3, dataset_id=self._dataset_id) except Exception as e: - result = {"table_name": None, "raw": {"error": str(e)}} - self._cache.set(key, json.dumps(result, ensure_ascii=False), self._cache_ttl) - return result + return {"table_name": None, "raw": {"error": str(e)}} candidates = [] data = response.get("data") if isinstance(response, dict) else None @@ -58,7 +33,15 @@ class TemplateMatcher: candidates.append(table_name) matched = candidates[0] if candidates else None - result = {"table_name": matched, "raw": response} + return {"table_name": matched, "raw": response} - self._cache.set(key, json.dumps(result, ensure_ascii=False), self._cache_ttl) - return result + +_GLOBAL_TEMPLATE_MATCHER: TemplateMatcher | None = None + + +def get_template_matcher() -> TemplateMatcher: + """获取全局 TemplateMatcher（单例）""" + global _GLOBAL_TEMPLATE_MATCHER + if _GLOBAL_TEMPLATE_MATCHER is None: + _GLOBAL_TEMPLATE_MATCHER = TemplateMatcher() + return _GLOBAL_TEMPLATE_MATCHER