fix: eliminate duplicate SQLAlchemy table definitions in ai_models.py
Some checks failed
CD Pipeline / deploy (push) Failing after 2m47s

AgentContext/ActionPlan/ActionOutcome/AgentStrategyWeights were defined
in both ai_models.py and autoheal_models.py, causing:
  "Table 'agent_context' is already defined for this MetaData instance"
on every scheduler startup.

ai_models.py is now a pure re-export shim from autoheal_models.py.
autoheal_models.py remains the single source of truth (ADR-013).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
ogt
2026-04-20 04:47:23 +08:00
parent 266af27fd6
commit f2b20c1892

View File

@@ -1,89 +1,125 @@
# database/ai_models.py
from sqlalchemy import Column, Integer, String, DateTime, Text, Float, ForeignKey, Index
from sqlalchemy.orm import relationship
# ⚠️ 這四個 class 的原始定義已移至 autoheal_models.pyADR-013 統一管理)。
# 此檔僅作向後相容 re-export shim不再重複定義 SQLAlchemy Table
# 以避免 "Table already defined for this MetaData instance" 衝突。
from .autoheal_models import ( # noqa: F401
AgentContext,
ActionPlan,
ActionOutcome,
AgentStrategyWeights,
)
# AI history and template models
from sqlalchemy import Column, Integer, String, DateTime, Text, Boolean, Float
from database.models import Base
from datetime import datetime, timezone
from datetime import datetime
# helper for default timestamps
datetime_now = lambda: datetime.now(timezone.utc)
class AgentContext(Base):
class AIGenerationHistory(Base):
"""
Shared context table (replaces hardcoded chain), supporting multi-agent access and TTL.
Index: (session_id, agent_name, context_key) for fast cross-agent queries.
AI generation history tracking
"""
__tablename__ = 'agent_context'
__tablename__ = 'ai_generation_history'
id = Column(Integer, primary_key=True, autoincrement=True)
session_id = Column(String(64), nullable=False, index=True)
agent_name = Column(String(50), nullable=False, index=True)
context_key = Column(String(100), nullable=False)
context_val = Column(Text) # JSON string
created_at = Column(DateTime, default=datetime_now)
ttl_minutes = Column(Integer, default=60)
generation_type = Column(String(50), nullable=False) # product_desc, marketing_copy, etc.
product_name = Column(String(200))
input_keywords = Column(Text) # JSON string
input_trend_topic = Column(String(500))
output_content = Column(Text)
generation_duration = Column(Float) # seconds
is_favorite = Column(Boolean, default=False)
is_used = Column(Boolean, default=False)
created_by = Column(String(100))
created_at = Column(DateTime, default=datetime.now)
__table_args__ = (
Index('idx_agent_context_session_key', 'session_id', 'agent_name', 'context_key'),
Index('idx_agent_context_session_ttl', 'session_id', 'created_at'),
)
def to_dict(self):
return {
'id': self.id,
'generation_type': self.generation_type,
'product_name': self.product_name,
'input_keywords': self.input_keywords,
'input_trend_topic': self.input_trend_topic,
'output_content': self.output_content,
'generation_duration': self.generation_duration,
'is_favorite': self.is_favorite,
'is_used': self.is_used,
'created_by': self.created_by,
'created_at': self.created_at.isoformat() if self.created_at else None
}
class ActionPlan(Base):
class AIPromptTemplate(Base):
"""
Action plan table (NemoTron output, awaiting review/execution tracking).
AI prompt templates
"""
__tablename__ = 'action_plans'
__tablename__ = 'ai_prompt_templates'
id = Column(Integer, primary_key=True, autoincrement=True)
session_id = Column(String(64), nullable=True)
plan_type = Column(String(50), nullable=True) # price_adjust / restock / campaign
sku = Column(String(100), nullable=True, index=True)
payload = Column(Text) # JSON payload
status = Column(String(20), default='pending') # pending/approved/rejected/executed
created_by = Column(String(50)) # nemotron / openclaw
approved_by = Column(String(100), nullable=True) # Telegram user_id
created_at = Column(DateTime, default=datetime_now)
executed_at = Column(DateTime, nullable=True)
name = Column(String(200), nullable=False)
description = Column(Text)
template_type = Column(String(50), nullable=False) # product_desc, marketing_copy, etc.
prompt_content = Column(Text, nullable=False)
variables = Column(Text) # JSON string of variable definitions
is_active = Column(Boolean, default=True)
is_system = Column(Boolean, default=False)
created_by = Column(String(100))
created_at = Column(DateTime, default=datetime.now)
updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
__table_args__ = (
Index('idx_action_plan_sku_status', 'sku', 'status'),
Index('idx_action_plan_created', 'created_at'),
)
def to_dict(self):
return {
'id': self.id,
'name': self.name,
'description': self.description,
'template_type': self.template_type,
'prompt_content': self.prompt_content,
'variables': self.variables,
'is_active': self.is_active,
'is_system': self.is_system,
'created_by': self.created_by,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None
}
class ActionOutcome(Base):
class AIUsageTracking(Base):
"""
Action outcome tracking (closed-loop learning core).
AI usage tracking for analytics and billing
"""
__tablename__ = 'action_outcomes'
__tablename__ = 'ai_usage_tracking'
id = Column(Integer, primary_key=True, autoincrement=True)
plan_id = Column(Integer, ForeignKey('action_plans.id'), nullable=False)
metric_type = Column(String(50), nullable=True) # sales_7d / price_rank / conversion
before_val = Column(Float)
after_val = Column(Float)
measured_at = Column(DateTime)
verdict = Column(String(20)) # effective / neutral / backfired
created_at = Column(DateTime, default=datetime_now)
user_id = Column(String(100))
session_id = Column(String(100))
service_type = Column(String(50), nullable=False) # openai, ollama, etc.
model_name = Column(String(100))
prompt_tokens = Column(Integer, default=0)
completion_tokens = Column(Integer, default=0)
total_tokens = Column(Integer, default=0)
cost_usd = Column(Float, default=0.0)
request_type = Column(String(50)) # generation, analysis, etc.
status = Column(String(20), default='success') # success, failed
error_message = Column(Text)
duration_ms = Column(Float)
created_at = Column(DateTime, default=datetime.now)
plan = relationship("ActionPlan", backref="outcomes")
def to_dict(self):
return {
'id': self.id,
'user_id': self.user_id,
'session_id': self.session_id,
'service_type': self.service_type,
'model_name': self.model_name,
'prompt_tokens': self.prompt_tokens,
'completion_tokens': self.completion_tokens,
'total_tokens': self.total_tokens,
'cost_usd': self.cost_usd,
'request_type': self.request_type,
'status': self.status,
'error_message': self.error_message,
'duration_ms': self.duration_ms,
'created_at': self.created_at.isoformat() if self.created_at else None
}
class AgentStrategyWeights(Base):
"""
Agent strategy weights (OpenClaw learning accumulation).
Index: strategy_key for fast updates/query.
"""
__tablename__ = 'agent_strategy_weights'
id = Column(Integer, primary_key=True, autoincrement=True)
strategy_key = Column(String(100), unique=True, nullable=False) # e.g. price_cut_when_gap_gt_5pct
weight = Column(Float, default=1.0)
success_cnt = Column(Integer, default=0)
fail_cnt = Column(Integer, default=0)
updated_at = Column(DateTime, default=datetime_now)
__table_args__ = (
Index('idx_strategy_key', 'strategy_key'),
)
__all__ = [
"AgentContext", "ActionPlan", "ActionOutcome", "AgentStrategyWeights",
"AIGenerationHistory", "AIPromptTemplate", "AIUsageTracking"
]