Action
SerializationMixin：
Pydantic在序列化一个对象时，会将其序列化为其最顶层父类的形式，而不是它实际子类的形式。这意味着，如果你有一个父类和一个继承自该父类的子类，当你将子类的实例序列化时，得到的字典将只包含父类的字段，而不会包含子类特有的字段。同样，在反序列化时，Pydantic也无法根据数据内容自动选择正确的子类来实例化，而是只能实例化父类。
因此定义了一个名为SerializationMixin的Python类，用于在Pydantic模型中实现多态序列化和反序列化的混合类
@classmethod
def __get_pydantic_core_schema__(
cls, source: type["SerializationMixin"], handler: Callable[[Any], core_schema.CoreSchema]
) -> core_schema.CoreSchema:
# 调用传入的handler函数，获取模型的默认核心架构
schema = handler(source)
# 保存原始的核心架构引用，这个引用是Pydantic用于识别模型的一个唯一标识
og_schema_ref = schema["ref"]
# 在原始的核心架构引用后面添加一个后缀`:mixin`，以便在序列化和反序列化过程中能够识别这个被修改过的架构
schema["ref"] += ":mixin"
# 创建一个验证器函数，它将在序列化和反序列化过程中被调用，先于标准的Pydantic验证器执行
return core_schema.no_info_before_validator_function(
cls.__deserialize_with_real_type__, # 这个类方法将在反序列化过程中被调用
schema=schema, # 修改后的核心架构
ref=og_schema_ref, # 原始的核心架构引用
serialization=core_schema.wrap_serializer_function_ser_schema(cls.__serialize_add_class_type__), # 包装序列化函数
)
覆盖了Pydantic的__get_pydantic_core_schema__方法，用于自定义模型的序列化和反序列化过程
@classmethod
def __serialize_add_class_type__(
cls,
value,
handler: core_schema.SerializerFunctionWrapHandler,
) -> Any:
# 调用传入的handler函数，这个函数是Pydantic用于序列化模型的默认函数
ret = handler(value)
# 检查当前类是否有子类，如果没有子类，说明它是一个具体的子类而不是基类
if not len(cls.__subclasses__()):
# 只有具体的子类才添加`__module_class_name`字段，这个字段包含了子类的全限定类名
ret["__module_class_name"] = f"{cls.__module__}.{cls.__qualname__}"
# 返回修改后的字典，这个字典将包含额外的类型信息
return ret在序列化过程中被调用，目的是在序列化过程中添加额外的类型信息，以便在反序列化时能够恢复正确的子类类型
@classmethod
def __deserialize_with_real_type__(cls, value: Any):
# 如果传入的值不是字典类型，直接返回该值，因为只有字典类型的值才可能包含序列化的模型数据
if not isinstance(value, dict):
return value
# 如果当前类不是多态基类，或者有子类且序列化的数据中没有`__module_class_name`字段，
# 直接返回传入的值，不进行特殊处理
if not cls.__is_polymorphic_base or (len(cls.__subclasses__()) and "__module_class_name" not in value):
return value
# 从序列化的数据中获取`__module_class_name`字段的值，这个值是子类的全限定类名
module_class_name = value.get("__module_class_name", None)
# 如果没有找到`__module_class_name`字段，抛出ValueError异常
if module_class_name is None:
raise ValueError("Missing field: __module_class_name")
# 从`__subclasses_map__`中获取与全限定类名对应的类类型
class_type = cls.__subclasses_map__.get(module_class_name, None)
# 如果没有找到对应的类类型，抛出TypeError异常
if class_type is None:
raise TypeError(f"Trying to instantiate {module_class_name} which not defined yet.")
# 使用找到的类类型和传入的数据来实例化子类
return class_type(**value)
目的是在反序列化过程中使用之前序列化时保存的类型信息来实例化正确的子类
def __init_subclass__(cls, is_polymorphic_base: bool = False, **kwargs):
# 将is_polymorphic_base参数设置为子类的多态基类标志
cls.__is_polymorphic_base = is_polymorphic_base
# 将当前子类添加到__subclasses_map__映射中，以便在反序列化时能够找到正确的子类
# __subclasses_map__是一个字典，用于存储子类与其全限定类名的映射关系
cls.__subclasses_map__[f"{cls.__module__}.{cls.__qualname__}"] = cls
# 调用基类的__init_subclass__方法，以便子类可以继承基类的其他设置
super().__init_subclass__(**kwargs)在定义子类时被自动调用，确保在创建子类时子类被正确地注册到__subclasses_map__中，从而在__deserialize_with_real_type__能找到正确的与全限定类名字、对应的类类型
ContexMixin：
用于处理上下文、配置和大型语言模型的相关操作。这个混入类提供了对上下文和配置的访问和设置方法，以及一个大型语言模型的实例
class ContextMixin(BaseModel):
"""Mixin class for context and config"""
# 定义了一个ConfigDict类型的模型配置，允许任意类型的字段。
model_config = ConfigDict(arbitrary_types_allowed=True, extra="allow")
# Pydantic has bug on _private_attr when using inheritance, so we use private_* instead
# - https://github.com/pydantic/pydantic/issues/7142
# - https://github.com/pydantic/pydantic/issues/7083
# - https://github.com/pydantic/pydantic/issues/7091
# Env/Role/Action will use this context as private context, or use self.context as public context
# 用于存储一个Context类型的私有上下文。
private_context: Optional[Context] = Field(default=None, exclude=True)
# Env/Role/Action will use this config as private config, or use self.context.config as public config
# 用于存储一个Config类型的私有配置。
private_config: Optional[Config] = Field(default=None, exclude=True)
# Env/Role/Action will use this llm as private llm, or use self.context._llm instance
# 用于存储一个BaseLLM类型的私有大型语言模型。
private_llm: Optional[BaseLLM] = Field(default=None, exclude=True)定义了三个私有字段：private_context、private_config和private_llm，用于存储私有上下文、配置和大型语言模型。
@model_validator(mode="after")
def validate_context_mixin_extra(self):
self._process_context_mixin_extra()
return self
def _process_context_mixin_extra(self):
"""Process the extra field"""
# 从model_extra字段中获取额外的参数，这是一个字典，包含了模型创建时传入的所有额外字段
kwargs = self.model_extra or {}
# 如果context键存在于字典中，则使用self.set_context方法来设置上下文
self.set_context(kwargs.pop("context", None))
# 如果config键存在于字典中，则使用self.set_config方法来设置配置
self.set_config(kwargs.pop("config", None))
# 如果llm键存在于字典中，则使用self.set_llm方法来设置大型语言模型
self.set_llm(kwargs.pop("llm", None))确保在模型实例化后，能够正确地设置上下文、配置和大型语言模型
def set(self, k, v, override=False):
"""Set attribute"""
# 这个方法用于设置模型的属性。如果override参数为True或者当前没有这个属性的值，
# 它将设置这个属性。
if override or not self.__dict__.get(k):
self.__dict__[k] = v
def set_context(self, context: Context, override=True):
"""Set context"""
# 这个方法用于设置上下文。如果override参数为True或者当前没有上下文，
# 它将设置私有上下文。
self.set("private_context", context, override)
def set_config(self, config: Config, override=False):
"""Set config"""
# 这个方法用于设置配置。如果override参数为True或者当前没有配置，
# 它将设置私有配置。如果配置不为None，它还会初始化LLM。
self.set("private_config", config, override)
if config is not None:
_ = self.llm # init llm
def set_llm(self, llm: BaseLLM, override=False):
"""Set llm"""
# 这个方法用于设置大型语言模型。如果override参数为True或者当前没有大型语言模型，
# 它将设置私有大型语言模型。
self.set("private_llm", llm, override)
@property
def config(self) -> Config:
"""Role config: role config > context config"""
# 这个属性用于获取配置。它首先检查是否有私有配置，如果没有，
# 则从上下文获取配置。
if self.private_config:
return self.private_config
return self.context.config
@config.setter
def config(self, config: Config) -> None:
"""Set config"""
# 这个属性设置器用于设置配置。
self.set_config(config)
@property
def context(self) -> Context:
"""Role context: role context > context"""
# 这个属性用于获取上下文。它首先检查是否有私有上下文，如果没有，
# 则创建一个新的上下文实例。
if self.private_context:
return self.private_context
return Context()
@context.setter
def context(self, context: Context) -> None:
"""Set context"""
# 这个属性设置器用于设置上下文。
self.set_context(context)
@property
def llm(self) -> BaseLLM:
"""Role llm: if not existed, init from role.config"""
# 这个属性用于获取大型语言模型（LLM）。如果私有LLM不存在，
# 它会从角色的配置中初始化一个LLM。
if not self.private_llm:
self.private_llm = self.context.llm_with_cost_manager_from_llm_config(self.config.llm)
return self.private_llm
@llm.setter
def llm(self, llm: BaseLLM) -> None:
"""Set llm"""
# 这个属性设置器用于设置LLM。
self.private_llm = llm不断设置和获取那三个属性
ProjectRepo：
继承自FileRepository类，FileRepository提供了一系列与文件操作相关的方法，这些操作包括保存文件、获取文件依赖、获取已更改的依赖项、获取文件内容、列出所有文件、保存文档、删除文件等。这些方法主要用于处理存储在Git仓库中的文件，这些功能对于维护Git仓库中的文件和跟踪文件之间的依赖关系非常有用。
GitRepository对象提供了一个全面的接口，用于在Python中与Git仓库交互，包括管理仓库、跟踪变更、提交更改、获取文件列表等
class ProjectRepo(FileRepository):
def __init__(self, root: str | Path | GitRepository):
# 如果传入的root参数是字符串或Path对象，则创建一个新的GitRepository对象
# 如果传入的root参数是一个已存在的GitRepository对象，则直接使用该对象
# 如果传入的root参数无效，则抛出一个ValueError
if isinstance(root, str) or isinstance(root, Path):
git_repo_ = GitRepository(local_path=Path(root))
elif isinstance(root, GitRepository):
git_repo_ = root
else:
raise ValueError("Invalid root")
# 调用父类的构造函数，初始化FileRepository对象
# git_repo_是GitRepository对象，relative_path是相对于Git仓库根目录的相对路径
super().__init__(git_repo=git_repo_, relative_path=Path("."))
# 初始化ProjectRepo的属性
self._git_repo = git_repo_
self.docs = DocFileRepositories(self._git_repo)
self.resources = ResourceFileRepositories(self._git_repo)
self.tests = self._git_repo.new_file_repository(relative_path=TEST_CODES_FILE_REPO)
self.test_outputs = self._git_repo.new_file_repository(relative_path=TEST_OUTPUTS_FILE_REPO)
self._srcs_path = None
self.code_files_exists()
def __str__(self):
# 返回一个字符串表示，包括Git仓库的工作目录、文档、资源、测试代码和测试输出
repo_str = f"ProjectRepo({self._git_repo.workdir})"
docs_str = f"Docs({self.docs.all_files})"
srcs_str = f"Srcs({self.srcs.all_files})"
return f"{repo_str}\n{docs_str}\n{srcs_str}"
@property
async def requirement(self):
# 异步获取REQUIREMENT_FILENAME文件的内容，通常是一个依赖列表
return await self.docs.get(filename=REQUIREMENT_FILENAME)
@property
def git_repo(self) -> GitRepository:
# 返回与Git仓库交互的GitRepository对象
return self._git_repo
@property
def workdir(self) -> Path:
# 返回Git仓库的工作目录的路径
return Path(self.git_repo.workdir)
@property
def srcs(self) -> FileRepository:
# 返回一个用于访问源代码文件的FileRepository对象
if not self._srcs_path:
raise ValueError("Call with_srcs first.")
return self._git_repo.new_file_repository(self._srcs_path)
def code_files_exists(self) -> bool:
# 检查Git仓库中是否存在代码文件
git_workdir = self.git_repo.workdir
src_workdir = git_workdir / git_workdir.name
if not src_workdir.exists():
return False
code_files = self.with_src_path(path=git_workdir / git_workdir.name).srcs.all_files
if not code_files:
return False
return bool(code_files)
def with_src_path(self, path: str | Path) -> ProjectRepo:
# 设置源代码文件的路径，并返回当前ProjectRepo对象
try:
self._srcs_path = Path(path).relative_to(self.workdir)
except ValueError:
self._srcs_path = Path(path)
return self
@property
def src_relative_path(self) -> Path | None:
# 返回源代码文件的相对路径
return self._srcs_path
pass
