在 AWS 上构建生产级多智能体系统的 6 条经验

import json
import boto3
import jsonschema
from botocore.config import Config

bedrock = boto3.client("bedrock-runtime", config=Config(read_timeout=300))   # 长答案会超过默认的 60 秒

class BudgetExceeded(Exception): ...
class SchemaInvalid(Exception): ...

def llm_call(task, model, system, user, schema, validate=None, max_repairs=2, max_tokens=6000):
    """一个范围很窄的任务：解码时强制 Schema，然后代码校验，最多修复 2 次。"""
    validator = jsonschema.Draft202012Validator(schema)
    messages = [{"role": "user", "content": [{"text": user}]}]
    for _ in range(max_repairs + 1):
        resp = bedrock.converse(
            modelId=model, system=[{"text": system}], messages=messages,
            inferenceConfig={"maxTokens": max_tokens},
            outputConfig={"textFormat": {"type": "json_schema", "structure": {"jsonSchema": {
                "schema": json.dumps(schema), "name": "result", "description": "result"}}}})
        if resp["stopReason"] == "max_tokens":
            raise BudgetExceeded(task)               # 答案被截断：没有可修复的内容
        text = "".join(c.get("text", "") for c in resp["output"]["message"]["content"])
        try:
            obj = json.loads(text)
            errors = [f"{'/'.join(map(str, e.absolute_path)) or '(root)'}: {e.message}"
                      for e in validator.iter_errors(obj)]
        except json.JSONDecodeError as e:
            obj, errors = None, [f"(root): not valid JSON ({e.msg})"]
        if not errors and validate:
            errors = validate(obj)                   # 业务规则和依据检查，在代码中完成
        if not errors:
            return obj
        messages += [{"role": "assistant", "content": [{"text": text}]},
                     {"role": "user", "content": [{"text": "Your answer has these problems. "
                      "Return the full corrected JSON.\n- " + "\n- ".join(errors)}]}]
    raise SchemaInvalid(f"{task}: {errors[:3]}")

def test_every_model_schema_follows_the_contract():
    seen = collect_task_schemas()        # [(task, schema)]：运行主要 LLM 任务，并替换 llm_call 以记录 Schema

    def walk(node, path):
        if isinstance(node, dict):
            assert not {"minimum", "maximum", "exclusiveMinimum", "exclusiveMaximum"} & set(node), path
            if node.get("type") == "array":
                assert "maxItems" in node, path
            for k, v in node.items():
                walk(v, f"{path}/{k}")
        elif isinstance(node, list):
            for v in node:
                walk(v, path)

    for task, schema in seen:
        walk(schema, task)

    # 在每条提取的需求中，引用的原文排在由它决定的值之前
    req = next(s for t, s in seen if t == "extract_needs")["properties"]["needs"]["items"]["required"]
    for quote, decision in (("quantityText", "quantity"), ("osText", "osFamily"),
                            ("dbEngineText", "dbEngine"), ("deploymentText", "deployment")):
        assert req.index(quote) < req.index(decision)

def need_errors(n):
    """代码检查一条带类型的需求。每个错误连同 JSON 路径返回给模型。"""
    errs, line = [], n["sourceLine"]
    for f in ("osText", "dbEngineText"):
        if n[f] and not grounded(n[f], line):
            errs.append(f"{f}: must be words from sourceLine")
    for k, m in enumerate(n["measures"]):
        if not grounded(m["evidence"], line):
            errs.append(f"measures/{k}/evidence: must be words from sourceLine")
        if m["value"] not in numbers_in(m["evidence"]):
            errs.append(f"measures/{k}/value: {m['value']:g} is not written in its evidence \"{m['evidence']}\"")
    if n["osFamily"] != "not_stated" and not n["osText"]:
        errs.append("osFamily: quote the words naming the operating system in osText, or use not_stated")
    if n["dbEngine"] != "not_stated" and not n["dbEngineText"]:
        errs.append("dbEngine: quote the words naming the engine in dbEngineText, or use not_stated")
    return errs

def ungrounded(requested, line):
    """当声称不匹配中的 'requested' 一侧与该行的请求或计划没有任何共同词时，返回 True。"""
    claim = words(requested)                 # 小写词，去掉停用词
    source = words(json.dumps({k: line.get(k) for k in ("service", "request", "cobraSettings")}))
    return bool(claim) and not (claim & source)

def validate(obj):
    errors = []
    for n, x in enumerate(obj["lines"]):
        x["verdict"] = "fail" if x["mismatches"] else "pass"     # 由代码给出结论，而不是模型
        if x["verdict"] == "fail" and not x["fix"].strip():
            errors.append(f"lines/{n}: a fail needs its mismatches and a fix")
        for k, m in enumerate(x["mismatches"]):
            if ungrounded(m["requested"], lines_by_id[x["itemId"]]):
                errors.append(f"lines/{n}/mismatches/{k}/requested: quote what the line's request "
                              f"or COBRA's settings say; {m['requested'][:60]!r} is not in them")
    return errors