"""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)