通信成本、延迟与收益对照

mkdir l6-cost
cd l6-cost

"""用预置用量、时长与回答记录计算协作取舍。"""
from dataclasses import dataclass
from decimal import Decimal

INPUT_RATE = Decimal("1")   # 假设计费单位/千个输入token
OUTPUT_RATE = Decimal("2")  # 假设计费单位/千个输出token

@dataclass(frozen=True)
class Call:
    id: str
    input_tokens: int
    output_tokens: int
    seconds: int
    deps: tuple = ()

SINGLE = [Call("answer", 1000, 200, 4)]
PARALLEL = [
    Call("plan", 300, 100, 1),
    Call("order", 900, 200, 4, ("plan",)),
    Call("policy", 800, 200, 3, ("plan",)),
    Call("merge", 600, 150, 2, ("order", "policy")),
]
SERIAL = [PARALLEL[0], PARALLEL[1],
          Call("policy", 800, 200, 3, ("order",)), PARALLEL[3]]
EXPECTED = {"A": "120", "B": "人工核查", "C": "原支付渠道", "D": "需要订单号"}
SINGLE_ANSWERS = {"A": "120", "B": "直接退款", "C": "原支付渠道", "D": "120"}
TEAM_ANSWERS = {"A": "120", "B": "人工核查", "C": "原支付渠道", "D": "120"}

def totals(calls):
    inputs = sum(call.input_tokens for call in calls)
    outputs = sum(call.output_tokens for call in calls)
    fee = (Decimal(inputs) * INPUT_RATE + Decimal(outputs) * OUTPUT_RATE) / 1000
    return inputs, outputs, fee

def critical_path(calls):
    finish, remaining = {}, list(calls)
    if len({call.id for call in calls}) != len(calls):
        raise ValueError("调用编号重复")
    while remaining:
        ready = [call for call in remaining if set(call.deps) <= set(finish)]
        if not ready:
            raise ValueError("缺少依赖或存在循环")
        for call in ready:
            finish[call.id] = max((finish[dep] for dep in call.deps), default=0) + call.seconds
            remaining.remove(call)
    return max(finish.values(), default=0)

def passed(answers):
    # 对固定记录逐条核对，不调用模型，也不给模型输出伪造评分。
    return sum(answers.get(key) == value for key, value in EXPECTED.items())

def allowed(calls, limit):
    return totals(calls)[2] <= Decimal(str(limit))

def main():
    for name, calls, answers in [("单Agent", SINGLE, SINGLE_ANSWERS),
                                ("串行协作", SERIAL, TEAM_ANSWERS),
                                ("并行协作", PARALLEL, TEAM_ANSWERS)]:
        inputs, outputs, fee = totals(calls)
        print(f"{name}：输入={inputs}；输出={outputs}；费用={fee:.2f}；路径={critical_path(calls)}秒；固定记录通过={passed(answers)}/4")
    print(f"并行总工作时长：{sum(call.seconds for call in PARALLEL)}秒")
    print("费用上限3.00允许并行协作：" + str(allowed(PARALLEL, "3.00")))
    extra = [key for key, value in EXPECTED.items()
             if TEAM_ANSWERS.get(key) == value and SINGLE_ANSWERS.get(key) != value]
    failures = [key for key, value in EXPECTED.items() if TEAM_ANSWERS.get(key) != value]
    print("增加的通过样例：" + ",".join(extra) + "；仍失败的样例：" + ",".join(failures))

if __name__ == "__main__":
    main()

ls
python -X utf8 cost_demo.py

单Agent：输入=1000；输出=200；费用=1.40；路径=4秒；固定记录通过=2/4
串行协作：输入=2600；输出=650；费用=3.90；路径=10秒；固定记录通过=3/4
并行协作：输入=2600；输出=650；费用=3.90；路径=7秒；固定记录通过=3/4
并行总工作时长：10秒
费用上限3.00允许并行协作：False
增加的通过样例：B；仍失败的样例：D

费用 = (总输入token × 输入千token单价 + 总输出token × 输出千token单价) / 1000
单Agent = (1000 × 1 + 200 × 2) / 1000 = 1.40
协作 = ((300+900+800+600) × 1 + (100+200+200+150) × 2) / 1000
     = (2600 + 1300) / 1000 = 3.90

单次调用完成时间 = 所有依赖的最晚完成时间 + 本次调用时长
并行关键路径 = 1 + max(4, 3) + 2 = 7秒
并行总工作量 = 1 + 4 + 3 + 2 = 10秒

python -X utf8 -m unittest -v test_cost_demo.py