LangGraph: workflow-uri cu stare, routere și agenți multipli

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

# 1. State — datele care circulă prin graf
class TextState(TypedDict):
    text:  str
    count: int

# 2. Node — funcție care primește state și returnează modificări
def count_words(state: TextState) -> dict:
    n = len(state["text"].split())
    return {"count": n}      # doar ce s-a schimbat

# 3. Construim graful
workflow = StateGraph(TextState)
workflow.add_node("count_words", count_words)
workflow.add_edge(START, "count_words")
workflow.add_edge("count_words", END)

# 4. Compilăm și rulăm
graph = workflow.compile()
result = graph.invoke({"text": "LangGraph orchestrează workflow-uri AI.", "count": 0})
print(result)
# → {'text': 'LangGraph orchestrează workflow-uri AI.', 'count': 4}

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

class LengthState(TypedDict):
    text:   str
    result: str

# Noduri — ramurile
def short_response(state: LengthState) -> dict:
    return {"result": "Text scurt."}

def long_response(state: LengthState) -> dict:
    return {"result": "Text lung."}

# Routerul — decide ramura, returnează numele etichetei
def choose_path(state: LengthState) -> str:
    if len(state["text"].split()) < 10:
        return "short"
    return "long"

# Construim graful
workflow = StateGraph(LengthState)
workflow.add_node("short_response", short_response)
workflow.add_node("long_response",  long_response)

# Muchie condițională cu mapping explicit etichetă → nod
workflow.add_conditional_edges(
    START,
    choose_path,
    {"short": "short_response", "long": "long_response"},
)
workflow.add_edge("short_response", END)
workflow.add_edge("long_response",  END)

graph = workflow.compile()
graph.invoke({"text": "Salut lume.", "result": ""})
# → {'text': 'Salut lume.', 'result': 'Text scurt.'}

from IPython.display import Image, display

# Imaginea PNG generată din graf (necesită conexiune pentru rendering)
display(Image(graph.get_graph().draw_mermaid_png()))

# Fallback: textul Mermaid brut, dacă rendering-ul nu merge
print(graph.get_graph().draw_mermaid())

---
config:
  flowchart:
    curve: linear
---
graph TD;
        __start__([<p>__start__</p>]):::first
        short_response(short_response)
        long_response(long_response)
        __end__([<p>__end__</p>]):::last
        __start__ -.-> short_response;
        __start__ -.-> long_response;
        short_response --> __end__;
        long_response --> __end__;

HANDLES = {
    "anti_sistem":     "@LibertateRO99",
    "conspirationist": "@AdevarulViu",
    "pro_european":    "@EuroOptimistRO",
}

messages = [
    {"agent": "Anti-sistem",  "slug": "anti_sistem",
     "text":  "Instituțiile par din nou rupte de oameni.", "turn": 1},
    {"agent": "Pro-european", "slug": "pro_european",
     "text":  "Trebuie să discutăm pe baza procedurilor și a dovezilor.", "turn": 2},
]

def thread_to_text(messages: list) -> str:
    lines = []
    for m in messages:
        handle = HANDLES.get(m["slug"], m["slug"])
        lines.append(f"Turn {m['turn']} — {handle}: {m['text']}")
    return "\n".join(lines)

print(thread_to_text(messages))
# Turn 1 — @LibertateRO99: Instituțiile par din nou rupte de oameni.
# Turn 2 — @EuroOptimistRO: Trebuie să discutăm pe baza procedurilor...

from typing import TypedDict

class ThreadState(TypedDict):
    stimulus:     str        # inputul politic inițial
    messages:     list       # mesajele produse până acum
    active_slugs: list       # agenții care participă la thread
    total_turns:  int        # câte intervenții vrem în total
    current_turn: int        # câte au fost deja produse
    next_slug:    str        # agentul ales pentru intervenția următoare
    provider:     str        # "gemini" sau "deepseek"
    k:            int        # câte fragmente FAISS recuperăm

def pick_next_round_robin(active_slugs: list, current_turn: int) -> str:
    """Alege agentul următor: tura N → al N-lea agent (modulo nr. agenți)."""
    return active_slugs[current_turn % len(active_slugs)]

def router_node(state: ThreadState) -> dict:
    """Setează next_slug în state, sau '__end__' dacă thread-ul s-a terminat."""
    if state["current_turn"] >= state["total_turns"]:
        return {"next_slug": "__end__"}

    next_slug = pick_next_round_robin(
        state["active_slugs"],
        state["current_turn"],
    )
    return {"next_slug": next_slug}

def route_decision(state: ThreadState) -> str:
    """Muchia condițională: citește next_slug și-l returnează ca etichetă."""
    return state["next_slug"]

from core.agent import generate_agent_response

def make_agent_node(slug: str):
    """Factory: returnează un nod specializat pentru agentul `slug`."""

    def agent_node(state: ThreadState) -> dict:
        thread_ctx = thread_to_text(state["messages"])

        # input extins: stimulusul + thread-ul anterior
        extended_input = f"""[STIMULUS]
{state['stimulus']}

[THREAD ANTERIOR]
{thread_ctx if thread_ctx else "Prima intervenție."}

[SARCINĂ]
Răspunde ca agentul tău. Dacă există mesaje anterioare, reacționează la ele."""

        # apel către agentul RAG din C6
        result = generate_agent_response(
            agent_slug=slug,
            input_text=extended_input,
            provider=state["provider"],
            k=state["k"],
        )

        new_turn = state["current_turn"] + 1
        new_msg = {
            "agent": result["agent_name"],
            "slug":  slug,
            "text":  result["response"],
            "turn":  new_turn,
        }

        # întoarcem messages cu mesajul nou adăugat la sfârșit
        return {
            "messages":     state["messages"] + [new_msg],
            "current_turn": new_turn,
        }

    return agent_node

from langgraph.graph import StateGraph, START, END

def build_graph(active_slugs: list[str]):
    workflow = StateGraph(ThreadState)

    # nodul router
    workflow.add_node("router", router_node)

    # câte un nod pentru fiecare agent activ
    for slug in active_slugs:
        workflow.add_node(slug, make_agent_node(slug))
        workflow.add_edge(slug, "router")     # agent → router (întotdeauna)

    # intrare
    workflow.add_edge(START, "router")

    # muchia condițională: router → un agent sau END
    mapping = {slug: slug for slug in active_slugs} | {"__end__": END}
    workflow.add_conditional_edges("router", route_decision, mapping)

    return workflow.compile()

graph = build_graph(["anti_sistem", "conspirationist", "pro_european"])

initial_state = {
    "stimulus":     "CCR a decis anularea alegerilor după suspiciuni privind influențe externe.",
    "messages":     [],
    "active_slugs": ["anti_sistem", "conspirationist", "pro_european"],
    "total_turns":  4,
    "current_turn": 0,
    "next_slug":    "",
    "provider":     "gemini",
    "k":            3,
}

result = graph.invoke(initial_state)
for m in result["messages"]:
    print(f"Turn {m['turn']} — {m['agent']}: {m['text']}")

from core.agent import make_llm

def router_node_llm(state: ThreadState) -> dict:
    if state["current_turn"] >= state["total_turns"]:
        return {"next_slug": "__end__"}

    active = state["active_slugs"]
    if not state["messages"]:
        return {"next_slug": active[0]}

    # constrângerea hard: nu alege agentul care tocmai a vorbit
    last_slug  = state["messages"][-1]["slug"]
    candidates = [s for s in active if s != last_slug] or active

    prompt = f"""[STIMULUS]
{state["stimulus"]}

[THREAD]
{thread_to_text(state["messages"])}

[Agenți disponibili]
{chr(10).join(f"- {s}" for s in candidates)}

Alege agentul care ar trebui să răspundă următorul.
Răspunde DOAR cu slug-ul exact."""

    llm    = make_llm(provider=state["provider"], temperature=0.1)
    chosen = llm.invoke(prompt).content.strip().lower()
    return {"next_slug": chosen if chosen in candidates else candidates[0]}

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from core.agent import generate_agent_response

# --- State ---------------------------------------------------------
class ThreadState(TypedDict):
    stimulus: str; messages: list; active_slugs: list
    total_turns: int; current_turn: int; next_slug: str
    provider: str; k: int

# --- Helpers -------------------------------------------------------
def thread_to_text(messages: list) -> str: ...

# --- Router --------------------------------------------------------
def router_node(state):    ...
def route_decision(state): return state["next_slug"]

# --- Agent node factory --------------------------------------------
def make_agent_node(slug: str): ...

# --- Builder + runner ---------------------------------------------
def build_graph(active_slugs): ...

def run_thread(stimulus: str, active_slugs: list,
               total_turns: int = 4, provider: str = "gemini",
               k: int = 3) -> dict:
    graph = build_graph(active_slugs)
    return graph.invoke({
        "stimulus": stimulus, "messages": [],
        "active_slugs": active_slugs, "total_turns": total_turns,
        "current_turn": 0, "next_slug": "",
        "provider": provider, "k": k,
    })

python -m core.graph \
  --agents anti_sistem conspirationist pro_european \
  --text "CCR a decis anularea alegerilor după suspiciuni privind influențe externe." \
  --turns 4 \
  --provider gemini

from core.graph import run_thread

# În tab-ul „Multi-agent thread"
def on_run_click(stimulus, active_slugs, total_turns, provider):
    result = run_thread(
        stimulus=stimulus,
        active_slugs=active_slugs,
        total_turns=total_turns,
        provider=provider,
    )
    return result["messages"]   # UI le afișează ca thread cu carduri

# Toate testele folosesc același input
INPUT="CCR a decis anularea alegerilor după suspiciuni privind influențe externe."
AGENT="anti_sistem"   # sau agentul pentru care echipa are sigur vectorstore

python -m core.retriever --agent $AGENT --query "$INPUT" --k 5
python -m core.agent     --agent $AGENT --text "$INPUT" --provider gemini
python -m core.graph     --agents anti_sistem conspirationist pro_european \
                          --text "$INPUT" --turns 4 --provider gemini

DONE C7 team work session

roles.yaml works:           yes / no
retriever works:            yes / no
agent.py works:             yes / no
graph.py / notebook works:  yes / no
app multi-agent tab works:  yes / no
ethics_and_limits.md:       yes / no

Observații:
- [orice problemă rămasă deschisă]
- [ce trebuie reluat în sesiunea următoare]