103 lines
3.2 KiB
Python
103 lines
3.2 KiB
Python
"""运行观察的共同接入合同；内容用途从本作者的真实任务读取。"""
|
|
|
|
from dataclasses import replace
|
|
from typing import Literal
|
|
from uuid import UUID
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, StrictInt
|
|
|
|
from muse.作者经验.接口 import (
|
|
参考挑选反馈请求,
|
|
改进定位,
|
|
经验错误,
|
|
规则启停请求,
|
|
规则改进请求,
|
|
规则验证请求,
|
|
运行观察请求,
|
|
)
|
|
from muse.共享.调用身份 import 内容用途
|
|
|
|
|
|
class 运行来源请求(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
task_id: UUID
|
|
attempt_id: UUID
|
|
business_kind: Literal[
|
|
"", "planning_candidate", "reference_analysis", "review_report", "revision_result"
|
|
] = ""
|
|
business_ref: str = ""
|
|
business_revision: StrictInt = Field(default=0, ge=0)
|
|
|
|
|
|
class 运行观察保存请求(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
command_id: str = Field(min_length=1)
|
|
observation: 运行观察请求
|
|
|
|
|
|
def _运行身份(app, identity, task_id):
|
|
task = app.任务运行.读取任务(str(task_id))
|
|
if task.作者 != identity.作者:
|
|
raise 经验错误("SCOPE_DENIED", "运行观察的来源任务不属于当前作者")
|
|
return replace(identity, 内容用途=内容用途(task.冻结输入["内容用途"]))
|
|
|
|
|
|
def 读取运行来源(app, identity, request: 运行来源请求):
|
|
return app.要求作者经验().读取运行观察来源(
|
|
_运行身份(app, identity, request.task_id),
|
|
str(request.task_id),
|
|
str(request.attempt_id),
|
|
业务类型=request.business_kind,
|
|
业务引用=request.business_ref,
|
|
业务版本=request.business_revision,
|
|
)
|
|
|
|
|
|
def 保存运行观察(app, identity, request: 运行观察保存请求):
|
|
return app.要求作者经验().记录运行观察(
|
|
_运行身份(app, identity, request.observation.sources[0].task_id),
|
|
request.command_id,
|
|
request.observation,
|
|
)
|
|
|
|
|
|
class 规则改进输入(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
proposal: 规则改进请求
|
|
|
|
|
|
class 规则改进验证输入(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
command_id: str = Field(min_length=1)
|
|
selection: 改进定位
|
|
request: 规则验证请求
|
|
|
|
|
|
class 规则改进启停输入(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
command_id: str = Field(min_length=1)
|
|
selection: 改进定位
|
|
request: 规则启停请求
|
|
|
|
|
|
class 参考挑选输入(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
command_id: str = Field(min_length=1)
|
|
selection: 参考挑选反馈请求
|
|
|
|
|
|
def 保存参考挑选反馈(app, identity, request: 参考挑选输入):
|
|
return app.要求作者经验().记录参考挑选反馈(identity, request.command_id, request.selection)
|
|
|
|
|
|
def 读取挑选信号视图(app, identity, *, 领域=""):
|
|
return app.要求作者经验().读取参考挑选信号(identity, 领域=领域)
|
|
|
|
|
|
def 读取推断建议视图(app, identity, *, 领域=""):
|
|
return app.要求作者经验().汇总推断偏好建议(identity, 领域=领域)
|
|
|
|
|
|
def 读取复利度量视图(app, identity):
|
|
return app.要求作者经验().读取复利度量(identity)
|