2026 年 5 月 AI 行业大变局：Agent 融资爆发、Gemini 全生态升级、Claude 冲击企业高风险场景

pip install anthropic python-dotenv

ANTHROPIC_API_KEY=sk-ant-xxxxx

import os
import json
import asyncio
from dotenv import load_dotenv
from anthropic import AsyncAnthropic

load_dotenv()
client = AsyncAnthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

MODEL = "claude-opus-4-8-20260501"  # Claude Opus 4.8 模型 ID，发布后替换为实际值

# ---------- 子 Agent 定义 ----------

SUB_AGENT_PROMPTS = {
    "policy_check": """你是合规政策检查 Agent。
检查输入文本是否违反以下政策：
- 不得包含未披露的利益冲突
- 不得使用误导性统计数据
- 不得遗漏重大风险提示

输出 JSON 格式：
{"violations": [{"rule": "规则名", "detail": "具体描述", "severity": "high/medium/low"}], "summary": "一句话总结"}""",

    "data_accuracy": """你是数据准确性验证 Agent。
检查输入文本中的所有数值、百分比、时间声明：
- 数值是否有明确来源
- 百分比计算是否自洽
- 时间声明是否逻辑一致

输出 JSON 格式：
{"issues": [{"claim": "原文声明", "problem": "问题描述", "confidence": "high/medium/low"}], "summary": "一句话总结"}""",

    "risk_flag": """你是风险标识 Agent。
从输入文本中识别所有风险相关表述，评估：
- 风险是否被充分披露
- 是否存在隐性风险未被提及
- 风险措辞是否合规（不得使用绝对化保证语言）

输出 JSON 格式：
{"risks": [{"description": "风险描述", "disclosed": true/false, "severity": "high/medium/low"}], "summary": "一句话总结"}""",
}

# ---------- 子 Agent 执行 ----------

async def run_sub_agent(name: str, prompt_template: str, document: str) -> dict:
    response = await client.messages.create(
        model=MODEL,
        max_tokens=1024,
        system=prompt_template,
        messages=[{"role": "user", "content": document}],
    )
    text = response.content[0].text
    # 尝试解析 JSON，失败则返回原始文本
    try:
        return {"agent": name, "result": json.loads(text)}
    except json.JSONDecodeError:
        return {"agent": name, "result": text, "raw": True}

# ---------- 父 Agent 汇总 ----------

PARENT_PROMPT = """你是合规审查汇总 Agent。
你将收到三个子 Agent 的审查结果：policy_check、data_accuracy、risk_flag。
请综合所有发现，输出最终审查报告，格式如下：

## 合规审查报告

### 整体评级：PASS / CONDITIONAL_PASS / FAIL

### 各维度发现
- 政策合规：...
- 数据准确性：...
- 风险披露：...

### 必须修改项（按优先级排序）
1. ...

### 建议修改项
1. ...

### 审查备注
任何不确定的判断必须在此标注，不得隐瞒。"""

async def run_parent_agent(sub_results: list[dict], document: str) -> str:
    results_text = json.dumps(sub_results, ensure_ascii=False, indent=2)
    response = await client.messages.create(
        model=MODEL,
        max_tokens=2048,
        system=PARENT_PROMPT,
        messages=[{"role": "user", "content": f"原文档：\n{document}\n\n子 Agent 审查结果：\n{results_text}"}],
    )
    return response.content[0].text

# ---------- 主流程 ----------

async def compliance_review(document: str) -> str:
    # 三个子 Agent 并行执行
    tasks = [
        run_sub_agent(name, prompt, document)
        for name, prompt in SUB_AGENT_PROMPTS.items()
    ]
    sub_results = await asyncio.gather(*tasks)

    # 父 Agent 汇总
    final_report = await run_parent_agent(sub_results, document)
    return final_report

# ---------- 运行示例 ----------

if __name__ == "__main__":
    sample_doc = """
    本基金过去三年年化收益率达 28.5%，远超市场平均水平。
    我们的投资策略确保稳健增长，没有任何重大风险需要担忧。
    管理团队中 3 位合伙人来自被投资企业董事会，这增强了我们的行业洞察力。
    """

    report = asyncio.run(compliance_review(sample_doc))
    print(report)

python compliance_agent.py