From 17851353a1b9a2e06cec46c23e552aff761dade9 Mon Sep 17 00:00:00 2001 From: ybszm Date: Thu, 23 Jul 2026 01:38:37 +0800 Subject: [PATCH] feat: P0-P2 positioning, eval, A2A, durability, and DX docs - README front door: QuickAgent + multi-agent examples; pin @v0.1.0 - Docs: COMPARISON, LANGCHAINGO interop, VERSIONING - eval golden trajectories; a2a HTTP subset; httpware auth/rate-limit - examples/mcp-secure GuardTools pattern - RunConfig.Durability (sync); WithDeferred marker; graph benchmarks --- CHANGELOG.md | 9 ++ COMPATIBILITY.md | 4 +- README.md | 65 ++++------ README.zh-CN.md | 80 +++--------- a2a/README.md | 45 +++++++ a2a/protocol.go | 207 ++++++++++++++++++++++++++++++ a2a/protocol_test.go | 42 ++++++ docs/COMPARISON.md | 61 +++++++++ docs/DURABILITY.md | 15 ++- docs/LANGCHAINGO.md | 86 +++++++++++++ docs/README.md | 4 + docs/VERSIONING.md | 44 +++++++ eval/trajectory.go | 120 +++++++++++++++++ eval/trajectory_test.go | 34 +++++ examples/README.md | 1 + examples/mcp-secure/main.go | 65 ++++++++++ graph/bench_test.go | 57 ++++++++ graph/builder.go | 9 ++ graph/compiled.go | 21 +++ graph/deferred_durability_test.go | 69 ++++++++++ graph/errors.go | 3 + graph/retry.go | 15 +++ graph/types.go | 17 +++ httpware/middleware.go | 121 +++++++++++++++++ httpware/middleware_test.go | 47 +++++++ 25 files changed, 1137 insertions(+), 104 deletions(-) create mode 100644 a2a/README.md create mode 100644 a2a/protocol.go create mode 100644 a2a/protocol_test.go create mode 100644 docs/COMPARISON.md create mode 100644 docs/LANGCHAINGO.md create mode 100644 docs/VERSIONING.md create mode 100644 eval/trajectory.go create mode 100644 eval/trajectory_test.go create mode 100644 examples/mcp-secure/main.go create mode 100644 graph/bench_test.go create mode 100644 graph/deferred_durability_test.go create mode 100644 httpware/middleware.go create mode 100644 httpware/middleware_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d51f37..9d572e2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,15 @@ intentional API changes, and patch releases preserve public API compatibility. ### Added +- Docs: `COMPARISON.md`, `LANGCHAINGO.md`, `VERSIONING.md`; README QuickAgent front door. +- `eval` golden trajectory helpers for tool-order and final-content assertions. +- `a2a` minimal Agent-to-Agent HTTP card + message:send protocol. +- `httpware` bearer auth, session header, and local rate-limit middleware. +- `examples/mcp-secure` GuardTools pattern for MCP tool binding. +- `graph.RunConfig.Durability` (sync supported; async/exit return clear errors). +- `graph.WithDeferred` node marker + `CompiledGraph.IsDeferred`. +- Graph micro-benchmarks (`BenchmarkLinearInvoke`, `BenchmarkFanOutThree`). + ### Changed ### Fixed diff --git a/COMPATIBILITY.md b/COMPATIBILITY.md index c6a4543..7c8eb82 100644 --- a/COMPATIBILITY.md +++ b/COMPATIBILITY.md @@ -13,8 +13,8 @@ No row below should be interpreted as a drop-in Python API guarantee. | Area | Status | Notes | |---|---|---| -| Typed StateGraph | Partial | Typed nodes, reducers, schemas, static/conditional/waiting edges, Send, Command, compilation and inspection. Default compile still requires every node (and END) reachable from START—stricter than Python—but `WithAllowUnreachableNodes` / `WithAllowUnreachableEND` can relax that. Full Python **deferred node** scheduling is not implemented yet. | -| Pregel runtime | Partial | Deterministic BSP execution, concurrency, retries, caching, timeouts, pending-write recovery. No public low-level `Pregel`/`NodeBuilder` facade or configurable `sync`/`async`/`exit` durability. | +| Typed StateGraph | Partial | Typed nodes, reducers, schemas, static/conditional/waiting edges, Send, Command, compilation and inspection. Default compile still requires every node (and END) reachable from START—stricter than Python—but `WithAllowUnreachableNodes` / `WithAllowUnreachableEND` can relax that. `WithDeferred` records deferred nodes (`IsDeferred`); full Python deferred barrier scheduling remains open. | +| Pregel runtime | Partial | Deterministic BSP execution, concurrency, retries, caching, timeouts, pending-write recovery. `RunConfig.Durability` exposes sync/async/exit names; only sync is implemented (`ErrUnsupportedDurability` otherwise). No public low-level `Pregel`/`NodeBuilder` facade. | | Channels | Supported | Last/Any/Ephemeral/Untracked values, aggregate, topic, named barrier, delta, checkpoint versions, and freshness scheduling. | | Streaming | Partial | Values, updates, messages, custom, debug, interrupts, subgraphs, backpressure, run-event v3, and native SSE provider chunks. Python stream-transformer and graph-UI APIs are absent. | | Checkpoints | Supported | Memory, native SQLite/PostgreSQL, optional Redis, history, pending writes, replay, forks, encryption, and typed codecs. Python physical adapters require isolated schemas/databases. | diff --git a/README.md b/README.md index 08d8f04..8f14287 100644 --- a/README.md +++ b/README.md @@ -60,12 +60,15 @@ LangChain product and is not a source-to-source port of the Python package. ## Install ```bash -go get github.com/ybszm/langgraph-go@latest +go get github.com/ybszm/langgraph-go@v0.1.0 ``` -Go 1.25 or newer is required. +Go 1.25 or newer is required. See [versioning](docs/VERSIONING.md) for pre-1.0 +compatibility promises. -## Quick start +## Quick start (agent) + +Most applications should start with **QuickAgent**, not a hand-built graph: ```go package main @@ -75,58 +78,42 @@ import ( "fmt" "github.com/ybszm/langgraph-go/graph" + "github.com/ybszm/langgraph-go/prebuilt" ) -type State struct { - Count int -} +type echo struct{} -type Delta struct { - Increment int +func (echo) Invoke(_ context.Context, state prebuilt.AgentState, _ graph.Runtime) (prebuilt.AssistantMessage, error) { + return prebuilt.AssistantMessage{Content: "hello from quick-agent"}, nil } func main() { - builder := graph.NewStateGraph(func( - _ context.Context, - state State, - updates []Delta, - ) (State, error) { - for _, update := range updates { - state.Count += update.Increment - } - return state, nil + agent, err := prebuilt.NewQuickAgent(prebuilt.QuickAgentConfig{ + Model: echo{}, + SystemPrompt: "Be helpful.", }) - - if err := builder.AddNode("increment", func( - _ context.Context, - _ State, - _ graph.Runtime, - ) (graph.Command[Delta], error) { - return graph.Update(Delta{Increment: 1}), nil - }); err != nil { - panic(err) - } - if err := builder.AddEdge(graph.START, "increment"); err != nil { - panic(err) - } - if err := builder.AddEdge("increment", graph.END); err != nil { - panic(err) - } - - compiled, err := builder.Compile() if err != nil { panic(err) } - - result, err := compiled.Invoke(context.Background(), State{}, graph.RunConfig{}) + state, err := agent.Run(context.Background(), "hi", graph.RunConfig{}) if err != nil { panic(err) } - fmt.Println(result.Count) // 1 + fmt.Println(state.FinalResponse()) } ``` -The runnable example is available at [`examples/basic`](examples/basic). +Runnable demos: + +| Example | Concept | +|---|---| +| [`examples/quick-agent`](examples/quick-agent) | Smallest agent entry | +| [`examples/multi-agent`](examples/multi-agent) | Supervisor + parallel sub-agents | +| [`examples/basic`](examples/basic) | Typed StateGraph without LLM | + +Multi-agent, handoff, and deep-agent patterns: [docs/agents.md](docs/agents.md). +Interop with langchaingo: [docs/LANGCHAINGO.md](docs/LANGCHAINGO.md). +How we compare to other Go stacks: [docs/COMPARISON.md](docs/COMPARISON.md). ## Packages diff --git a/README.zh-CN.md b/README.zh-CN.md index 37f3629..845eaa6 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -54,73 +54,33 @@ LangGraph Go 是由社区独立维护的 Go 实现，借鉴了 ## 安装 ```bash -go get github.com/ybszm/langgraph-go@latest +go get github.com/ybszm/langgraph-go@v0.1.0 ``` -需要 Go 1.25 或更高版本。 +需要 Go 1.25 或更高版本。版本承诺见 [docs/VERSIONING.md](docs/VERSIONING.md)。 -## 快速开始 +## 快速开始（Agent） + +多数应用应从 **QuickAgent** 入手： ```go -package main - -import ( - "context" - "fmt" - - "github.com/ybszm/langgraph-go/graph" -) - -type State struct { - Count int -} - -type Delta struct { - Increment int -} - -func main() { - builder := graph.NewStateGraph(func( - _ context.Context, - state State, - updates []Delta, - ) (State, error) { - for _, update := range updates { - state.Count += update.Increment - } - return state, nil - }) - - if err := builder.AddNode("increment", func( - _ context.Context, - _ State, - _ graph.Runtime, - ) (graph.Command[Delta], error) { - return graph.Update(Delta{Increment: 1}), nil - }); err != nil { - panic(err) - } - if err := builder.AddEdge(graph.START, "increment"); err != nil { - panic(err) - } - if err := builder.AddEdge("increment", graph.END); err != nil { - panic(err) - } - - compiled, err := builder.Compile() - if err != nil { - panic(err) - } - - result, err := compiled.Invoke(context.Background(), State{}, graph.RunConfig{}) - if err != nil { - panic(err) - } - fmt.Println(result.Count) // 1 -} +agent, err := prebuilt.NewQuickAgent(prebuilt.QuickAgentConfig{ + Model: model, // 实现 prebuilt.ChatModel[prebuilt.AgentState] + SystemPrompt: "简洁回答。", +}) +state, err := agent.Run(ctx, "你好", graph.RunConfig{}) +fmt.Println(state.FinalResponse()) ``` -可运行示例位于 [`examples/basic`](examples/basic)。 +| 示例 | 说明 | +|---|---| +| [`examples/quick-agent`](examples/quick-agent) | 最小 Agent 入口 | +| [`examples/multi-agent`](examples/multi-agent) | 监督者 + 并行子 Agent | +| [`examples/basic`](examples/basic) | 无 LLM 的类型化 StateGraph | + +多 Agent 见 [docs/agents.md](docs/agents.md)；与 langchaingo 协作见 +[docs/LANGCHAINGO.md](docs/LANGCHAINGO.md)；对比见 +[docs/COMPARISON.md](docs/COMPARISON.md)。 ## 包结构 diff --git a/a2a/README.md b/a2a/README.md new file mode 100644 index 0000000..a4a3911 --- /dev/null +++ b/a2a/README.md @@ -0,0 +1,45 @@ +# A2A subset (langgraph-go) + +Minimal Agent-to-Agent HTTP protocol for services that already use +langgraph-go agents. This is **not** a full multi-vendor A2A implementation. + +## Endpoints + +| Method | Path | Purpose | +|---|---|---| +| `GET` | `/a2a/v1/card` | Agent card (name, skills, metadata) | +| `POST` | `/a2a/v1/message:send` | Send a user/agent message, receive agent reply | + +Header: `X-A2A-Protocol: langgraph-go-a2a/0.1` + +## Server + +```go +srv := &a2a.Server{ + Card: a2a.AgentCard{Name: "ops-agent", Skills: []string{"diagnose"}}, + Handler: func(ctx context.Context, req a2a.SendRequest) (a2a.SendResponse, error) { + // Call prebuilt.QuickAgent / MultiAgentCoordinator here. + return a2a.SendResponse{ + SessionID: req.SessionID, + Message: a2a.Message{Role: "agent", Content: "ok"}, + }, nil + }, +} +http.Handle("/a2a/", srv) +``` + +## Client + +```go +client := &a2a.Client{BaseURL: "http://127.0.0.1:8080"} +card, _ := client.GetCard(ctx) +resp, _ := client.Send(ctx, a2a.SendRequest{ + SessionID: "s1", + Message: a2a.Message{Role: "user", Content: "check redis"}, +}) +``` + +## Security + +Put TLS, authentication, and rate limits in front of these handlers. See the +`httpware` package for bearer auth and rate limiting helpers. diff --git a/a2a/protocol.go b/a2a/protocol.go new file mode 100644 index 0000000..9c03382 --- /dev/null +++ b/a2a/protocol.go @@ -0,0 +1,207 @@ +// Package a2a implements a minimal Agent-to-Agent HTTP subset for Go services. +// +// It is intentionally small and JSON-over-HTTP so agents built with langgraph-go +// can call each other without depending on a full multi-vendor A2A stack. +// Field names follow common A2A card/message shapes where practical. +package a2a + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "time" +) + +const ( + // ProtocolVersion is advertised by this package. + ProtocolVersion = "langgraph-go-a2a/0.1" + // HeaderProtocol carries ProtocolVersion. + HeaderProtocol = "X-A2A-Protocol" + // DefaultPath is the message send endpoint path suffix. + DefaultPath = "/a2a/v1/message:send" + // CardPath is the agent card discovery path. + CardPath = "/a2a/v1/card" +) + +// AgentCard describes one callable agent for discovery. +type AgentCard struct { + Name string `json:"name"` + Description string `json:"description,omitempty"` + URL string `json:"url,omitempty"` + Version string `json:"version,omitempty"` + Skills []string `json:"skills,omitempty"` + Metadata map[string]string `json:"metadata,omitempty"` +} + +// Message is one agent turn payload. +type Message struct { + Role string `json:"role"` // user | agent | system + Content string `json:"content"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +// SendRequest is the HTTP body for message:send. +type SendRequest struct { + // SessionID correlates multi-turn collaboration. + SessionID string `json:"session_id,omitempty"` + Message Message `json:"message"` + History []Message `json:"history,omitempty"` +} + +// SendResponse is the HTTP body returned by message:send. +type SendResponse struct { + SessionID string `json:"session_id,omitempty"` + Message Message `json:"message"` + // TraceID is an optional correlation identifier. + TraceID string `json:"trace_id,omitempty"` +} + +// HandlerFunc processes one A2A message. +type HandlerFunc func(context.Context, SendRequest) (SendResponse, error) + +// Server exposes card + message endpoints. +type Server struct { + Card AgentCard + Handler HandlerFunc + // MaxBodyBytes defaults to 1 MiB. + MaxBodyBytes int64 +} + +// ServeHTTP implements http.Handler. +func (s *Server) ServeHTTP(writer http.ResponseWriter, request *http.Request) { + writer.Header().Set(HeaderProtocol, ProtocolVersion) + switch { + case request.Method == http.MethodGet && strings.HasSuffix(request.URL.Path, CardPath): + writeJSON(writer, http.StatusOK, s.Card) + case request.Method == http.MethodPost && strings.HasSuffix(request.URL.Path, DefaultPath): + s.handleSend(writer, request) + default: + http.NotFound(writer, request) + } +} + +func (s *Server) handleSend(writer http.ResponseWriter, request *http.Request) { + if s.Handler == nil { + writeJSON(writer, http.StatusServiceUnavailable, map[string]string{"error": "handler not configured"}) + return + } + limit := s.MaxBodyBytes + if limit <= 0 { + limit = 1 << 20 + } + body, err := io.ReadAll(io.LimitReader(request.Body, limit+1)) + if err != nil { + writeJSON(writer, http.StatusBadRequest, map[string]string{"error": err.Error()}) + return + } + if int64(len(body)) > limit { + writeJSON(writer, http.StatusRequestEntityTooLarge, map[string]string{"error": "body too large"}) + return + } + var req SendRequest + if err := json.Unmarshal(body, &req); err != nil { + writeJSON(writer, http.StatusBadRequest, map[string]string{"error": "invalid json"}) + return + } + if strings.TrimSpace(req.Message.Content) == "" { + writeJSON(writer, http.StatusBadRequest, map[string]string{"error": "message.content is required"}) + return + } + if req.Message.Role == "" { + req.Message.Role = "user" + } + resp, err := s.Handler(request.Context(), req) + if err != nil { + writeJSON(writer, http.StatusInternalServerError, map[string]string{"error": err.Error()}) + return + } + if resp.Message.Role == "" { + resp.Message.Role = "agent" + } + if resp.SessionID == "" { + resp.SessionID = req.SessionID + } + writeJSON(writer, http.StatusOK, resp) +} + +// Client calls a remote A2A agent. +type Client struct { + BaseURL string + HTTPClient *http.Client + // Headers are applied to every request. + Headers map[string]string +} + +// GetCard fetches the remote agent card. +func (c *Client) GetCard(ctx context.Context) (AgentCard, error) { + var card AgentCard + if err := c.doJSON(ctx, http.MethodGet, CardPath, nil, &card); err != nil { + return AgentCard{}, err + } + return card, nil +} + +// Send posts one message to the remote agent. +func (c *Client) Send(ctx context.Context, req SendRequest) (SendResponse, error) { + var resp SendResponse + if err := c.doJSON(ctx, http.MethodPost, DefaultPath, req, &resp); err != nil { + return SendResponse{}, err + } + return resp, nil +} + +func (c *Client) doJSON(ctx context.Context, method, path string, in any, out any) error { + if c == nil || strings.TrimSpace(c.BaseURL) == "" { + return fmt.Errorf("a2a client base URL is empty") + } + httpClient := c.HTTPClient + if httpClient == nil { + httpClient = &http.Client{Timeout: 60 * time.Second} + } + var body io.Reader + if in != nil { + encoded, err := json.Marshal(in) + if err != nil { + return err + } + body = bytes.NewReader(encoded) + } + request, err := http.NewRequestWithContext(ctx, method, strings.TrimRight(c.BaseURL, "/")+path, body) + if err != nil { + return err + } + request.Header.Set(HeaderProtocol, ProtocolVersion) + request.Header.Set("Accept", "application/json") + if in != nil { + request.Header.Set("Content-Type", "application/json") + } + for key, value := range c.Headers { + request.Header.Set(key, value) + } + response, err := httpClient.Do(request) + if err != nil { + return err + } + defer response.Body.Close() + payload, err := io.ReadAll(io.LimitReader(response.Body, 4<<20)) + if err != nil { + return err + } + if response.StatusCode < 200 || response.StatusCode >= 300 { + return fmt.Errorf("a2a http %d: %s", response.StatusCode, strings.TrimSpace(string(payload))) + } + if out == nil { + return nil + } + return json.Unmarshal(payload, out) +} + +func writeJSON(writer http.ResponseWriter, status int, value any) { + writer.Header().Set("Content-Type", "application/json; charset=utf-8") + writer.WriteHeader(status) + _ = json.NewEncoder(writer).Encode(value) +} diff --git a/a2a/protocol_test.go b/a2a/protocol_test.go new file mode 100644 index 0000000..95e3d1c --- /dev/null +++ b/a2a/protocol_test.go @@ -0,0 +1,42 @@ +package a2a_test + +import ( + "context" + "net/http/httptest" + "testing" + + "github.com/ybszm/langgraph-go/a2a" +) + +func TestServerCardAndSend(t *testing.T) { + handler := &a2a.Server{ + Card: a2a.AgentCard{Name: "demo", Skills: []string{"echo"}}, + Handler: func(_ context.Context, req a2a.SendRequest) (a2a.SendResponse, error) { + return a2a.SendResponse{ + SessionID: req.SessionID, + Message: a2a.Message{Role: "agent", Content: "echo:" + req.Message.Content}, + }, nil + }, + } + server := httptest.NewServer(handler) + defer server.Close() + + client := &a2a.Client{BaseURL: server.URL} + card, err := client.GetCard(context.Background()) + if err != nil { + t.Fatal(err) + } + if card.Name != "demo" { + t.Fatalf("card=%+v", card) + } + resp, err := client.Send(context.Background(), a2a.SendRequest{ + SessionID: "s1", + Message: a2a.Message{Content: "ping"}, + }) + if err != nil { + t.Fatal(err) + } + if resp.Message.Content != "echo:ping" || resp.SessionID != "s1" { + t.Fatalf("resp=%+v", resp) + } +} diff --git a/docs/COMPARISON.md b/docs/COMPARISON.md new file mode 100644 index 0000000..94e74d8 --- /dev/null +++ b/docs/COMPARISON.md @@ -0,0 +1,61 @@ +# Comparison with other Go agent frameworks + +This document positions **langgraph-go** relative to common Go LLM/agent stacks. +It is a product and architecture guide, not a competitive attack. + +## One-line positions + +| Project | Position | +|---|---| +| **langgraph-go** | Typed, durable **graph runtime** inspired by LangGraph | +| **LangChain Go** ([langchaingo](https://github.com/tmc/langchaingo)) | Broad **LLM application toolkit** (chains, RAG, loaders, many integrations) | +| **Google ADK-Go** | Google **Agent Development Kit** — agent product skeleton and multi-agent patterns | +| **tRPC-Agent-Go** | Production agent platform in the **tRPC** ecosystem (tools, MCP/A2A, eval, observability) | +| **go-agent** (lightweight libraries) | Minimal ReAct / tool loops with few dependencies | + +## Capability matrix + +| Dimension | langgraph-go | LangChain Go | ADK-Go | tRPC-Agent-Go | light go-agent | +|---|---|---|---|---|---| +| Graph / state machine | Strong (BSP, Send, Command, subgraphs) | Medium | Medium | Strong | Weak | +| Typed state | Strong (`StateGraph[S,D]`) | Medium | Medium | Medium | Medium–weak | +| Checkpoint / resume / time travel | Strong | Limited | Session-focused | Present | Usually weak | +| Streaming modes | Strong | Medium | Medium | Medium–strong | Varies | +| Multi-agent | Present (supervisor/router/handoff/deep) | Composable | Design focus | Design focus | Rare | +| Model / RAG ecosystem | Focused adapters | Very broad | Google-leaning | Broad (cloud + OSS) | Varies | +| MCP / A2A | MCP module + A2A subset package | Community | Ecosystem | First-class | Rare | +| Evaluation | `eval` golden trajectories | Community | Medium | First-class | Rare | +| RPC binding | Optional remote / Temporal | None | Medium | tRPC-native | None | +| Dependency coupling | Core stays lean | App-level | Vendor ecosystem | tRPC ecosystem | Minimal | + +## When to choose langgraph-go + +- Long-running workflows that must **interrupt, resume, and fork history** +- Deterministic concurrent super-steps with typed reducers +- Multi-sub-agent orchestration that shares the same durable runtime +- You want **LangGraph-class semantics in Go** without binding to one cloud or RPC stack + +## When another stack fits better + +- Fast RAG / document pipelines / widest model catalog → **langchaingo** +- Google-centric agent product templates → **ADK-Go** +- tRPC microservices + A2A/MCP platform + evaluation suite → **tRPC-Agent-Go** +- Ten-line ReAct demo → lightweight **go-agent** + +## Interop strategy + +langgraph-go is designed as the **execution core**. Models and data loaders can +come from other libraries: + +- Bridge messages with [LANGCHAINGO.md](LANGCHAINGO.md) +- Secure MCP tools with `prebuilt.GuardTools` + [mcpclient](../mcpclient/README.md) +- Expose agents over HTTP with [remote](../remote/README.md) or [a2a](../a2a/README.md) + +## Honest gaps (tracked) + +See [COMPATIBILITY.md](../COMPATIBILITY.md), [DURABILITY.md](DURABILITY.md), and +the roadmap notes in [CHANGELOG.md](../CHANGELOG.md) Unreleased. Priority themes: + +1. Broader ecosystem examples (langchaingo / enterprise auth) +2. Evaluation and multi-agent protocols +3. Deeper deferred-node scheduling and remote SDK surface diff --git a/docs/DURABILITY.md b/docs/DURABILITY.md index 0f4228f..a34e8f6 100644 --- a/docs/DURABILITY.md +++ b/docs/DURABILITY.md @@ -23,11 +23,20 @@ in-flight state. ## Mapping from Python durability modes +`graph.RunConfig.Durability` accepts: + +| Value | Behavior | +|---|---| +| `""` / `DurabilityUnspecified` | Same as Sync when a checkpointer is configured | +| `DurabilitySync` (`"sync"`) | Supported: commit at super-step boundaries | +| `DurabilityAsync` (`"async"`) | Returns `ErrUnsupportedDurability` | +| `DurabilityExit` (`"exit"`) | Returns `ErrUnsupportedDurability` | + | Python concept | Closest Go behavior | |---|---| -| `sync` (checkpoint after every super-step) | Default when `WithPersistence` is configured: the runtime commits at super-step boundaries before scheduling the next work set. | -| `async` (checkpoint in the background) | Not offered. Background checkpointing would require an explicit outbox/worker design (see `backend/distributed` for multi-process equivalents). | -| `exit` (checkpoint only on exit/interrupt) | Not offered as a mode. Interrupt paths always require durable markers; normal completion still writes the final checkpoint under persistence. | +| `sync` | `DurabilitySync` / default with `WithPersistence` | +| `async` | Not implemented (use `backend/distributed` for multi-worker async IO) | +| `exit` | Not implemented as a mode | ## Practical guidance diff --git a/docs/LANGCHAINGO.md b/docs/LANGCHAINGO.md new file mode 100644 index 0000000..32a9e58 --- /dev/null +++ b/docs/LANGCHAINGO.md @@ -0,0 +1,86 @@ +# Using langgraph-go with langchaingo + +[langchaingo](https://github.com/tmc/langchaingo) is a popular Go toolkit for +models, embeddings, and document pipelines. **langgraph-go does not replace it.** +Use langchaingo (or any provider SDK) for model I/O, and langgraph-go for durable +graph execution, interrupts, and multi-agent orchestration. + +## Recommended split + +| Concern | Prefer | +|---|---| +| Chat completions, embeddings, loaders, vector stores | langchaingo | +| Typed graph, checkpoint, resume, fan-out, supervisors | langgraph-go | +| Tool allowlists / human approval | `prebuilt.GuardTools` | +| Long-term memory in the graph | langgraph-go `store` + optional retrieval | + +## Message bridge pattern + +langgraph-go agents speak `prebuilt.Message` / `AgentState`. Adapters map +provider messages into that shape. A typical langchaingo bridge: + +```go +// Pseudo-code: adapt langchaingo llms.MessageContent into prebuilt.AgentState. +// Keep this adapter in your application module so core langgraph-go stays free +// of a hard langchaingo dependency. + +import ( + "context" + + "github.com/tmc/langchaingo/llms" + "github.com/ybszm/langgraph-go/graph" + "github.com/ybszm/langgraph-go/prebuilt" +) + +type LangChainGoModel struct { + Model llms.Model +} + +func (m LangChainGoModel) Invoke( + ctx context.Context, + state prebuilt.AgentState, + _ graph.Runtime, +) (prebuilt.AssistantMessage, error) { + messages, err := state.ProviderMessages() + if err != nil { + return prebuilt.AssistantMessage{}, err + } + // Convert prebuilt.Message → []llms.MessageContent in your app. + content, err := toLangChainMessages(messages) + if err != nil { + return prebuilt.AssistantMessage{}, err + } + resp, err := m.Model.GenerateContent(ctx, content) + if err != nil { + return prebuilt.AssistantMessage{}, err + } + return fromLangChainResponse(resp) +} +``` + +Then: + +```go +agent, err := prebuilt.NewQuickAgent(prebuilt.QuickAgentConfig{ + Model: LangChainGoModel{Model: openaiModel}, + Tools: tools, +}) +``` + +## Tools + +- Prefer `prebuilt.Tool` / `ToolFunc` for graph-native tools. +- Wrap external HTTP tools with timeouts in the `Run` closure. +- Apply `prebuilt.GuardTools` before binding tools to the agent. + +## What not to do + +- Do not import langchaingo from langgraph-go core packages (keeps the module lean). +- Do not expect Python LangGraph + Python LangChain APIs to map 1:1. + +## Minimal checklist + +1. Implement `prebuilt.ChatModel[prebuilt.AgentState]` over your langchaingo model. +2. Build with `NewQuickAgent` or `NewChatModelAgent`. +3. Add `graph.WithPersistence` when you need resume. +4. Stream with `AgentRunner.Query` for application UIs. diff --git a/docs/README.md b/docs/README.md index a18fa35..784dd1d 100644 --- a/docs/README.md +++ b/docs/README.md @@ -11,6 +11,10 @@ | [Compatibility](../COMPATIBILITY.md) | Checking behavior against upstream LangGraph | | [Remote server](../remote/README.md) | HTTP/SSE, durable streams, reconnect | | [Publishing](PUBLISHING.md) | Safe GitHub release / consumer compatibility | +| [Versioning](VERSIONING.md) | Pre-1.0 SemVer promises | +| [Comparison](COMPARISON.md) | vs LangChain Go, ADK-Go, tRPC-Agent-Go, light go-agent | +| [langchaingo interop](LANGCHAINGO.md) | Split models vs durable graph | +| [A2A subset](../a2a/README.md) | Minimal agent-to-agent HTTP | All examples compile as part of `go test ./...`; none require provider credentials or external services. diff --git a/docs/VERSIONING.md b/docs/VERSIONING.md new file mode 100644 index 0000000..2fc67fd --- /dev/null +++ b/docs/VERSIONING.md @@ -0,0 +1,44 @@ +# Pre-1.0 versioning policy + +langgraph-go follows [Semantic Versioning](https://semver.org/) with explicit +pre-1.0 rules so consumers can upgrade safely. + +## Guarantees + +| Release type | Promise | +|---|---| +| **Patch** (`v0.1.x`) | No breaking changes to exported APIs or documented on-disk formats without a security exception. | +| **Minor** (`v0.x.0`, x>1) | May add APIs; may change APIs **only** with CHANGELOG + migration notes. Prefer deprecation for one minor when possible. | +| **v1.0.0** | Stability bar: breaking changes require major versions thereafter. | + +## Compatibility of persisted data + +- Checkpoint codecs carry `Type` + `Version` fields; bump codec versions when + encodings change incompatibly. +- Remote protocol version is advertised in `LangGraph-Protocol-Version`. +- Database migrations for Postgres/SQLite adapters are additive when possible; + incompatible migrations are called out in the changelog. + +## What “supported” means at v0.1 + +- CI (unit, race on core packages, optional integration jobs) must pass on `main`. +- Documented examples build without credentials. +- Security reports follow [SECURITY.md](../SECURITY.md). + +## What is not promised yet + +- Full Python LangGraph API parity ([COMPATIBILITY.md](../COMPATIBILITY.md)) +- Long-term support of old minors (only latest `main` receives security fixes + until an LTS policy is published) + +## Consumer guidance + +```bash +# Pin a version +go get github.com/ybszm/langgraph-go@v0.1.0 + +# Upgrade deliberately +go get github.com/ybszm/langgraph-go@v0.2.0 +``` + +Avoid `@latest` in production modules without review. diff --git a/eval/trajectory.go b/eval/trajectory.go new file mode 100644 index 0000000..a115ca8 --- /dev/null +++ b/eval/trajectory.go @@ -0,0 +1,120 @@ +// Package eval provides minimal golden-trajectory assertions for agents and +// graphs. It stays dependency-free so CI can run without external model calls. +package eval + +import ( + "fmt" + "strings" + "testing" +) + +// Step records one observed model/tool/node boundary. +type Step struct { + // Kind is free-form; common values: "model", "tool", "node", "final". + Kind string `json:"kind"` + // Name is a tool name, node ID, or model label. + Name string `json:"name,omitempty"` + // Content is optional textual payload (assistant text, tool args summary). + Content string `json:"content,omitempty"` +} + +// Trajectory is an ordered list of steps captured during a run. +type Trajectory []Step + +// ExpectTools asserts that tool names appear in order as a subsequence of the +// trajectory (other steps may intervene). +func ExpectTools(traj Trajectory, tools ...string) error { + if len(tools) == 0 { + return nil + } + index := 0 + for _, step := range traj { + if !strings.EqualFold(step.Kind, "tool") { + continue + } + if strings.EqualFold(step.Name, tools[index]) { + index++ + if index == len(tools) { + return nil + } + } + } + return fmt.Errorf("eval: tool sequence %v not found in trajectory (matched %d)", tools, index) +} + +// ExpectAnyTool asserts at least one of the tool names was invoked. +func ExpectAnyTool(traj Trajectory, tools ...string) error { + set := map[string]struct{}{} + for _, name := range tools { + set[strings.ToLower(name)] = struct{}{} + } + for _, step := range traj { + if strings.EqualFold(step.Kind, "tool") { + if _, ok := set[strings.ToLower(step.Name)]; ok { + return nil + } + } + } + return fmt.Errorf("eval: none of tools %v appeared", tools) +} + +// ExpectNoTool asserts no tool step is present. +func ExpectNoTool(traj Trajectory) error { + for _, step := range traj { + if strings.EqualFold(step.Kind, "tool") { + return fmt.Errorf("eval: unexpected tool %q", step.Name) + } + } + return nil +} + +// ExpectFinalContains asserts the last final/model content contains substr. +func ExpectFinalContains(traj Trajectory, substr string) error { + for i := len(traj) - 1; i >= 0; i-- { + step := traj[i] + if strings.EqualFold(step.Kind, "final") || strings.EqualFold(step.Kind, "model") { + if strings.Contains(step.Content, substr) { + return nil + } + return fmt.Errorf("eval: final content %q does not contain %q", step.Content, substr) + } + } + return fmt.Errorf("eval: no final/model step in trajectory") +} + +// ToolNames returns tool step names in order. +func (traj Trajectory) ToolNames() []string { + names := make([]string, 0) + for _, step := range traj { + if strings.EqualFold(step.Kind, "tool") { + names = append(names, step.Name) + } + } + return names +} + +// RequireTools is a testing helper around ExpectTools. +func RequireTools(t testing.TB, traj Trajectory, tools ...string) { + t.Helper() + if err := ExpectTools(traj, tools...); err != nil { + t.Fatal(err) + } +} + +// RequireFinalContains is a testing helper around ExpectFinalContains. +func RequireFinalContains(t testing.TB, traj Trajectory, substr string) { + t.Helper() + if err := ExpectFinalContains(traj, substr); err != nil { + t.Fatal(err) + } +} + +// FromAssistantAndTools builds a simple trajectory for offline tests. +func FromAssistantAndTools(assistant string, tools ...string) Trajectory { + traj := Trajectory{{Kind: "model", Name: "assistant", Content: assistant}} + for _, name := range tools { + traj = append(traj, Step{Kind: "tool", Name: name}) + } + traj = append(traj, Step{Kind: "final", Content: assistant}) + return traj +} diff --git a/eval/trajectory_test.go b/eval/trajectory_test.go new file mode 100644 index 0000000..72e44dd --- /dev/null +++ b/eval/trajectory_test.go @@ -0,0 +1,34 @@ +package eval_test + +import ( + "testing" + + "github.com/ybszm/langgraph-go/eval" +) + +func TestExpectToolsSubsequence(t *testing.T) { + traj := eval.Trajectory{ + {Kind: "model"}, + {Kind: "tool", Name: "search"}, + {Kind: "model"}, + {Kind: "tool", Name: "calc"}, + {Kind: "final", Content: "answer 42"}, + } + if err := eval.ExpectTools(traj, "search", "calc"); err != nil { + t.Fatal(err) + } + if err := eval.ExpectTools(traj, "calc", "search"); err == nil { + t.Fatal("expected order failure") + } + eval.RequireFinalContains(t, traj, "42") + eval.RequireTools(t, traj, "search") +} + +func TestExpectNoTool(t *testing.T) { + traj := eval.FromAssistantAndTools("hi only") + // FromAssistantAndTools with no tools still has no tool steps before final + traj = eval.Trajectory{{Kind: "model", Content: "hi"}, {Kind: "final", Content: "hi"}} + if err := eval.ExpectNoTool(traj); err != nil { + t.Fatal(err) + } +} diff --git a/examples/README.md b/examples/README.md index 030b572..e39bf72 100644 --- a/examples/README.md +++ b/examples/README.md @@ -10,6 +10,7 @@ does not require credentials or external services. |---|---| | `agent-runner` | High-level Run/Query events for every built-in agent harness | | `quick-agent` | Smallest model + optional tools + system prompt facade | +| `mcp-secure` | MCP-style tools + allowlist guard pattern (offline) | | `chat-model-agent` | Minimal model + tool agent with default state | | `deep-agent` | Structured planning and automatic general-purpose delegation | | `multi-agent` | Specialized workers and parallel supervisor coordination | diff --git a/examples/mcp-secure/main.go b/examples/mcp-secure/main.go new file mode 100644 index 0000000..2cf1cdc --- /dev/null +++ b/examples/mcp-secure/main.go @@ -0,0 +1,65 @@ +// Example: secure MCP tool binding pattern (offline). +// +// This program does not dial a real MCP server. It shows the production +// pattern: discover tools → allowlist/deny via GuardTools → attach to QuickAgent. +package main + +import ( + "context" + "fmt" + + "github.com/ybszm/langgraph-go/graph" + "github.com/ybszm/langgraph-go/prebuilt" +) + +// In production: +// +// client, err := mcpclient.ConnectHTTP(ctx, os.Getenv("MCP_URL"), nil, mcpclient.Options{}) +// tools, err := mcpclient.Tools[prebuilt.AgentState, prebuilt.AgentDelta](ctx, client) +// tools, err = prebuilt.GuardTools(tools, prebuilt.ToolGuardPolicy{ +// Allowed: []string{"get_metrics", "search_logs"}, +// Denied: []string{"run_shell"}, +// }) + +type demoModel struct{} + +func (demoModel) Invoke(_ context.Context, _ prebuilt.AgentState, _ graph.Runtime) (prebuilt.AssistantMessage, error) { + return prebuilt.AssistantMessage{Content: "mcp-secure pattern ready"}, nil +} + +type demoTool struct{ name string } + +func (t demoTool) Name() string { return t.name } +func (t demoTool) Invoke(context.Context, prebuilt.ToolCall, prebuilt.ToolRuntime[prebuilt.AgentState]) (prebuilt.ToolResult[prebuilt.AgentDelta], error) { + return prebuilt.TextResult[prebuilt.AgentDelta]("ok"), nil +} + +func main() { + raw := []prebuilt.Tool[prebuilt.AgentState, prebuilt.AgentDelta]{ + demoTool{name: "get_metrics"}, + demoTool{name: "run_shell"}, + } + tools, err := prebuilt.GuardTools(raw, prebuilt.ToolGuardPolicy{ + Allowed: []string{"get_metrics"}, + Denied: []string{"run_shell"}, + }) + if err != nil { + panic(err) + } + // run_shell remains in slice but policy denies invocation at call time. + _, _ = tools[1].Invoke(context.Background(), prebuilt.ToolCall{ID: "1", Name: "run_shell"}, prebuilt.ToolRuntime[prebuilt.AgentState]{}) + + agent, err := prebuilt.NewQuickAgent(prebuilt.QuickAgentConfig{ + Model: demoModel{}, + Tools: tools[:1], + }) + if err != nil { + panic(err) + } + state, err := agent.Run(context.Background(), "status?", graph.RunConfig{}) + if err != nil { + panic(err) + } + fmt.Println(state.FinalResponse()) + fmt.Println("tools after guard:", len(tools)) +} diff --git a/graph/bench_test.go b/graph/bench_test.go new file mode 100644 index 0000000..96bb665 --- /dev/null +++ b/graph/bench_test.go @@ -0,0 +1,57 @@ +package graph_test + +import ( + "context" + "testing" + + "github.com/ybszm/langgraph-go/graph" +) + +func BenchmarkLinearInvoke(b *testing.B) { + builder := graph.NewStateGraph(testReducer) + _ = builder.AddNode("n", func(context.Context, testState, graph.Runtime) (graph.Command[testDelta], error) { + return graph.Update(testDelta{Add: 1}), nil + }) + _ = builder.AddEdge(graph.START, "n") + _ = builder.AddEdge("n", graph.END) + compiled, err := builder.Compile() + if err != nil { + b.Fatal(err) + } + ctx := context.Background() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := compiled.Invoke(ctx, testState{}, graph.RunConfig{}); err != nil { + b.Fatal(err) + } + } +} + +func BenchmarkFanOutThree(b *testing.B) { + builder := graph.NewStateGraph(testReducer) + for _, id := range []graph.NodeID{"a", "b", "c"} { + name := id + _ = builder.AddNode(name, func(context.Context, testState, graph.Runtime) (graph.Command[testDelta], error) { + return graph.Update(testDelta{Add: 1, Label: string(name)}), nil + }) + } + _ = builder.AddEdge(graph.START, "a") + _ = builder.AddEdge(graph.START, "b") + _ = builder.AddEdge(graph.START, "c") + _ = builder.AddEdge("a", graph.END) + _ = builder.AddEdge("b", graph.END) + _ = builder.AddEdge("c", graph.END) + compiled, err := builder.Compile() + if err != nil { + b.Fatal(err) + } + ctx := context.Background() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := compiled.Invoke(ctx, testState{}, graph.RunConfig{}); err != nil { + b.Fatal(err) + } + } +} diff --git a/graph/builder.go b/graph/builder.go index fb060ba..e68429a 100644 --- a/graph/builder.go +++ b/graph/builder.go @@ -61,6 +61,7 @@ type StateGraph[S, D any] struct { nodeChannelReads map[NodeID][]string nodeChannelTriggers map[NodeID][]string dynamicInterruptNodes map[NodeID]struct{} + deferredNodes map[NodeID]struct{} } // NewStateGraph creates an empty graph using reducer for super-step updates. @@ -82,6 +83,7 @@ func NewStateGraph[S, D any](reducer Reducer[S, D]) *StateGraph[S, D] { nodeChannelReads: make(map[NodeID][]string), nodeChannelTriggers: make(map[NodeID][]string), dynamicInterruptNodes: make(map[NodeID]struct{}), + deferredNodes: make(map[NodeID]struct{}), } } @@ -193,6 +195,12 @@ func (g *StateGraph[S, D]) AddNode(id NodeID, node Node[S, D], options ...NodeOp if configured.dynamicInterrupt { g.dynamicInterruptNodes[id] = struct{}{} } + if configured.deferred { + if g.deferredNodes == nil { + g.deferredNodes = make(map[NodeID]struct{}) + } + g.deferredNodes[id] = struct{}{} + } if configured.errorHandler != nil { handler, ok := configured.errorHandler.(ErrorHandler[S, D]) if !ok { @@ -630,6 +638,7 @@ func (g *StateGraph[S, D]) Compile( nodeChannelReads: cloneNodeChannelReads(g.nodeChannelReads), nodeChannelTriggers: cloneNodeChannelReads(g.nodeChannelTriggers), dynamicInterruptNodes: cloneNodeSet(g.dynamicInterruptNodes), + deferredNodes: cloneNodeSet(g.deferredNodes), runLocks: newRunLockSet(), }, nil } diff --git a/graph/compiled.go b/graph/compiled.go index 4040bf1..821ac0f 100644 --- a/graph/compiled.go +++ b/graph/compiled.go @@ -55,6 +55,7 @@ type CompiledGraph[S, D any] struct { nodeChannelReads map[NodeID][]string nodeChannelTriggers map[NodeID][]string dynamicInterruptNodes map[NodeID]struct{} + deferredNodes map[NodeID]struct{} runLocks *runLockSet } @@ -223,6 +224,15 @@ func (g *CompiledGraph[S, D]) runInternal( ErrInvalidRunConfig, ) } + switch config.Durability { + case DurabilityUnspecified, DurabilitySync: + // supported (sync is the persistent runtime default) + case DurabilityAsync, DurabilityExit: + return input, fmt.Errorf("%w: %q (only %q is implemented; see docs/DURABILITY.md)", + ErrUnsupportedDurability, config.Durability, DurabilitySync) + default: + return input, fmt.Errorf("%w: unknown durability %q", ErrInvalidRunConfig, config.Durability) + } if g.persistence == nil && (config.CheckpointID != "" || config.CheckpointNamespace != "") { return input, fmt.Errorf( "%w: checkpoint selection requires a configured checkpointer", @@ -2031,6 +2041,17 @@ func (g *CompiledGraph[S, D]) scheduleDefaultTasks( return tasks, nil } +// IsDeferred reports whether a node was registered with WithDeferred. +// Runtime still schedules deferred nodes with peers today; full deferred +// barrier semantics remain Partial (COMPATIBILITY.md). +func (g *CompiledGraph[S, D]) IsDeferred(id NodeID) bool { + if g == nil { + return false + } + _, ok := g.deferredNodes[id] + return ok +} + func branchToTrigger(destination NodeID) NodeID { return NodeID("branch:to:" + string(destination)) } diff --git a/graph/deferred_durability_test.go b/graph/deferred_durability_test.go new file mode 100644 index 0000000..5c5ce69 --- /dev/null +++ b/graph/deferred_durability_test.go @@ -0,0 +1,69 @@ +package graph_test + +import ( + "context" + "errors" + "testing" + + "github.com/ybszm/langgraph-go/graph" +) + +func TestWithDeferredIsRecorded(t *testing.T) { + builder := graph.NewStateGraph(testReducer) + if err := builder.AddNode("ready", func(context.Context, testState, graph.Runtime) (graph.Command[testDelta], error) { + return graph.Update(testDelta{Add: 1}), nil + }); err != nil { + t.Fatal(err) + } + if err := builder.AddNode("later", func(context.Context, testState, graph.Runtime) (graph.Command[testDelta], error) { + return graph.Update(testDelta{Add: 2}), nil + }, graph.WithDeferred()); err != nil { + t.Fatal(err) + } + if err := builder.AddEdge(graph.START, "ready"); err != nil { + t.Fatal(err) + } + if err := builder.AddEdge("ready", "later"); err != nil { + t.Fatal(err) + } + if err := builder.AddEdge("later", graph.END); err != nil { + t.Fatal(err) + } + compiled, err := builder.Compile() + if err != nil { + t.Fatal(err) + } + if compiled.IsDeferred("later") != true || compiled.IsDeferred("ready") { + t.Fatalf("deferred flags wrong") + } + out, err := compiled.Invoke(context.Background(), testState{}, graph.RunConfig{}) + if err != nil { + t.Fatal(err) + } + if out.Total != 3 { + t.Fatalf("total=%d", out.Total) + } +} + +func TestDurabilityUnsupported(t *testing.T) { + builder := graph.NewStateGraph(testReducer) + if err := builder.AddNode("n", func(context.Context, testState, graph.Runtime) (graph.Command[testDelta], error) { + return graph.NoCommand[testDelta](), nil + }); err != nil { + t.Fatal(err) + } + if err := builder.AddEdge(graph.START, "n"); err != nil { + t.Fatal(err) + } + if err := builder.AddEdge("n", graph.END); err != nil { + t.Fatal(err) + } + compiled, err := builder.Compile() + if err != nil { + t.Fatal(err) + } + _, err = compiled.Invoke(context.Background(), testState{}, graph.RunConfig{Durability: graph.DurabilityAsync}) + if !errors.Is(err, graph.ErrUnsupportedDurability) { + t.Fatalf("err=%v", err) + } +} diff --git a/graph/errors.go b/graph/errors.go index 1175a5f..7023d93 100644 --- a/graph/errors.go +++ b/graph/errors.go @@ -16,6 +16,9 @@ var ( ErrUnknownNode = errors.New("unknown node") // ErrInvalidRunConfig indicates an invalid execution configuration. ErrInvalidRunConfig = errors.New("invalid run config") + // ErrUnsupportedDurability indicates a Durability mode that is not + // implemented by this runtime (async/exit today). + ErrUnsupportedDurability = errors.New("unsupported durability mode") // ErrRecursionLimit indicates that execution exceeded its super-step limit. ErrRecursionLimit = errors.New("graph recursion limit reached") // ErrCheckpointerRequired indicates that an API needs persistence but the diff --git a/graph/retry.go b/graph/retry.go index 938948f..d1df648 100644 --- a/graph/retry.go +++ b/graph/retry.go @@ -34,6 +34,21 @@ type nodeOptions struct { channelReads []string channelTriggers []string dynamicInterrupt bool + deferred bool +} + +// WithDeferred marks a node as deferred for introspection and future +// scheduling policies. The marker is stored on the compiled graph +// (`CompiledGraph.IsDeferred`). Full LangGraph deferred barrier scheduling is +// still Partial — see COMPATIBILITY.md and docs/DURABILITY.md. +func WithDeferred() NodeOption { + return func(options *nodeOptions) error { + if options.deferred { + return fmt.Errorf("%w: duplicate deferred declaration", ErrInvalidGraph) + } + options.deferred = true + return nil + } } // WithDynamicInterrupts declares that a node may call AwaitResume. The diff --git a/graph/types.go b/graph/types.go index 3d7d96d..cf09aa4 100644 --- a/graph/types.go +++ b/graph/types.go @@ -274,6 +274,21 @@ func UpdateAndGoto[D any](delta D, destinations ...NodeID) Command[D] { } } +// Durability selects checkpoint aggressiveness for a run. Empty defaults to +// Sync when a checkpointer is configured. See docs/DURABILITY.md. +type Durability string + +const ( + // DurabilityUnspecified uses Sync when persistence is configured. + DurabilityUnspecified Durability = "" + // DurabilitySync commits checkpoints at super-step boundaries (supported). + DurabilitySync Durability = "sync" + // DurabilityAsync is reserved; requesting it returns ErrUnsupportedDurability. + DurabilityAsync Durability = "async" + // DurabilityExit is reserved; requesting it returns ErrUnsupportedDurability. + DurabilityExit Durability = "exit" +) + // RunConfig controls a single graph execution. type RunConfig struct { // Context contains immutable dependencies scoped to this invocation. It is @@ -294,6 +309,8 @@ type RunConfig struct { // MaxConcurrency limits simultaneously executing nodes. Zero means the // number of tasks in the current super-step. A negative value is invalid. MaxConcurrency int + // Durability controls checkpoint timing. Only Sync is implemented today. + Durability Durability // ThreadID selects the durable execution thread when the graph was // compiled with a checkpointer. ThreadID string diff --git a/httpware/middleware.go b/httpware/middleware.go new file mode 100644 index 0000000..8568e95 --- /dev/null +++ b/httpware/middleware.go @@ -0,0 +1,121 @@ +// Package httpware provides small, optional HTTP middleware helpers for agent +// servers (remote, a2a, or custom). It does not depend on tRPC or cloud SDKs. +package httpware + +import ( + "context" + "net/http" + "strings" + "sync" + "time" +) + +// BearerAuth rejects requests missing a matching Authorization bearer token. +// Empty token disables the middleware (always allows). +func BearerAuth(token string) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + if strings.TrimSpace(token) == "" { + return next + } + want := "Bearer " + token + return http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if request.Header.Get("Authorization") != want { + http.Error(writer, "unauthorized", http.StatusUnauthorized) + return + } + next.ServeHTTP(writer, request) + }) + } +} + +// SessionFromHeader copies a header value into request context via a callback +// key function. Prefer application-specific context keys in real services. +type contextKey string + +// SessionIDHeader is the default session correlation header. +const SessionIDHeader = "X-Session-ID" + +// SessionIDKey is the context key for the session id middleware. +const SessionIDKey contextKey = "langgraph-go/session-id" + +// WithSessionID stores request.Header[SessionIDHeader] on the context when set. +func WithSessionID(next http.Handler) http.Handler { + return http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if id := strings.TrimSpace(request.Header.Get(SessionIDHeader)); id != "" { + request = request.WithContext(context.WithValue(request.Context(), SessionIDKey, id)) + } + next.ServeHTTP(writer, request) + }) +} + +// RateLimit is a trivial token-bucket per remote IP for demos and local gateways. +// It is not a distributed limiter. +type RateLimit struct { + mu sync.Mutex + interval time.Duration + burst int + clients map[string]*bucket +} + +type bucket struct { + tokens float64 + last time.Time +} + +// NewRateLimit allows burst events, refilling one token every interval. +func NewRateLimit(interval time.Duration, burst int) *RateLimit { + if interval <= 0 { + interval = time.Second + } + if burst <= 0 { + burst = 1 + } + return &RateLimit{interval: interval, burst: burst, clients: make(map[string]*bucket)} +} + +// Middleware enforces the limit. +func (r *RateLimit) Middleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if r == nil { + next.ServeHTTP(writer, request) + return + } + key := request.RemoteAddr + if !r.allow(key, time.Now()) { + http.Error(writer, "rate limit exceeded", http.StatusTooManyRequests) + return + } + next.ServeHTTP(writer, request) + }) +} + +func (r *RateLimit) allow(key string, now time.Time) bool { + r.mu.Lock() + defer r.mu.Unlock() + b := r.clients[key] + if b == nil { + r.clients[key] = &bucket{tokens: float64(r.burst - 1), last: now} + return true + } + elapsed := now.Sub(b.last).Seconds() + b.tokens += elapsed / r.interval.Seconds() + if b.tokens > float64(r.burst) { + b.tokens = float64(r.burst) + } + b.last = now + if b.tokens < 1 { + return false + } + b.tokens-- + return true +} + +// Chain applies middlewares in order (first is outermost). +func Chain(handler http.Handler, middlewares ...func(http.Handler) http.Handler) http.Handler { + for i := len(middlewares) - 1; i >= 0; i-- { + if middlewares[i] != nil { + handler = middlewares[i](handler) + } + } + return handler +} diff --git a/httpware/middleware_test.go b/httpware/middleware_test.go new file mode 100644 index 0000000..4944da8 --- /dev/null +++ b/httpware/middleware_test.go @@ -0,0 +1,47 @@ +package httpware_test + +import ( + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/ybszm/langgraph-go/httpware" +) + +func TestBearerAuth(t *testing.T) { + h := httpware.BearerAuth("secret")(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNoContent) + })) + req := httptest.NewRequest(http.MethodGet, "/", nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusUnauthorized { + t.Fatalf("code=%d", rec.Code) + } + req.Header.Set("Authorization", "Bearer secret") + rec = httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusNoContent { + t.Fatalf("code=%d", rec.Code) + } +} + +func TestRateLimit(t *testing.T) { + limiter := httpware.NewRateLimit(time.Hour, 1) + h := limiter.Middleware(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + })) + req := httptest.NewRequest(http.MethodGet, "/", nil) + req.RemoteAddr = "1.2.3.4:9" + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("first=%d", rec.Code) + } + rec = httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusTooManyRequests { + t.Fatalf("second=%d", rec.Code) + } +}