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

• 支持广播-receiver=None时广播给所有 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( receiver=classification.primary_agent, message_type=MessageType.TASK_ASSIGN, payload={"query": query, "classification": asdict(classification)}, correlation_id=correlation_id, requires_response=True )

分配策略：

非阻塞 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) # 安全合规

具体检测内容：

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 = [...] # 仇恨/暴力内容

检测机制：

白名单机制（上下文感知）：

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() }

总结