Agent相关面试题

你做的多 agent 之间是怎么进行通讯的中央 agent 是怎么给下面的子 agent 分配任务的串行并行一、多 Agent 通讯与任务分配机制1. 通讯架构异步消息总线 (MessageBus)Agent 之间通过MessageBus进行异步消息通信第 822-903 行class MessageBus: def __init__(self): # 每个Agent独立队列避免全局锁竞争 self._agent_queues: Dict[str, asyncio.Queue] {} self._subscribers: Dict[str, List[Callable]] defaultdict(list)核心特点每个 Agent 有独立队列- 减少锁竞争提高并发性能支持点对点通信- 通过receiver字段指定目标 Agent支持广播-receiverNone时广播给所有 Agent除发送者外消息优先级- 支持 1-10 的优先级设置消息持久化- 记录消息历史用于追溯2. 任务分配方式串行主任务 并行 QC中央 Agent 的任务调度逻辑第 4965-4994 行async def _dispatch_to_agents(self, correlation_id, query, classification): agents_to_invoke [classification.primary_agent] classification.supporting_agents # 串行发送任务给主 Agent await self.send_message( receiverclassification.primary_agent, message_typeMessageType.TASK_ASSIGN, payload{query: query, classification: asdict(classification)}, correlation_idcorrelation_id, requires_responseTrue )分配策略维度实现方式主任务执行串行- 由一个主 Agentprimary_agent执行任务辅助 Agent串行- 主 Agent 可调用辅助 Agent但通常也是串行协作QC 审核并行非阻塞- 主 Agent 返回结果后后台异步提交 QC 审核第 5034-5036 行非阻塞 QC 机制# 并行提交QC审核不阻塞结果返回 if workflow.get(requires_qc, False): asyncio.create_task(self._submit_to_qc_async(message, correlation_id)) ​ # 直接整合并返回结果不等待QC final_result await self._integrate_results(correlation_id)你是怎么做的agent 回答质量检测的功能怎么识别敏感输出二、Agent 回答质量检测与敏感输出识别1. 质量检测QualityControlAgent (Agent5)审核维度第 6855-7258 行class QualityControlAgent(BaseAgent): async def _handle_qc_review_request(self, message: AgentMessage): # 执行四维审核 fact_check await self._check_facts(output) # 事实核查 citation_check await self._verify_citations(citations) # 引用验证 logic_check await self._check_logic(output) # 逻辑一致性 safety_check await self._check_safety(output) # 安全合规具体检测内容检测项方法实现方式事实核查_check_facts()提取科学声明 → RAG 检索验证 → 计算准确性分数引用验证_verify_citations()检查引用格式完整性标题、年份、作者逻辑一致性_check_logic()使用 LLM 检查自相矛盾、推理合理性安全合规_check_safety()调用 SafetyGuard 进行多维度安全检查QC 报告生成class QCReport: overall_score: float # 总体质量分数 0-1 factual_accuracy: float # 事实准确性 citation_quality: float # 引用质量 logical_consistency: float # 逻辑一致性 safety_compliance: float # 安全合规性 issues: List[Dict] # 发现的问题列表 corrections: List[Dict] # 建议的修正 approved: bool # 是否通过审核2. 敏感输出识别SafetyGuard多维度敏感内容检测第 3715-4337 行class SafetyGuard: # 检测维度 POLITICAL_PATTERNS {...} # 政治敏感内容三级分类 CHEMICAL_WEAPON_PATTERNS {...} # 化学武器/危险品滥用 PROMPT_INJECTION_PATTERNS [...] # 提示词注入攻击 PRIVACY_PATTERNS {...} # 隐私数据泄露 LAB_SAFETY_PATTERNS {...} # 实验安全违规 HATE_VIOLENCE_PATTERNS [...] # 仇恨/暴力内容检测机制风险类别检测方法风险等级政治敏感三级正则匹配critical/high/mediumCRITICAL/HIGH/MEDIUM化学武器滥用关键词 上下文模式匹配CRITICAL爆炸物/毒品特定化合物名称 制备动词CRITICAL提示词注入角色绕过、指令覆盖、分割攻击模式HIGH隐私泄露身份证、手机号、邮箱、银行卡号正则MEDIUM实验安全违规无防护操作、危险处置方式CRITICAL/HIGH白名单机制上下文感知RESEARCH_WHITELIST { core_terms: [电池, 电解液, 锂离子电池, 电化学, ...], legitimate_chemicals: [碳酸乙烯酯, EC, DEC, LiPF6, ...] } ​ def _calculate_total_risk(self, risk_scores, whitelist_analysis): # 如果是明显的研究上下文降低化学品相关风险分数 if whitelist_analysis[is_likely_research]: if chemical_weapons in risk_scores and whitelist_analysis[chemical_matches] 2: whitelist_discount 20风险评分与响应RISK_THRESHOLDS { RiskLevel.CRITICAL: 80, # 严重风险 - 必须阻止 RiskLevel.HIGH: 60, # 高风险 - 强烈建议阻止 RiskLevel.MEDIUM: 40, # 中等风险 - 需要人工审核 RiskLevel.LOW: 20, # 低风险 - 记录日志但允许 }预编译正则优化def _compile_patterns(self): 预编译正则表达式以提高性能 self._compiled_political { level: [re.compile(p, re.IGNORECASE) for p in patterns] for level, patterns in self.POLITICAL_PATTERNS.items() }总结问题答案通讯方式基于MessageBus的异步消息队列每个 Agent 独立队列任务分配串行执行主任务单主 Agent并行后台 QC 审核质量检测四维检测事实准确性、引用质量、逻辑一致性、安全合规性敏感识别基于预编译正则的多级风险检测 白名单上下文感知