110 lines
4.2 KiB
Python
110 lines
4.2 KiB
Python
from django.db import models
|
|
|
|
|
|
class AgentAuditLog(models.Model):
|
|
"""
|
|
保存一次 Agent 执行的完整审计快照。
|
|
|
|
该模型是“系统可解释性”的核心：
|
|
- 对话页负责触发执行
|
|
- Agent Core 负责生成结果
|
|
- Audit 模型负责长期保存输入、引用、工具调用和模型输出
|
|
"""
|
|
# 审计状态需要同时服务数据库检索和前端展示。
|
|
STATUS_SUCCESS = "success"
|
|
STATUS_FAILED = "failed"
|
|
|
|
scenario_id = models.CharField(max_length=100, db_index=True)
|
|
scenario_name = models.CharField(max_length=200, blank=True)
|
|
batch_id = models.CharField(max_length=64, blank=True, db_index=True)
|
|
conversation_id = models.CharField(max_length=64, blank=True, db_index=True)
|
|
product_name = models.CharField(max_length=255, blank=True, db_index=True)
|
|
user_input = models.TextField()
|
|
retrieved_chunks = models.JSONField(default=list, blank=True)
|
|
tool_calls = models.JSONField(default=list, blank=True)
|
|
structured_output = models.JSONField(default=dict, blank=True)
|
|
final_answer = models.TextField(blank=True)
|
|
raw_output = models.TextField(blank=True)
|
|
model_name = models.CharField(max_length=100, blank=True)
|
|
latency_ms = models.PositiveIntegerField(default=0)
|
|
status = models.CharField(max_length=20, default=STATUS_SUCCESS, db_index=True)
|
|
error_message = models.TextField(blank=True)
|
|
created_at = models.DateTimeField(auto_now_add=True, db_index=True)
|
|
|
|
class Meta:
|
|
ordering = ["-created_at"]
|
|
|
|
def __str__(self) -> str:
|
|
return f"{self.scenario_name or self.scenario_id} #{self.pk}"
|
|
|
|
def get_status_display_text(self) -> str:
|
|
"""返回更适合页面展示的中文状态。"""
|
|
return {
|
|
self.STATUS_SUCCESS: "执行成功",
|
|
self.STATUS_FAILED: "执行失败",
|
|
}.get(self.status, self.status)
|
|
|
|
def get_user_input_summary(self, max_length: int = 28) -> str:
|
|
"""在列表页展示用户输入摘要，避免长文本撑破表格。"""
|
|
if len(self.user_input) <= max_length:
|
|
return self.user_input
|
|
return f"{self.user_input[:max_length]}..."
|
|
|
|
|
|
class DemoBusinessRecord(models.Model):
|
|
"""
|
|
演示用业务记录表。
|
|
|
|
该表不直接参与页面主流程，而是供内置工具 `query_demo_records`
|
|
查询，证明 Agent 除知识库外也可以结合结构化业务数据。
|
|
"""
|
|
scenario_id = models.CharField(max_length=100, db_index=True)
|
|
record_type = models.CharField(max_length=100, db_index=True)
|
|
title = models.CharField(max_length=255)
|
|
payload = models.JSONField(default=dict, blank=True)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
|
|
class Meta:
|
|
ordering = ["-created_at"]
|
|
|
|
def __str__(self) -> str:
|
|
return self.title
|
|
|
|
|
|
class NotificationRecord(models.Model):
|
|
"""
|
|
飞书通知留痕。
|
|
|
|
首版只保存离线通知载荷与结果状态，不直接依赖真实飞书网络。
|
|
"""
|
|
|
|
STATUS_PENDING = "pending"
|
|
STATUS_SENT = "sent"
|
|
STATUS_FAILED = "failed"
|
|
|
|
batch_id = models.CharField(max_length=64, db_index=True)
|
|
conversation_id = models.CharField(max_length=64, db_index=True)
|
|
product_name = models.CharField(max_length=255, blank=True, db_index=True)
|
|
trigger_source = models.CharField(max_length=64, blank=True)
|
|
notify_reason = models.CharField(max_length=32, db_index=True)
|
|
owner_role = models.CharField(max_length=100, blank=True)
|
|
feishu_user_id = models.CharField(max_length=100, blank=True)
|
|
message_status = models.CharField(max_length=32, default=STATUS_PENDING, db_index=True)
|
|
web_detail_url = models.URLField(blank=True)
|
|
receipt = models.JSONField(default=dict, blank=True)
|
|
created_at = models.DateTimeField(auto_now_add=True, db_index=True)
|
|
|
|
class Meta:
|
|
ordering = ["-created_at"]
|
|
|
|
def __str__(self) -> str:
|
|
return f"{self.notify_reason}:{self.batch_id}"
|
|
|
|
def get_message_status_display_text(self) -> str:
|
|
"""返回通知状态的中文展示文案。"""
|
|
return {
|
|
self.STATUS_PENDING: "处理中",
|
|
self.STATUS_SENT: "已发送",
|
|
self.STATUS_FAILED: "失败",
|
|
}.get(self.message_status, self.message_status)
|