selfrelease
fad458b2a7
- UIUX 文档：填充 19 个缺口（多主体画像/健康度/AI+看板/增长域/洞察域/创始人端/OODA/助推/商密） - UIUX 文档：插入 6 个新章节（十四~十九），旧章节重编号为二十~三十一，更新目录和交叉引用 - 作业指导书 x5：导航改为 6 域分组，新增 Context Bar/工作模式/Insight Rail/决策线程/多工作区等 UI 概念 - 新建 docs/2-task-uiux.md：50 个代码落地开发任务，按 P0-P6 分优先级 + 8 Sprint 规划 - 后端/前端：大量新增模型、路由、组件（来自之前 Phase 开发）
274 lines
9.9 KiB
Python
274 lines
9.9 KiB
Python
"""Phase 3-4 AI Agent Service 层测试。
|
||
|
||
测试所有依赖 LLM 的 AI Agent service：
|
||
- synergy_matcher: 协同匹配
|
||
- alpha_attribution: Alpha 归因
|
||
- exit_predictor: 退出预测
|
||
- aar_agent: AAR 复盘
|
||
- digital_twin_engine: 数字孪生
|
||
- milestone_agent: 里程碑路径切换
|
||
- talent_agent: 人才流动预测
|
||
- board_agent: 董事会摘要
|
||
- innovation_discoverer: 组合创新
|
||
- knowledge_graph_builder: 知识图谱
|
||
- agent_orchestrator: Agent 编排
|
||
- portfolio_rebalancer: 组合再平衡（纯计算，无 LLM）
|
||
"""
|
||
|
||
import pytest
|
||
|
||
from app.services.synergy_matcher import match_synergy
|
||
from app.services.alpha_attribution import attribute_alpha
|
||
from app.services.exit_predictor import predict_exit
|
||
from app.services.aar_agent import generate_aar, check_aar_triggers
|
||
from app.services.digital_twin_engine import build_twin_model, simulate_scenario
|
||
from app.services.milestone_agent import suggest_path_switch
|
||
from app.services.talent_agent import predict_talent_flow, recommend_talent
|
||
from app.services.board_agent import generate_meeting_summary, generate_questions
|
||
from app.services.portfolio_rebalancer import (
|
||
calculate_marginal_return,
|
||
rebalance_portfolio,
|
||
run_monte_carlo,
|
||
)
|
||
from app.services.agent_orchestrator import orchestrate_agent
|
||
|
||
|
||
class TestSynergyMatcher:
|
||
"""协同匹配 Agent 测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_match_synergy_returns_list(self):
|
||
"""match_synergy 应返回 list。"""
|
||
result = await match_synergy("企业A需要数据标注", "企业B有数据标注团队")
|
||
assert isinstance(result, list)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_match_synergy_non_empty(self):
|
||
"""有匹配条件时应返回非空列表。"""
|
||
result = await match_synergy("企业A需要AI算力", "企业B有GPU集群")
|
||
assert len(result) >= 1
|
||
assert "type" in result[0]
|
||
|
||
|
||
class TestAlphaAttribution:
|
||
"""Alpha 归因 Agent 测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_attribute_returns_dict(self):
|
||
"""attribute_alpha 应返回 dict。"""
|
||
result = await attribute_alpha(
|
||
{"type": "战略建议", "description": "调整产品方向"},
|
||
{"revenue": "+20%", "users": "+15%"},
|
||
)
|
||
assert isinstance(result, dict)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_attribute_has_fields(self):
|
||
"""归因结果应包含关键字段。"""
|
||
result = await attribute_alpha({"type": "人才引进"}, {"headcount": "+10"})
|
||
assert "score" in result or "confidence" in result or "summary" in result
|
||
|
||
|
||
class TestExitPredictor:
|
||
"""退出预测 Agent 测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_predict_exit_returns_dict(self):
|
||
"""predict_exit 应返回 dict。"""
|
||
result = await predict_exit("企业估值5亿，年收入1亿，增长率30%")
|
||
assert isinstance(result, dict)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_predict_exit_non_empty(self):
|
||
"""退出预测结果应非空。"""
|
||
result = await predict_exit("企业估值5亿，年收入1亿，增长率30%")
|
||
assert len(result) > 0
|
||
|
||
|
||
class TestAARAgent:
|
||
"""AAR 复盘 Agent 测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_aar_returns_dict(self):
|
||
"""generate_aar 应返回 dict。"""
|
||
result = await generate_aar(
|
||
"融资失败",
|
||
"计划Q2完成A轮融资5000万",
|
||
"仅获得2000万意向，未达成目标",
|
||
)
|
||
assert isinstance(result, dict)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_check_aar_triggers_funding(self):
|
||
"""融资完成事件应触发 AAR。"""
|
||
events = [
|
||
{"type": "funding_completed", "title": "A轮融资完成"},
|
||
{"type": "normal_event", "title": "例会"},
|
||
]
|
||
triggers = await check_aar_triggers("company-1", events)
|
||
assert len(triggers) == 1
|
||
assert triggers[0]["trigger_type"] == "funding_completed"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_check_aar_triggers_talent(self):
|
||
"""人才离职事件应触发 AAR。"""
|
||
events = [
|
||
{"type": "talent_left", "title": "CTO离职"},
|
||
]
|
||
triggers = await check_aar_triggers("company-1", events)
|
||
assert len(triggers) == 1
|
||
assert triggers[0]["trigger_type"] == "talent_left"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_check_aar_triggers_empty(self):
|
||
"""无触发事件时应返回空列表。"""
|
||
events = [
|
||
{"type": "normal_event", "title": "日常会议"},
|
||
]
|
||
triggers = await check_aar_triggers("company-1", events)
|
||
assert triggers == []
|
||
|
||
|
||
class TestDigitalTwinEngine:
|
||
"""数字孪生 Agent 测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_build_twin_model(self):
|
||
"""build_twin_model 应返回 dict。"""
|
||
result = await build_twin_model("企业A：年收入1亿，团队50人，产品SaaS")
|
||
assert isinstance(result, dict)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_simulate_scenario(self):
|
||
"""simulate_scenario 应返回 dict。"""
|
||
result = await simulate_scenario(
|
||
{"revenue_growth": 0.2, "burn_rate": 500000},
|
||
"市场下行20%",
|
||
)
|
||
assert isinstance(result, dict)
|
||
|
||
|
||
class TestMilestoneAgent:
|
||
"""里程碑 Agent 测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_suggest_path_switch(self):
|
||
"""suggest_path_switch 应返回 dict。"""
|
||
result = await suggest_path_switch(
|
||
"当前里程碑：产品MVP完成",
|
||
"竞品提前发布，市场窗口缩小",
|
||
)
|
||
assert isinstance(result, dict)
|
||
|
||
|
||
class TestTalentAgent:
|
||
"""人才 Agent 测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_predict_talent_flow(self):
|
||
"""predict_talent_flow 应返回 dict。"""
|
||
result = await predict_talent_flow("企业A：50人，近期3人离职")
|
||
assert isinstance(result, dict)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_recommend_talent(self):
|
||
"""recommend_talent 应返回 list。"""
|
||
result = await recommend_talent("需要AI算法工程师", "人才池：张三、李四")
|
||
assert isinstance(result, list)
|
||
|
||
|
||
class TestBoardAgent:
|
||
"""董事会 Agent 测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_meeting_summary(self):
|
||
"""generate_meeting_summary 应返回 str。"""
|
||
result = await generate_meeting_summary("Q3财报：收入增长20%，但利润下降")
|
||
assert isinstance(result, str)
|
||
assert len(result) > 0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_questions(self):
|
||
"""generate_questions 应返回 list。"""
|
||
result = await generate_questions("Q3财报：收入增长20%，但利润下降")
|
||
assert isinstance(result, list)
|
||
|
||
|
||
class TestPortfolioRebalancer:
|
||
"""组合再平衡计算测试（纯计算，无 LLM 依赖）。"""
|
||
|
||
def test_calculate_marginal_return_positive(self):
|
||
"""正投资边际回报率应正确计算。"""
|
||
result = calculate_marginal_return(1000000, 500000, 0.2)
|
||
assert result > 0
|
||
assert result == 0.2 # 边际回报率 = 期望回报率
|
||
|
||
def test_calculate_marginal_return_zero_investment(self):
|
||
"""追加投资为 0 时边际回报率应为 0。"""
|
||
result = calculate_marginal_return(1000000, 0, 0.2)
|
||
assert result == 0.0
|
||
|
||
def test_rebalance_portfolio(self):
|
||
"""再平衡应按边际回报率排序。"""
|
||
companies = [
|
||
{"company_id": "c1", "marginal_return": 0.05},
|
||
{"company_id": "c2", "marginal_return": 0.15},
|
||
{"company_id": "c3", "marginal_return": 0.10},
|
||
{"company_id": "c4", "marginal_return": 0.02},
|
||
]
|
||
result = rebalance_portfolio(companies)
|
||
assert "reallocation_plan" in result
|
||
assert "increase" in result["reallocation_plan"]
|
||
assert "decrease" in result["reallocation_plan"]
|
||
# c2 (0.15) 应在 increase 中
|
||
assert "c2" in result["reallocation_plan"]["increase"]
|
||
# c4 (0.02) 应在 decrease 中
|
||
assert "c4" in result["reallocation_plan"]["decrease"]
|
||
|
||
def test_rebalance_empty(self):
|
||
"""空列表应返回空计划。"""
|
||
result = rebalance_portfolio([])
|
||
assert result["reallocation_plan"]["increase"] == []
|
||
assert result["reallocation_plan"]["decrease"] == []
|
||
|
||
def test_run_monte_carlo_with_floats(self):
|
||
"""纯 float 列表应正常模拟。"""
|
||
result = run_monte_carlo([0.15, 0.10, 0.05], iterations=100)
|
||
assert "irr_distribution" in result
|
||
assert "percentile_p5" in result
|
||
assert "percentile_p50" in result
|
||
assert "percentile_p95" in result
|
||
|
||
def test_run_monte_carlo_with_dicts(self):
|
||
"""dict 列表（含 irr 字段）应正常模拟。"""
|
||
result = run_monte_carlo(
|
||
[{"company_id": "c1", "irr": 0.15}, {"company_id": "c2", "irr": 0.08}],
|
||
iterations=100,
|
||
)
|
||
assert "irr_distribution" in result
|
||
assert result["percentile_p50"] > 0
|
||
|
||
def test_run_monte_carlo_empty(self):
|
||
"""空列表应返回零值。"""
|
||
result = run_monte_carlo([], iterations=100)
|
||
assert result["percentile_p5"] == 0
|
||
assert result["percentile_p50"] == 0
|
||
assert result["percentile_p95"] == 0
|
||
|
||
|
||
class TestAgentOrchestrator:
|
||
"""Agent 编排测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_orchestrate_low_autonomy(self):
|
||
"""低自治级别应标记需要人工审核。"""
|
||
result = await orchestrate_agent("test_agent", "low", {"task": "分析"})
|
||
assert isinstance(result, dict)
|
||
assert "needs_review" in result or "autonomy_level" in result or "status" in result
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_orchestrate_high_autonomy(self):
|
||
"""高自治级别应直接执行。"""
|
||
result = await orchestrate_agent("test_agent", "high", {"task": "生成报告"})
|
||
assert isinstance(result, dict)
|