Протоколы консенсуса в мультиагентных системах: Искоренение галлюцинаций через дебаты

from typing import TypedDict, List
from langgraph.graph import StateGraph, END

class ConsensusState(TypedDict):
    task: str
    code_draft: str
    security_feedback: str
    approved: bool
    rounds: int

def coder_agent(state: ConsensusState):
    """Генерирует или исправляет код."""
    iteration = state.get("rounds", 0) + 1
    feedback = state.get("security_feedback", "")
    print(f"[Coder] Раунд {iteration}: Исправление кода на основе замечаний...")
    return {
        "code_draft": f"# Ревизия {iteration} с учетом: {feedback}",
        "rounds": iteration
    }

def security_auditor(state: ConsensusState):
    """Проверяет безопасность решения."""
    # Если это второй раунд — одобряем
    is_safe = state["rounds"] >= 2
    feedback = "Все чисто" if is_safe else "Обнаружен риск SQL-инъекции в конкатенации строк"
    print(f"[Auditor] Вердикт: {'ОДОБРЕНО' if is_safe else 'ОТКЛОНЕНО'}")
    return {"security_feedback": feedback, "approved": is_safe}

def consensus_router(state: ConsensusState):
    if state["approved"] or state["rounds"] >= 3:
        return "finish"
    return "revise"

workflow = StateGraph(ConsensusState)
workflow.add_node("coder", coder_agent)
workflow.add_node("auditor", security_auditor)

workflow.set_entry_point("coder")
workflow.add_edge("coder", "auditor")
workflow.add_conditional_edges("auditor", consensus_router, {
    "revise": "coder",
    "finish": END
})

app = workflow.compile()