上下文:我正在尝试解决锦标赛安排问题。
我想分析 Timefold 算法生成的解决方案,但我面临序列化问题:
INFO: @ timefold_solver : Validating score: type=<class 'str'>, value=0hard/-1.195230medium/-2.563480soft
INFO: @ timefold_solver : Analysis result: Explanation of score (0hard/-1.19523medium/-2.56348soft):
Constraint matches:
-1.19523medium: constraint (fairAssignmentCountPerTeam) has 1 matches:
-1.19523medium: justified with ([ai.timefold.solver.core.impl.score.stream.collector.LoadBalanceImpl@34afdb84])
-2.56348soft: constraint (evenlyConfrontationCount) has 1 matches:
-2.56348soft: justified with ([ai.timefold.solver.core.impl.score.stream.collector.LoadBalanceImpl@3192c24f])
0: constraint (oneAssignmentPerDatePerTeam) has no matches.
0: constraint (unavailabilityPenalty) has no matches.
INFO: @ timefold_solver : Constraint Analysis: evenlyConfrontationCount, Weight: 0hard/0medium/-1soft, Score: 0hard/0medium/-2.56348soft
INFO: @ timefold_solver : Match Analysis: evenlyConfrontationCount, Score: 0hard/0medium/-2.56348soft, Justification: DefaultConstraintJustification(facts=(<java object 'ai.timefold.solver.core.impl.score.stream.collector.LoadBalanceImpl'>,), impact=HardMediumSoftDecimalScore(init_score=0, hard_score=Decimal('0'), medium_score=Decimal('0'), soft_score=Decimal('-2.56348')))
INFO: @ timefold_solver : Constraint Analysis: fairAssignmentCountPerTeam, Weight: 0hard/-1medium/0soft, Score: 0hard/-1.19523medium/0soft
INFO: @ timefold_solver : Match Analysis: fairAssignmentCountPerTeam, Score: 0hard/-1.19523medium/0soft, Justification: DefaultConstraintJustification(facts=(<java object 'ai.timefold.solver.core.impl.score.stream.collector.LoadBalanceImpl'>,), impact=HardMediumSoftDecimalScore(init_score=0, hard_score=Decimal('0'), medium_score=Decimal('-1.19523'), soft_score=Decimal('0')))
INFO: @ timefold_solver : Constraint Analysis: oneAssignmentPerDatePerTeam, Weight: -1hard/0medium/0soft, Score: 0hard/0medium/0soft
INFO: @ timefold_solver : Constraint Analysis: unavailabilityPenalty, Weight: -1hard/0medium/0soft, Score: 0hard/0medium/0soft
INFO: @ uvicorn.access : 127.0.0.1:59834 - "PUT /schedules/analyze HTTP/1.1" 500
ERROR: @ uvicorn.error : Exception in ASGI application
Traceback (most recent call last):
[...]
File "/Users/user/Desktop/workspace/timefold-quickstarts/python/tournament-scheduling/.venv/lib/python3.11/site-packages/starlette/routing.py", line 72, in app
response = await func(request)
^^^^^^^^^^^^^^^^^^^
File "/Users/user/Desktop/workspace/timefold-quickstarts/python/tournament-scheduling/.venv/lib/python3.11/site-packages/fastapi/routing.py", line 296, in app
content = await serialize_response(
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/user/Desktop/workspace/timefold-quickstarts/python/tournament-scheduling/.venv/lib/python3.11/site-packages/fastapi/routing.py", line 160, in serialize_response
return field.serialize(
^^^^^^^^^^^^^^^^
File "/Users/user/Desktop/workspace/timefold-quickstarts/python/tournament-scheduling/.venv/lib/python3.11/site-packages/fastapi/_compat.py", line 149, in serialize
return self._type_adapter.dump_python(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/user/Desktop/workspace/timefold-quickstarts/python/tournament-scheduling/.venv/lib/python3.11/site-packages/pydantic/type_adapter.py", line 339, in dump_python
return self.serializer.to_python(
^^^^^^^^^^^^^^^^^^^^^^^^^^
pydantic_core._pydantic_core.PydanticSerializationError: Unable to serialize unknown type: <java class 'ai.timefold.solver.core.impl.score.stream.collector.LoadBalanceImpl'>
下面是我的代码(与此问题相关的代码片段):
rest_api.py
async def setup_context(request: Request) -> TournamentSchedule:
json = await request.json()
return TournamentSchedule.model_validate(json,
context={
'teams': {
team['id']: Team.model_validate(team) for
team in json.get('teams', [])
},
'days': {
day['dateIndex']: Day.model_validate(day) for
day in json.get('days', [])
}
})
@app.put("/schedules/analyze")
async def analyze_timetable(tournament_schedule: Annotated[TournamentSchedule, Depends(setup_context)]) -> dict:
# Call the analyze method and log its raw output
analysis_result = solution_manager.analyze(tournament_schedule)
logger.info(f"Analysis result: {analysis_result}")
# Log detailed constraint analyses
for constraint in analysis_result.constraint_analyses:
logger.info(
f"Constraint Analysis: {constraint.constraint_name}, Weight: {constraint.weight}, Score: {constraint.score}")
for match in constraint.matches:
logger.info(
f" Match Analysis: {match.constraint_ref.constraint_name}, Score: {match.score}, Justification: {match.justification}")
return {'constraints': [ConstraintAnalysisDTO(
name=constraint.constraint_name,
weight=constraint.weight,
score=constraint.score,
matches=[
MatchAnalysisDTO(
name=match.constraint_ref.constraint_name,
score=match.score,
justification=match.justification
)
for match in constraint.matches
]
) for constraint in solution_manager.analyze(tournament_schedule).constraint_analyses]}
json_序列化.py
ScoreSerializer = PlainSerializer(lambda score: str(score) if score is not None else None,
return_type=str | None)
def validate_score(v: Any, info: ValidationInfo) -> Any:
logger.info(f"Validating score: type={type(v)}, value={v}")
if isinstance(v, HardMediumSoftDecimalScore) or v is None:
return v
if isinstance(v, str):
return HardMediumSoftDecimalScore.parse(v)
raise ValueError('"score" should be a string')
ScoreValidator = BeforeValidator(validate_score)
class JsonDomainBase(BaseModel):
model_config = ConfigDict(
alias_generator=to_camel,
populate_by_name=True,
from_attributes=True,
)
域.py
class Day(JsonDomainBase):
date_index: int
class Team(JsonDomainBase):
id: Annotated[int, PlanningId]
name: Annotated[str, Field(default=None)]
class UnavailabilityPenalty(JsonDomainBase):
team: Annotated[Team | None,
IdSerializer,
TeamDeserializer,
Field(default=None)]
day: Annotated[Day | None,
IdSerializer,
DayDeserializer,
Field(default=None)]
@planning_entity
class TeamAssignment(JsonDomainBase):
id: Annotated[int, PlanningId]
day: Annotated[Day | None,
IdSerializer,
DayDeserializer,
Field(default=None)]
index_in_day: int
pinned: Annotated[bool, PlanningPin]
team: Annotated[Team | None,
PlanningVariable,
IdSerializer,
TeamDeserializer,
Field(default=None)]
@planning_solution
class TournamentSchedule(JsonDomainBase):
teams: Annotated[list[Team],
ProblemFactCollectionProperty,
ValueRangeProvider]
days: Annotated[list[Day],
ProblemFactCollectionProperty]
unavailability_penalties: Annotated[list[UnavailabilityPenalty],
ProblemFactCollectionProperty]
team_assignments: Annotated[list[TeamAssignment],
PlanningEntityCollectionProperty]
score: Annotated[HardMediumSoftDecimalScore | None,
PlanningScore,
ScoreSerializer,
ScoreValidator,
Field(default=None)]
solver_status: Annotated[SolverStatus | None, Field(default=SolverStatus.NOT_SOLVING)]
约束.py
def unavailability_penalty(constraint_factory: ConstraintFactory) -> Constraint:
return (constraint_factory
.for_each(TeamAssignment)
.group_by(ConstraintCollectors.load_balance(lambda team_assignment: team_assignment.team))
.penalize_decimal(HardMediumSoftDecimalScore.ONE_MEDIUM, lambda balance: balance.unfairness())
.as_constraint("unavailability_penalty"))
def evenly_confrontation_count(constraint_factory: ConstraintFactory) -> Constraint:
return (constraint_factory
.for_each(TeamAssignment)
.join(TeamAssignment,
Joiners.equal(lambda team_assignment: team_assignment.day),
Joiners.less_than(lambda assignment: assignment.team.id))
.group_by(ConstraintCollectors.load_balance(lambda assignment, other_assignment: (assignment.team, other_assignment.team)))
.penalize_decimal(HardMediumSoftDecimalScore.ONE_SOFT, lambda balance: balance.unfairness())
.as_constraint("evenlyConfrontationCount"))
score_analysis.py
class MatchAnalysisDTO(JsonDomainBase):
name: str
score: Annotated[HardMediumSoftDecimalScore, ScoreSerializer]
justification: object
class ConstraintAnalysisDTO(JsonDomainBase):
name: str
weight: Annotated[HardMediumSoftDecimalScore, ScoreSerializer]
matches: list[MatchAnalysisDTO]
score: Annotated[HardMediumSoftDecimalScore, ScoreSerializer]
我该如何解决这个问题?