zyi
b19a2415f9
- 新增 app/api、app/service 数据分析 Agent 全套服务与接口 - schemas.py 重构为 schemas 包（analyst schema） - 新增 SQL 防注入、guardrail、缓存、字典、LLM 等服务 - 新增 tests 测试套件与 scripts/dev、scripts/setup 脚本 - 补充需求规格、架构说明书、开发清单、表设计等文档
31 lines
1.1 KiB
Python
31 lines
1.1 KiB
Python
"""诊断单个问题的 SQL 生成与执行（用法：python scripts/dev/diag_question.py "问题" 角色）。"""
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
|
|
from app.service.analyst_agent import AnalystAgent
|
|
from app.utils.auth import AuthContext
|
|
|
|
|
|
def main():
|
|
question = sys.argv[1] if len(sys.argv) > 1 else "近30天申购金额总额"
|
|
role = sys.argv[2] if len(sys.argv) > 2 else "ops"
|
|
agent = AnalystAgent()
|
|
ctx = AuthContext(subject_id=f"STAFF-{role}", token_type="staff", roles=[role], staff_type=role)
|
|
domain = {"analyst": "full", "advisor": "assigned", "risk_officer": "risk", "ops": "aggregate"}[role]
|
|
scope = []
|
|
if domain == "assigned":
|
|
scope = agent.repo.resolve_advisor_scope(ctx.subject_id)
|
|
sql, _ = agent._generate_sql(question, domain, scope)
|
|
print("SQL:", sql)
|
|
try:
|
|
res = agent.repo.execute_readonly(sql)
|
|
print("OK cols=", res["columns"], "rows[:3]=", res["rows"][:3])
|
|
except Exception as exc: # noqa: BLE001
|
|
print("EXEC ERROR:", type(exc).__name__, exc)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|