Nexttag.Ai.Graph 0.3.0

Nexttag.Ai.Graph

Orquestração de agentes em grafo de estado (estilo LangGraph) — fachada estável sobre o Microsoft Agent Framework Workflows.

Instalar

dotnet add package Nexttag.Ai.Graph

Requer .NET 10. Depende de: Microsoft.Agents.AI 1.17.0, Microsoft.Agents.AI.Workflows 1.17.0, Nexttag.Ai.Agents.

Registrar (Program.cs)

O grafo é construído com NexttagGraph.Create<TState>() e compilado com .Compile(). Registre como singleton se for reutilizado:

builder.Services.AddSingleton(sp =>
    NexttagGraph.Create<MeuEstado>()
        .AddNode("etapa1", handler1)
        .AddAgentNode("agente", sp.GetRequiredService<IChatClient>(), messages, apply)
        .AddEdge(NexttagGraph.Start, "etapa1")
        .AddEdge("etapa1", "agente")
        .AddEdge("agente", NexttagGraph.End)
        .Compile());

Configurar

Sem configuração de appsettings própria. O IChatClient vem do Nexttag.Ai.Agents; o GraphCheckpointer.InMemory() não precisa de infraestrutura extra.

Usar

Definir o estado

using Nexttag.Ai.Graph;
using Microsoft.Extensions.AI;

record MeuEstado
{
    // Campos sem [Reduce] usam OverwriteReducer (substitui)
    public string? Resposta { get; init; }
    public int Tentativas  { get; init; }

    // [Reduce(typeof(AppendReducer))] — acumula (delta, não a lista inteira)
    [Reduce(typeof(AppendReducer))]
    public IReadOnlyList<ChatMessage> Conversa { get; init; } = [];
}

Construir e executar o grafo

var grafo = NexttagGraph.Create<MeuEstado>()
    // nó-função: lógica .NET pura
    .AddNode("preparar", async (estado, ctx) =>
    {
        var dados = await servico.BuscarAsync(estado.Conversa[^1].Text, ctx.CancellationToken);
        return StateUpdate.For<MeuEstado>()
            .Set(x => x.Tentativas, estado.Tentativas + 1);
    })
    // nó-agente: roda loop de tool-calling do Nexttag.Ai.Agents
    .AddAgentNode("responder", chatClient,
        messages: s => [.. s.Conversa],
        apply:    (s, r) => StateUpdate.For<MeuEstado>()
            .Set(x => x.Resposta, r.FinalText)
            .Append(x => x.Conversa, [new ChatMessage(ChatRole.Assistant, r.FinalText)]),
        tools:    minhasTools)
    .AddEdge(NexttagGraph.Start, "preparar")
    .AddEdge("preparar", "responder")
    .AddEdge("responder", NexttagGraph.End)
    .Compile(GraphCheckpointer.InMemory());

// Execução simples — devolve o estado final
MeuEstado final = await grafo.RunAsync(estadoInicial, ct);

Arestas condicionais e ciclos

// Router: devolve o id do próximo nó
.AddConditionalEdges("avaliar",
    s => s.Qualidade < 0.7 && s.Tentativas < 3 ? "recuperar" : NexttagGraph.End,
    "recuperar", NexttagGraph.End)

// Aresta condicional simples
.AddConditionalEdge("verificar", "corrigir", s => s.TemErro, label: "com-erro")
.AddConditionalEdge("verificar", NexttagGraph.End, s => !s.TemErro, label: "ok")

Streaming de GraphEvent (SSE para o front)

await foreach (GraphEvent ev in grafo.StreamAsync(estadoInicial, ct))
{
    switch (ev)
    {
        case GraphNodeStarted n:   /* pinte o nó n.NodeId de "rodando" */   break;
        case GraphNodeCompleted n: /* pinte de "ok" */                       break;
        case GraphNodeFailed f:    /* pinte de "erro" — f.Message */         break;
        case GraphAgentEvent a:    /* a.NodeId, a.Inner (AgentEvent ao vivo) */ break;
        case GraphCompleted c:     /* c.Output é o estado final */           break;
        case GraphFailed f:        /* falha geral */                         break;
    }
}

Aprovação humana (human-in-the-loop)

Um nó AddHumanNode<TRequest, TResponse> PAUSA o grafo e espera uma decisão externa. O estado é carregado pela pausa automaticamente; a resposta é dobrada no estado e a execução continua (inclusive ramificando na decisão). Generaliza o "propor/aprovar" do Nexttag.Ai.Agents.

var grafo = NexttagGraph.Create<Pedido>()
    .AddHumanNode<string, bool>("aprovacao",
        buildRequest:  p => $"Aprovar pagamento de R$ {p.Valor}?",   // o que o humano vê
        applyResponse: (p, ok) => StateUpdate.For<Pedido>().Set(x => x.Aprovado, ok))
    .AddNode("aplicar", (p, _) => new(StateUpdate.For<Pedido>().Set(x => x.Desfecho, "pago")))
    .AddNode("negar",   (p, _) => new(StateUpdate.For<Pedido>().Set(x => x.Desfecho, "recusado")))
    .AddEdge(NexttagGraph.Start, "aprovacao")
    .AddConditionalEdge("aprovacao", "aplicar", p => p.Aprovado == true)
    .AddConditionalEdge("aprovacao", "negar",   p => p.Aprovado != true)
    .AddEdge("aplicar", NexttagGraph.End)
    .AddEdge("negar", NexttagGraph.End)
    .Compile();

// onInterrupt devolve o TResponse. Em produção, aguarde a decisão real (ex.: um
// TaskCompletionSource alimentado por um endpoint HTTP de aprovação).
Pedido final = await grafo.RunAsync(pedidoInicial,
    onInterrupt: req => ObterDecisaoDoUsuarioAsync(req)); // req.Payload = a pergunta; retorne bool

// Ou em streaming: o GraphInterrupted aparece no fluxo de eventos para a UI mostrar o card de aprovação.
await foreach (var ev in grafo.StreamAsync(pedidoInicial, onInterrupt: ObterDecisaoAsync, ct))
    if (ev is GraphInterrupted i) { /* mostre o card: i.NodeId, i.Payload */ }

Visualização (React Flow)

// Endpoint .NET — devolve a topologia
GraphView view = grafo.ToGraphView();
return Results.Ok(view.ToJson()); // JSON camelCase

// Front (Next.js / React Flow)
const rfNodes = view.nodes.map(n => ({
  id: n.id,
  data: { label: n.label },
  type: n.kind,              // 'start' | 'end' | 'function' | 'agent'
  position: { x: 0, y: 0 }, // preenchido pelo dagre/elk
}));
const rfEdges = view.edges.map(e => ({
  id: e.id, source: e.source, target: e.target,
  label: e.condition, animated: e.isConditional,
}));

Receitas

Mini-RAG com ciclo de revisão:

var grafo = NexttagGraph.Create<RagState>()
    .AddNode("recuperar", async (s, ctx) =>
    {
        var docs = await vectorStore.BuscarAsync(s.Conversa[^1].Text, ctx.CancellationToken);
        return StateUpdate.For<RagState>()
            .Set(x => x.Documentos, docs)
            .Set(x => x.Tentativas, s.Tentativas + 1);
    })
    .AddAgentNode("responder", chatClient,
        messages: s => MontarPrompt(s),
        apply:    (s, r) => StateUpdate.For<RagState>()
            .Set(x => x.Resposta, r.FinalText)
            .Append(x => x.Conversa, [new ChatMessage(ChatRole.Assistant, r.FinalText)]))
    .AddEdge(NexttagGraph.Start, "recuperar")
    .AddEdge("recuperar", "responder")
    .AddConditionalEdges("responder",
        s => s.Documentos.Count == 0 && s.Tentativas < 3 ? "recuperar" : NexttagGraph.End,
        "recuperar", NexttagGraph.End)
    .Compile(GraphCheckpointer.InMemory());

RagState final = await grafo.RunAsync(estadoInicial, ct);

Grafo com ponto de entrada no estado:

var estadoInicial = new MeuEstado
{
    Conversa = [new ChatMessage(ChatRole.User, perguntaDoUsuario)]
};
MeuEstado saida = await grafo.RunAsync(estadoInicial, ct);
Console.WriteLine(saida.Resposta);

Telemetria:

tracerProviderBuilder.AddSource("Nexttag.Ai.Graph");  // ActivitySource da lib
// Nós-agente já rastreiam no Langfuse via litellm (Nexttag.Ai.Agents)

Notas

• O estado (TState) precisa ter construtor sem parâmetros — o padrão é record com propriedades init.
• Campos sem [Reduce] usam OverwriteReducer (substitui). Use [Reduce(typeof(AppendReducer))] para histórico/listas acumuladas.
• StateUpdate.Append recebe apenas o delta (itens novos) — não a lista acumulada.
• NexttagGraph.Start e NexttagGraph.End são sentinelas; não adicione nós com esses ids.
• Compile() valida a existência de pelo menos uma aresta saindo de Start e chegando em End.
• GraphCheckpointer.InMemory() é volátil (mesma instância do processo). Checkpoint Postgres (durabilidade entre processos) é roadmap.
• Nenhum tipo do Microsoft.Agents.AI.Workflows vaza na API pública — atualizações do MAF (ainda RC) ficam contidas nesta lib.