Katharos 是一个 Python 的功能编程和并发库。Katharos 将代数抽象（Functor、Applicative、Monad、Semigroup、Monoid）与像 Maybe、Result、ImmutableList 和 IO 这样的具体类型相结合，使用基于相同功能核心的消息传递并发。这两个部分共享一个想法：将错误、效果和并发通信建模为可组合的、安全的类型值。并发传递返回一个 Result，因此“通道关闭”是你处理的内容，而不是你捕获的异常。 安装：pip install katharos 或使用 uv uv add katharos 之前：分散的 None 检查和异常处理： user = find_user(user_id) if user is None: return None account = find_account(user) if account is None: return None return account.discount 之后：在 Nothing 上干净地短路的 do-notation： from katharos.types import Maybe from katharos.syntax_sugar import do, DoBlock @do(Maybe) def lookup_discount(user_id: int) -> DoBlock[Maybe, float]: user = yield find_user(user_id) account = yield find_account(user) return account.discount # Just(0.15) 或 Nothing() 之前：嵌套的 try/except 来传播错误： def process(raw: str) -> int: try: n = parse_int(raw) except ValueError as e: raise RuntimeError("坏输入") from e try: return validate_positive(n) except ValueError as e: raise RuntimeError("坏值") from e 之后：将错误作为值，使用 | 链接： from katharos.types import Result def process(raw: str) -> Result[Exception, int]: return parse_int(raw) | validate_positive # 失败自动短路 更多示例 处理可选值而不需要 None 检查： from katharos.types import Maybe result = Maybe[int].Just(5) | (lambda x: Maybe[int].Just(x * 2)) # Just(10) nothing = Maybe[int].Nothing() | (lambda x: Maybe[int].Just(x * 2)) # Nothing() 将错误建模为值而不是异常： from katharos.types import Result def parse_int(s: str) -> Result[ValueError, int]: try: return Result.Success(int(s)) except ValueError as e: return Result.Failure(e) parse_int("42").fmap(lambda n: n * 2) # Success(84) parse_int("??").fmap(lambda n: n * 2) # Failure(...) 跳过模板代码使用 Result.catch： Result.catch 将引发异常的函数转为返回 Result 的函数，无需手动 try/except。只有声明的异常类型会变为 Failure；捕获的异常保留其回溯，因此你仍然可以找到失败的行。 import traceback from katharos.types import Result @Result.catch(ValueError) def parse_int(s: str) -> int: return int(s) parse_int("42") # 成功 (42) parse_int("??") # 失败(ValueError("无效的字面量对于 int() 使用基数 10: '??'")) failure = parse_int("??") if failure.is_failure(): traceback.print_exception(failure.error) # 完整回溯，指向失败行 使用 Semigroup 运算符合并值： from katharos.types import ImmutableList ImmutableList([1, 2]) @ ImmutableList([3, 4]) # ImmutableList([1, 2, 3, 4]) Do-notation do-notation 与任何 monad 都适用：Maybe、Result、IO、ImmutableList 和你的自定义 monads。每个 yield 解开单子值： from katharos.syntax_sugar import do, DoBlock from katharos.types import Result def parse_positive(x: int) -> Result[ValueError, int]: return Result.Success(x) if x > 0 else Result.Failure(ValueError(f"{x} 不是正数")) # 干净、命令式风格的单子代码 @do(Result) def do_block() -> DoBlock[Result, int]: x: int = yield parse_positive(5) y: int = yield parse_positive(3) return x + y print(do_block()) # Success(8) 并发 Katharos 提供基于相同功能核心的消息传递并发，支持多个并发模型。第一个可用模型是 Go 风格的 CSP：并发方式启动工作，采用 go（如 Go 的 go f(x)），通过类型通道通信，并（关键地）接收值作为 Result，因此关闭或超时的通道是你模式匹配的值，而不是你在 try 中包装的异常： from katharos.concurrency.csp import csp ch = csp.Channel[int](capacity=1) csp.go(ch.send, 42) # 像 Go 的 `go f(x)` 一样并发运行工作 ch.recv() # Success(42) ch.close() ch.recv() # Failure(ChannelClosedError(...)): closure 是一个值，而不是 raise 作为上下文管理器使用时，go 变成结构化并发作用域，在块退出之前将内部启动的所有任务加入： from katharos.concurrency.csp import csp with csp.go: # 作用域等待内部启动的所有工作 csp.go(worker, 1) csp.go(worker, 2) # 两个工作线程此时已经完成 每个并发模型都绑定到可交换的 BaseThreadingBackend（默认使用标准线程），并且 csp 运行时会自动提供它，因此你可以将工作重定向到不同的后端。