diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000000..d6465eb3d0 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,8 @@ +.git +.dockerignore +.DS_Store +internal/docker/models/ +internal/docker/micro/ +internal/website/ +linux/ +*.gguf diff --git a/.gitignore b/.gitignore index abb71641e3..7cdf294a55 100644 --- a/.gitignore +++ b/.gitignore @@ -58,6 +58,15 @@ examples/mcp/hello/hello .DS_Store /micro +# Local docker compose runtime data (LLM model files, mounted home dirs) +/internal/docker/models/ +/internal/docker/micro/ + +# Observability stack runtime state (grafana/prometheus/tempo data dirs) +/internal/docker/grafana/data/ +/internal/docker/prometheus/data/ +/internal/docker/tempo/data/ + # Built example/harness binaries (go build ./path/... drops these at repo root) /plan-delegate /agent-plan-delegate diff --git a/.micro.Dockerfile b/.micro.Dockerfile new file mode 100644 index 0000000000..6a904d0451 --- /dev/null +++ b/.micro.Dockerfile @@ -0,0 +1,12 @@ +FROM golang:1.25-alpine AS builder +WORKDIR /app +COPY go.mod go.sum ./ +RUN go mod download +COPY . . +ARG SVC +RUN CGO_ENABLED=0 go build -o /service ./$SVC + +FROM scratch +COPY --from=builder /service /service +EXPOSE 8080 +CMD ["/service"] diff --git a/AGENTS.md b/AGENTS.md index 395c13e8e9..0506a0cc50 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,25 +1,21 @@ # Repository agent instructions -These instructions apply to the entire repository. +Repo-wide instructions. ## Pull requests from Codex tasks -When a Codex task makes repository changes and the requested outcome is a PR: +For PR-focused Codex tasks: -1. Keep the change focused on the assigned issue or prompt. -2. Run the relevant verification commands and capture their results - (`go build ./...`, `go test ./...`, `golangci-lint run ./...`). -3. Check `git status --short` and review the diff before finishing. -4. Create a uniquely-named branch under the `codex/` prefix (do not work on - `master`, and do not use a generic name like `work`): +1. Focus changes on issue/prompt. +2. Run verification (`go fmt ./...`, `go build ./...`, `go test ./...`, `golangci-lint run ./...`). +3. Check `git status --short` and diff. +4. Branch under `codex/` (not `master` or `work`): ```sh git switch -c codex/- ``` -5. Stage the intended files and commit on that branch. -6. Open the pull request yourself with the GitHub CLI, which is installed in the - environment and whose `origin` points at this repository, then enable - auto-merge so GitHub merges it once the required CI checks pass: +5. Stage and commit. +6. Open PR via GitHub CLI (`origin` set), enable auto-merge: ```sh git push -u origin HEAD @@ -29,11 +25,6 @@ When a Codex task makes repository changes and the requested outcome is a PR: gh pr merge --squash --auto --delete-branch ``` -The branch should start with `codex/` and the PR should carry the `codex` -label. Auto-merge waits for the required status checks (build, tests, -golangci-lint) — never merge a PR manually before CI is green. +Branch `codex/`, PR label `codex`. Auto-merge waits for CI (build, tests, golangci-lint) — never merge manually before CI green. -Do not just say that a PR was opened, and do **not** rely on the `make_pr` tool: -in this environment `make_pr` only records the title/body and never pushes a -branch or creates a PR. The task is not complete until `gh pr create` has opened -a real pull request and printed its URL. +Don't just claim PR opened. DO NOT use `make_pr` tool (`make_pr` only records title/body, never pushes/creates). Task incomplete until `gh pr create` opens real PR and prints URL. diff --git a/CHANGELOG.md b/CHANGELOG.md index cd0d5dff64..f67c1a3bb6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,7 @@ below is kept current between tags and rolled into the next version when it ship ## [Unreleased] ### Added +- **Handler before/after hooks** — `server.BeforeHandler` and `server.AfterHandler` turn a `func(context.Context, server.Request) error` into a `HandlerWrapper`, giving RPC handlers declarative before/after request hooks to complement the existing start/stop hooks (`service.BeforeStart`/`AfterStart`/`BeforeStop`/`AfterStop`). The maintenance idea from issue #15 ("before/after functions for start, stop and handler requests") is now fully covered. (`server/`) - **Anthropic prompt caching** — the request prefix that never changes (tools + system prompt) is marked with a single `cache_control` breakpoint, so it stops being re-billed at full input rate on every turn and every tool-loop round; with no system prompt the breakpoint moves to the last tool. On by default (the agent tool loop repays the cache write within one Generate); `ai.WithoutCache()` opts out for one-off callers. (`ai/anthropic/`, `ai/`) ### Fixed diff --git a/CODEX.md b/CODEX.md index 47a282819a..6145249a38 100644 --- a/CODEX.md +++ b/CODEX.md @@ -1,121 +1,92 @@ # Codex Maintainer Playbook -Go Micro has six months of Codex access through OpenAI's Codex for Open Source -program. Use it to increase maintainer throughput without changing the project's -bar for review, tests, or design taste. +Go Micro has 6 months OpenAI Codex access. Use it to boost maintainer throughput without lowering bar for review, tests, or design. ## Operating principles -1. **Humans set direction; Codex accelerates execution.** Maintainers choose the - issue, constraints, and acceptance criteria. Codex drafts, investigates, and - verifies. -2. **Small, reviewable changes win.** Prefer focused PRs that can be understood - in one sitting over large speculative rewrites. -3. **Keep the contract green.** Every Codex-assisted change should preserve the - CLI-first getting-started flow, the harnesses, `make test`, and `make lint`. -4. **Document while coding.** If behavior changes, ask Codex to update examples, - guides, and release notes in the same branch. -5. **No blind merges.** Codex output is treated like any contributor output: - reviewed by a maintainer, backed by tests, and checked for public API impact. +1. **Humans set direction; Codex accelerates execution.** Maintainers choose issue/constraints/criteria. Codex drafts/investigates/verifies. +2. **Small, reviewable changes win.** Prefer focused PRs over large speculative rewrites. +3. **Keep the contract green.** Preserve CLI getting-started, harnesses, `make test`, and `make lint`. +4. **Document while coding.** Update examples, guides, release notes in the same branch if behavior changes. +5. **No blind merges.** Review Codex output like any contributor; back by tests, check API impact. ## Coordination with Claude Code -Go Micro is maintained by two AI tools — **Codex** (you) and **Claude Code** (its guide is [CLAUDE.md](CLAUDE.md)) — plus the human maintainer, who routes work and owns every merge. +Go Micro is maintained by Codex and Claude Code (see [CLAUDE.md](CLAUDE.md)) plus human maintainer. -- **Lanes / branches.** You work on `codex/*` branches; Claude Code on `claude/*`. Never push to a branch the other owns, and never have both agents on one branch at once. -- **Base PRs on `master`.** Don't stack a PR on another agent's in-flight branch — if that base squash-merges, your changes can be orphaned. If the code you need isn't merged yet, wait, then branch off `master`. To improve a PR that hasn't merged, push to that PR's branch rather than opening a separate stacked PR — keep the change one mergeable unit. -- **One concern per PR.** Keep each PR single-purpose so a reviewer can read it in one sitting; don't bundle unrelated changes (e.g. a feature plus a docs rebrand). -- **Cross-review before merge.** Claude Code reviews your PRs; you review its with `@codex review`. A fresh pass from the other model catches what the author misses. -- **Dispatch.** Maintainers (or Claude Code) start your tasks with `@codex ` on the relevant issue/PR — that's your context. `@codex review` is review; any other instruction is a *task*. You run one task at a time: take the current one to a clean, green PR before the next is dispatched. -- **CI is the gate.** `go build`, `go test`, `golangci-lint` (blocking), and `make harness` must pass; never merge red. `internal/harness/` and `examples/` are excluded from errcheck; everything else gets the full set. -- **Backlog = GitHub issues**, each a scoped brief with acceptance criteria. +- **Lanes / branches.** Codex on `codex/*`, Claude on `claude/*`. Never share branches. +- **Base PRs on `master`.** Don't stack PRs on other agent branches. Wait or branch off `master`. Push to existing PR branch rather than opening stacked PRs. +- **One concern per PR.** Single-purpose PRs only. +- **Cross-review before merge.** Claude reviews your PRs; you review its with `@codex review`. +- **Dispatch.** Triggered via `@codex ` on issue/PR (`@codex review` is review). One task at a time to a clean, green PR. +- **CI is the gate.** `go build`, `go test`, `golangci-lint`, `make harness` must pass. `internal/harness/` and `examples/` excluded from errcheck. +- **Backlog = GitHub issues** with acceptance criteria. ## Best uses ### 1. PR review and triage -- Summarize a PR: changed surface area, public API impact, tests added or missing. -- Ask for targeted review passes: concurrency, cancellation, security, backwards - compatibility, docs drift, and examples. -- Convert review findings into small patch suggestions or issue comments. +- Summarize PR surface area, API impact, test status. +- Ask for targeted reviews (concurrency, cancellation, security, compat, docs). +- Convert findings into patches or comments. ### 2. Issue reproduction -- Turn bug reports into failing tests or runnable reproduction scripts. -- Minimize flakes by isolating registry, broker, store, transport, and AI-provider - dependencies behind deterministic fakes where possible. -- Attach the exact command that reproduces the failure to the issue. +- Turn bugs into failing tests or repro scripts. +- Isolate fakes for dependencies. +- Attach exact repro command. ### 3. Release support -- Draft changelog entries from merged commits, grouped by feature, fix, docs, and - compatibility notes. -- Check that `README.md`, `ROADMAP.md`, website docs, examples, and `CHANGELOG.md` - agree before tagging. -- Run dry-run release commands and summarize blockers. +- Draft changelogs from commits. +- Verify `README.md`, `ROADMAP.md`, docs, examples, `CHANGELOG.md` agree. +- Run dry-release commands. ### 4. Docs and examples -- Keep the 0→1 path current: scaffold, run, call, chat, inspect. -- Keep the 0→hero example current: a realistic multi-agent system that exercises - agents, services, flows, MCP, A2A, and observability. -- Add runnable examples for new primitives before adding broad prose. +- Keep 0→1 path current. +- Keep 0→hero example current (multi-agent, services, flows, MCP, A2A, observability). +- Add runnable examples before prose. ### 5. Hardening backlog -Use Codex to break roadmap items into small PRs, especially: - -- cross-provider conformance scenarios for all supported AI providers; -- timeout, cancellation, retry, and rate-limit behavior; -- durable agent loops on top of the existing checkpoint model; -- streaming across `ai.Stream` and A2A; -- agent run metadata mapped to OpenTelemetry spans. +Break roadmap items into small PRs: +- cross-provider conformance +- timeout, cancellation, retry, rate-limit +- durable loops +- streaming across `ai.Stream` +- OpenTelemetry spans ## Suggested weekly loop -1. Pick one maintenance lane: reviews, bugs, release prep, docs, or hardening. -2. Ask Codex for a branch-sized plan with acceptance criteria and test commands. -3. Have Codex implement the smallest valuable slice. -4. Run the relevant checks locally and in CI. -5. Review the diff as maintainer-owned code, then merge or send it back. -6. Record any recurring prompt, check, or failure mode in this playbook. - +1. Pick maintenance lane. +2. Get branch plan with criteria/commands. +3. Implement smallest slice. +4. Run checks locally/CI. +5. Review and merge/send back. +6. Log recurring patterns. ## First two weeks -Do not start with a giant feature. Start by making Codex pay rent on maintenance -work that is already on the roadmap and easy to review. +Start with roadmap maintenance. ### Day 1: set up the review loop -1. Pick three recent PRs or commits: one feature, one bug fix, and one docs-only - change. -2. Ask Codex to review each using the PR review template below. -3. Compare Codex findings with maintainer judgment. Keep the checks that found - real issues; delete the noisy ones. -4. Turn the final review prompt into a saved project note or issue comment - template. - -Success means Codex can produce a useful first-pass review in under ten minutes -without blocking a maintainer on false positives. +1. Pick 3 recent PRs (feature, bug, docs). +2. Review using template. +3. Compare with maintainer judgment. +4. Save review prompt. ### Days 2-3: make bugs reproducible -1. Pick one open bug or flaky area. -2. Ask Codex for a failing test only. Do not allow a fix in the first pass. -3. Review the test for whether it captures the real contract. -4. In a second branch, ask Codex to fix the failure with the smallest patch. - -Success means every accepted bug fix starts with a regression test or deterministic -harness case. +1. Pick open bug/flake. +2. Get failing test only (no fix). +3. Verify contract. +4. Fix in second branch with smallest patch. ### Days 4-5: audit the getting-started contract -Run through the 0→1 path from a clean checkout and ask Codex to patch only the -first broken or confusing step. The target is not new prose; it is a runnable -path that works exactly as documented. - -Candidate checks: +Run 0→1 path; patch first break. ```sh make test @@ -125,43 +96,32 @@ go run ./examples/hello-world go run ./internal/harness/universe ``` -### Week 2: choose one roadmap slice +## Week 2: choose one roadmap slice -Pick one hardening item and break it into PRs that each land independently. The -best first slice is usually test infrastructure, not product code. - -Recommended order: - -1. **Provider conformance skeleton**: define one deterministic agent scenario and - gate real-provider runs on credentials. -2. **Cancellation audit**: trace `context.Context` propagation through one package - at a time. -3. **Docs drift audit**: compare `README.md`, `ROADMAP.md`, website docs, and - examples for one shipped feature. -4. **Release checklist dry run**: have Codex build a release-blocker list from the - diff since the previous tag. +Pick hardening item. Recommended order: +1. Provider conformance skeleton. +2. Cancellation audit: trace `context.Context` propagation. +3. Docs drift audit: compare `README.md`, `ROADMAP.md`, website docs, and examples. +4. Release checklist dry run. ## Standing task queue -Keep Codex busy on tasks with clear acceptance criteria: - | Priority | Task | Acceptance criteria | | --- | --- | --- | -| P0 | PR first-pass review | Summary, risks, required changes, and exact verification commands. | -| P0 | Bug reproduction | A failing test or harness case committed before the fix. | -| P0 | 0→1 docs check | Fresh-checkout commands work as written or a patch fixes the first break. | -| P1 | Cross-provider conformance | One scenario runs against fakes by default and real providers when keys exist. | -| P1 | Cancellation hardening | Tests prove timeout/cancel behavior for the touched package. | -| P1 | Release audit | Changelog, docs, examples, and migration notes agree before tagging. | -| P2 | Example polish | Example is runnable, linked from docs, and covered by a lightweight check. | +| P0 | PR first-pass review | Summary, risks, required changes, verification commands. | +| P0 | Bug reproduction | Failing test committed before fix. | +| P0 | 0→1 docs check | Fresh-checkout commands work or patch fixes break. | +| P1 | Cross-provider conformance | Runs against fakes by default, real when keys exist. | +| P1 | Cancellation hardening | Tests prove timeout/cancel behavior. | +| P1 | Release audit | Changelog, docs, examples agree before tagging. | +| P2 | Example polish | Runnable, linked, checked. | ## What not to use Codex for yet -- Broad rewrites without a failing test, benchmark, or public design note. -- Public API changes before a maintainer writes the compatibility story. -- Large generated docs that nobody has run. -- Provider-specific behavior that is not checked against the shared `ai.Model` - contract. +- Broad rewrites without test/benchmark. +- Unvetted public API changes. +- Large unrun docs. +- Provider-specific behavior not checked against `ai.Model`. ## Prompt templates @@ -194,4 +154,4 @@ unless explicitly required, update docs/examples when behavior changes, and run Audit this release branch. Compare CHANGELOG, README, ROADMAP, website docs, and examples against the diff since the last tag. List inconsistencies, missing migration notes, and checks to run before tagging. -``` +``` \ No newline at end of file diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 1b53276958..0558895c78 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -1,10 +1,10 @@ # Contributing to Go Micro -Thank you for your interest in contributing to Go Micro! This document provides guidelines and instructions for contributing. +Guidelines/instructions for contributing. ## Code of Conduct -Be respectful, inclusive, and collaborative. We're all here to build great software together. +Respectful, inclusive, collaborative. ## How Go Micro is built @@ -19,10 +19,10 @@ Human contributions follow the same gate: green CI, one concern per PR. ## Getting Started -1. Fork the repository -2. Clone your fork: `git clone https://github.com/YOUR_USERNAME/go-micro.git` -3. Add upstream remote: `git remote add upstream https://github.com/micro/go-micro.git` -4. Create a feature branch: `git checkout -b feature/my-feature` +1. Fork repo +2. Clone: `git clone https://github.com/YOUR_USERNAME/go-micro.git` +3. Add upstream: `git remote add upstream https://github.com/micro/go-micro.git` +4. Feature branch: `git checkout -b feature/my-feature` ## Development Setup @@ -46,21 +46,21 @@ make lint make fmt ``` -See `make help` for all available commands. +See `make help` for all commands. ## Making Changes ### Code Guidelines -- Follow standard Go conventions (use `gofmt`, `golint`) -- Write clear, descriptive commit messages -- Add tests for new functionality -- Update documentation for API changes -- Keep PRs focused - one feature/fix per PR +- Follow Go conventions (`gofmt`, `golint`) +- Clear commit messages +- Tests for new functionality +- Update docs for API changes +- Focused PRs (one feature/fix per PR) ### Commit Messages -Use conventional commits format: +Conventional commits format: ``` type(scope): subject @@ -73,11 +73,11 @@ footer Types: - `feat`: New feature - `fix`: Bug fix -- `docs`: Documentation changes -- `test`: Test additions/changes -- `refactor`: Code refactoring -- `perf`: Performance improvements -- `chore`: Maintenance tasks +- `docs`: Docs changes +- `test`: Test changes +- `refactor`: Refactoring +- `perf`: Performance +- `chore`: Maintenance Examples: ``` @@ -88,10 +88,10 @@ docs(examples): add streaming example ### Testing -- Write unit tests for all new code +- Unit tests for new code - Ensure existing tests pass -- Add integration tests for plugin implementations -- Test with multiple Go versions (1.20+) +- Integration tests for plugins +- Test Go 1.20+ ```bash # Run specific package tests @@ -110,61 +110,61 @@ richgo test -v ./... ### Documentation -- Update relevant markdown files in `internal/website/docs/` -- Add examples to `internal/website/docs/examples/` for new features +- Update markdown in `internal/website/docs/` +- Add examples to `internal/website/docs/examples/` - Update README.md for major features -- Add godoc comments for exported functions/types +- Add godoc comments for exported items ## Pull Request Process -1. **Update your branch** +1. **Update branch** ```bash git fetch upstream git rebase upstream/master ``` -2. **Run tests and linting** +2. **Run tests and lint** ```bash go test ./... golangci-lint run ``` -3. **Push to your fork** +3. **Push fork** ```bash git push origin feature/my-feature ``` -4. **Create Pull Request** - - Use a descriptive title - - Reference any related issues - - Describe what changed and why - - Add screenshots for UI changes - - Mark as draft if work in progress +4. **Create PR** + - Descriptive title + - Reference issues + - Describe changes/why + - Screenshots for UI + - Mark draft if WIP -5. **PR Review** - - Respond to feedback promptly +5. **Review** + - Respond promptly - Make requested changes - - Re-request review after updates + - Re-request review ### PR Checklist - [ ] Tests pass locally - [ ] Code follows Go conventions -- [ ] Documentation updated -- [ ] Commit messages are clear -- [ ] Branch is up to date with master +- [ ] Docs updated +- [ ] Clear commit messages +- [ ] Branch up to date with master - [ ] No merge conflicts ## Adding Plugins -New plugins should: +New plugins: -1. Live in the appropriate interface directory (e.g., `registry/myplugin/`) -2. Implement the interface completely -3. Include comprehensive tests -4. Provide usage examples -5. Document configuration options (env vars, options) -6. Add to plugin documentation +1. In interface directory (e.g., `registry/myplugin/`) +2. Implement interface fully +3. Comprehensive tests +4. Usage examples +5. Document config (env vars, options) +6. Add plugin docs Example structure: ``` @@ -177,37 +177,32 @@ registry/myplugin/ ## Reporting Issues -Before creating an issue: +Before creating issue: 1. Search existing issues -2. Check documentation -3. Try the latest version +2. Check docs +3. Try latest version -When reporting bugs: -- Use the bug report template -- Include minimal reproduction code +Bug reports: +- Use bug report template +- Minimal reproduction code - Specify versions (Go, Go Micro, plugins) -- Provide relevant logs +- Relevant logs ## Documentation Contributions -Documentation improvements are always welcome! There are several ways to edit the -documentation pages: +Welcome. Ways to edit docs: -- **Fix or edit an existing page**: submit a pull request with your changes — either - directly on the site via a commit + PR, or by proposing changes from a fork of the - repository. -- **Add a new page**: create a new Markdown page under - `internal/website/content/en/docs/` and open a PR as above. +- **Fix/edit existing page**: submit PR. +- **Add new page**: create Markdown under `internal/website/content/en/docs/` and open PR. -The site is built with the Hugo engine and content lives in -`internal/website/content/en`. -For local development instructions (Hugo Extended, `npm ci`, `npm run serve`, production build), see the [internal/website README](internal/website/README.md). +Site built with Hugo, content in `internal/website/content/en`. +Local dev instructions (Hugo Extended, `npm ci`, `npm run serve`, production build): [internal/website README](internal/website/README.md). ## Community -- GitHub Issues: Bug reports and feature requests -- GitHub Discussions: Questions, ideas, and community chat +- GitHub Issues: Bug reports, feature requests +- GitHub Discussions: Questions, ideas, chat - Sponsorship: [GitHub Sponsors](https://github.com/sponsors/micro) ## Release Process @@ -215,7 +210,7 @@ For local development instructions (Hugo Extended, `npm ci`, `npm run serve`, pr Maintainers handle releases: 1. Update CHANGELOG.md -2. Tag release: `git tag -a v5.x.x -m "Release v5.x.x"` +2. Tag: `git tag -a v5.x.x -m "Release v5.x.x"` 3. Push tag: `git push origin v5.x.x` 4. GitHub Actions creates release @@ -223,6 +218,6 @@ Maintainers handle releases: - Check [documentation](internal/website/content/en/docs/) - Browse [examples](internal/website/content/en/docs/examples/) -- Open a [question issue](.github/ISSUE_TEMPLATE/question.md) +- Open [question issue](.github/ISSUE_TEMPLATE/question.md) -Thank you for contributing to Go Micro! 🎉 +Thanks for contributing! 🎉 \ No newline at end of file diff --git a/Dockerfile b/Dockerfile index 225dc3979d..c7eec3c7ab 100644 --- a/Dockerfile +++ b/Dockerfile @@ -21,6 +21,6 @@ COPY $TARGETPLATFORM/micro /usr/local/go/bin/ COPY $TARGETPLATFORM/protoc-gen-micro /usr/local/go/bin/ WORKDIR /micro -EXPOSE 8080 +EXPOSE 3000 8080 ENTRYPOINT ["/usr/local/go/bin/micro"] CMD ["gateway"] diff --git a/Makefile b/Makefile index 991f3665e7..9c7c5c2ae7 100644 --- a/Makefile +++ b/Makefile @@ -1,7 +1,7 @@ NAME = micro GIT_COMMIT = $(shell git rev-parse --short HEAD) GIT_TAG = $(shell git describe --abbrev=0 --tags --always --match "v*") -GIT_IMPORT = go-micro.dev/v5/cmd/micro +GIT_IMPORT = go-micro.dev/v6/cmd/micro BUILD_DATE = $(shell date +%s) LDFLAGS = -X $(GIT_IMPORT).BuildDate=$(BUILD_DATE) -X $(GIT_IMPORT).GitCommit=$(GIT_COMMIT) -X $(GIT_IMPORT).GitTag=$(GIT_TAG) @@ -33,7 +33,7 @@ help: @echo " make clean - Clean build artifacts" $(NAME): - CGO_ENABLED=0 go build -ldflags "-s -w ${LDFLAGS}" -o $(NAME) cmd/micro/main.go + CGO_ENABLED=0 go build -trimpath -ldflags "-s -w ${LDFLAGS}" -o $(NAME) cmd/micro/main.go # Run tests test: @@ -128,7 +128,7 @@ install-tools: go install github.com/golangci/golangci-lint/cmd/golangci-lint@latest go install golang.org/x/tools/cmd/goimports@latest go install github.com/kyoh86/richgo@latest - go install go-micro.dev/v5/cmd/protoc-gen-micro@latest + go install go-micro.dev/v6/cmd/protoc-gen-micro@latest @echo "Tools installed successfully" # Generate protobuf code diff --git a/agent/agent.go b/agent/agent.go index 5140fc9632..f08974a9fa 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -28,6 +28,7 @@ import ( "go-micro.dev/v6/ai" "go-micro.dev/v6/flow" "go-micro.dev/v6/gateway/a2a" + internalotel "go-micro.dev/v6/internal/otel" "go-micro.dev/v6/server" "go-micro.dev/v6/store" @@ -610,6 +611,7 @@ func (a *agentImpl) Chat(ctx context.Context, req *pb.ChatRequest, rsp *pb.ChatR // Run starts the agent as a service with a Chat RPC endpoint. func (a *agentImpl) Run() error { + defer internalotel.Shutdown() if a.model == nil { a.setup() } @@ -664,7 +666,11 @@ func (a *agentImpl) Run() error { return nil } +// Stop stops the agent and flushes any pending OTel spans. Stop is the +// synchronous exit point short-lived agent processes hit, so it also +// drains the exporter (agent.Run may not return before process exit). func (a *agentImpl) Stop() error { + defer internalotel.Shutdown() a.mu.Lock() if a.stopCh != nil { close(a.stopCh) diff --git a/agent/options.go b/agent/options.go index f7074e1a9c..b06a4d9d62 100644 --- a/agent/options.go +++ b/agent/options.go @@ -8,9 +8,11 @@ import ( "go-micro.dev/v6/broker" "go-micro.dev/v6/client" "go-micro.dev/v6/flow" + internalotel "go-micro.dev/v6/internal/otel" "go-micro.dev/v6/registry" "go-micro.dev/v6/store" "go-micro.dev/v6/wrapper/x402" + "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/trace" ) @@ -115,8 +117,9 @@ type Options struct { // on that address directly (no separate gateway), e.g. ":4000". A2AAddress string - // TraceProvider enables OpenTelemetry spans for agent runs, model calls, - // and tool calls. Nil disables instrumentation. + // TraceProvider emits OpenTelemetry spans for agent runs, model calls, + // and tool calls. Defaults to the global provider (noop when none is + // configured); pass an explicit provider to override. TraceProvider trace.TracerProvider // TraceInputs controls whether agent observability records include raw @@ -132,6 +135,7 @@ type Options struct { } func newOptions(opts ...Option) Options { + internalotel.Init() o := Options{ Registry: registry.DefaultRegistry, Client: client.DefaultClient, @@ -146,6 +150,10 @@ func newOptions(opts ...Option) Options { // On by default and lenient: identical repeated calls are a // no-progress loop, never useful. Set LoopLimit(0) to disable. LoopLimit: 3, + // Global provider (noop unless a tracer is configured, e.g. via + // OTEL_EXPORTER_OTLP_ENDPOINT) so agent code emits spans whenever + // the process has one. + TraceProvider: otel.GetTracerProvider(), } for _, opt := range opts { opt(&o) diff --git a/ai/anthropic/anthropic.go b/ai/anthropic/anthropic.go index b78440e347..19c6860448 100644 --- a/ai/anthropic/anthropic.go +++ b/ai/anthropic/anthropic.go @@ -94,8 +94,8 @@ func cacheableSystem(system string, tools []map[string]any, noCache bool) any { // cacheableTools returns the tools with a cache breakpoint on the last one, // but only when the system prompt cannot carry it: a request with an empty -// system prompt and a large, stable tool catalogue is still worth caching, -// and without this the whole catalogue would be re-sent and re-billed on +// system prompt and a large, stable tool catalog is still worth caching, +// and without this the whole catalog would be re-sent and re-billed on // every call. When a system prompt is present, cacheableSystem's single // breakpoint already covers the tools, and marking them again would spend a // second of the four breakpoints a request gets for nothing. @@ -117,7 +117,7 @@ func cacheableTools(tools []map[string]any, system string, noCache bool) []map[s } // cachePrefixSize estimates the byte size of the cacheable prefix. The tools -// are marshalled once as a slice — the real request marshals them again in +// are marshaled once as a slice — the real request marshals them again in // callAPI, so this stays an estimate, not a second serialization per tool. func cachePrefixSize(system string, tools []map[string]any) int { size := len(system) @@ -130,7 +130,7 @@ func cachePrefixSize(system string, tools []map[string]any) int { } // minCacheBytes is the smallest prefix worth asking the API to cache, in -// bytes (len of the UTF-8 text and marshalled tools, not characters): the +// bytes (len of the UTF-8 text and marshaled tools, not characters): the // API's minimum cacheable prefix is 1024 tokens, at roughly four bytes each. const minCacheBytes = 4096 diff --git a/ai/anthropic/cache_test.go b/ai/anthropic/cache_test.go index 5b9f9a88a4..4c6cc383e2 100644 --- a/ai/anthropic/cache_test.go +++ b/ai/anthropic/cache_test.go @@ -10,7 +10,7 @@ import ( // // An agent's request is mostly the same request every time: the tools and the // system prompt do not change between turns, or between the rounds of one tool -// loop. Uncached that is the whole catalogue re-sent and re-billed on every +// loop. Uncached that is the whole catalog re-sent and re-billed on every // call — for a caller with a hundred tools, tens of thousands of tokens a turn. func TestABigPrefixIsMarkedForCaching(t *testing.T) { tools := []map[string]any{} @@ -56,7 +56,7 @@ func TestTheToolsAreNotMarkedSeparately(t *testing.T) { } // Below the smallest cacheable prefix it stays a plain string. Asking the API -// to cache less than it will cache is an error, and the thing being optimised +// to cache less than it will cache is an error, and the thing being optimized // is not present anyway. func TestASmallRequestIsLeftAlone(t *testing.T) { for _, tc := range []struct { @@ -89,8 +89,8 @@ func TestToolsCountTowardsTheThreshold(t *testing.T) { } } -// With no system prompt to carry the breakpoint, a large tool catalogue gets -// it on the last tool instead — otherwise the whole catalogue is re-sent and +// With no system prompt to carry the breakpoint, a large tool catalog gets +// it on the last tool instead — otherwise the whole catalog is re-sent and // re-billed on every call of a system-prompt-less caller. The original slice // and its maps are left untouched. func TestAToolOnlyPrefixIsCachedOnTheLastTool(t *testing.T) { diff --git a/ai/groq/groq.go b/ai/groq/groq.go index 9ffc9bec23..82ff71e473 100644 --- a/ai/groq/groq.go +++ b/ai/groq/groq.go @@ -13,13 +13,7 @@ package groq import ( - "bytes" "context" - "encoding/json" - "fmt" - "io" - "net/http" - "strings" "go-micro.dev/v6/ai" "go-micro.dev/v6/ai/internal/openaiapi" @@ -34,194 +28,25 @@ func init() { } type Provider struct { - opts ai.Options + core *openaiapi.Client } func NewProvider(opts ...ai.Option) *Provider { - options := ai.NewOptions(opts...) - if options.Model == "" { - options.Model = "llama-3.3-70b-versatile" - } - if options.BaseURL == "" { - options.BaseURL = "https://api.groq.com/openai" - } - return &Provider{opts: options} + return &Provider{core: openaiapi.New(openaiapi.Config{ + Name: "groq", + DefaultBase: "https://api.groq.com/openai", + DefaultModel: "llama-3.3-70b-versatile", + }, opts...)} } -func (p *Provider) Init(opts ...ai.Option) error { - for _, o := range opts { - o(&p.opts) - } - return nil -} - -func (p *Provider) Options() ai.Options { return p.opts } -func (p *Provider) String() string { return "groq" } +func (p *Provider) Init(opts ...ai.Option) error { return p.core.Init(opts...) } +func (p *Provider) Options() ai.Options { return p.core.Options() } +func (p *Provider) String() string { return p.core.String() } func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (*ai.Response, error) { - var tools []map[string]any - for _, t := range req.Tools { - tools = append(tools, map[string]any{ - "type": "function", - "function": map[string]any{ - "name": t.Name, - "description": t.Description, - "parameters": map[string]any{ - "type": "object", - "properties": t.Properties, - }, - }, - }) - } - - messages := []map[string]any{ - {"role": "system", "content": req.SystemPrompt}, - {"role": "user", "content": req.Prompt}, - } - - apiReq := map[string]any{ - "model": p.opts.Model, - "messages": messages, - } - if len(tools) > 0 { - apiReq["tools"] = tools - } - - resp, rawMessage, err := p.callAPI(ctx, apiReq) - if err != nil { - return nil, err - } - if len(resp.ToolCalls) == 0 { - return resp, nil - } - - // Tool execution loop: execute tools, send results back, and keep the - // tools on offer so the model can take the next step. A follow-up without - // "tools" asks the model to continue with its hands tied — the call it - // wanted comes back written out as prose — and without a loop a second - // step is impossible whatever the model wants. Bounded so a model that - // never stops asking cannot run forever. - if p.opts.ToolHandler != nil { - // Copied rather than aliased: append on a slice that shares an array - // with messages would overwrite it on a later round. - followUpMessages := append([]map[string]any(nil), messages...) - pending := resp.ToolCalls - raw := rawMessage - for round := 0; len(pending) > 0 && round < maxToolRounds; round++ { - followUpMessages = append(followUpMessages, map[string]any{ - "role": "assistant", - "content": raw["content"], - "tool_calls": raw["tool_calls"], - }) - for _, tc := range pending { - content := p.opts.ToolHandler(ctx, tc).Content - followUpMessages = append(followUpMessages, map[string]any{ - "role": "tool", - "tool_call_id": tc.ID, - "content": content, - }) - } - - followUpReq := map[string]any{ - "model": p.opts.Model, - "messages": followUpMessages, - } - if len(tools) > 0 { - followUpReq["tools"] = tools - } - - followUpResp, followUpRaw, err := p.callAPI(ctx, followUpReq) - if err != nil { - break - } - if followUpResp.Reply != "" { - resp.Answer = followUpResp.Reply - } - pending, raw = followUpResp.ToolCalls, followUpRaw - resp.ToolCalls = append(resp.ToolCalls, followUpResp.ToolCalls...) - } - } - - return resp, nil + return p.core.Generate(ctx, req, opts...) } -// maxToolRounds bounds the tool-execution loop in a single Generate. Each -// round is a model call plus the tools it asks for, so this is the ceiling on -// one question's cost as well as its length; it is high enough that no honest -// piece of multi-step work reaches it. -const maxToolRounds = 12 - func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) { - return openaiapi.Stream(ctx, p.opts, req, "/v1/chat/completions") -} - -func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) { - reqBody, err := json.Marshal(req) - if err != nil { - return nil, nil, fmt.Errorf("failed to marshal request: %w", err) - } - - apiURL := strings.TrimRight(p.opts.BaseURL, "/") + "/v1/chat/completions" - httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(reqBody)) - if err != nil { - return nil, nil, fmt.Errorf("failed to create request: %w", err) - } - - httpReq.Header.Set("Content-Type", "application/json") - httpReq.Header.Set("Authorization", "Bearer "+p.opts.APIKey) - - httpResp, err := http.DefaultClient.Do(httpReq) - if err != nil { - return nil, nil, fmt.Errorf("API request failed: %w", err) - } - defer httpResp.Body.Close() - - respBody, _ := io.ReadAll(httpResp.Body) - if httpResp.StatusCode != http.StatusOK { - return nil, nil, ai.NewHTTPError(httpResp, respBody) - } - - var chatResp struct { - Choices []struct { - Message struct { - Content string `json:"content"` - ToolCalls []struct { - ID string `json:"id"` - Function struct { - Name string `json:"name"` - Arguments string `json:"arguments"` - } `json:"function"` - } `json:"tool_calls"` - } `json:"message"` - } `json:"choices"` - } - - if err := json.Unmarshal(respBody, &chatResp); err != nil { - return nil, nil, fmt.Errorf("failed to parse response: %w", err) - } - if len(chatResp.Choices) == 0 { - return nil, nil, fmt.Errorf("no response from API") - } - - choice := chatResp.Choices[0] - response := &ai.Response{Reply: choice.Message.Content} - - for _, tc := range choice.Message.ToolCalls { - var input map[string]any - if err := json.Unmarshal([]byte(tc.Function.Arguments), &input); err != nil { - input = map[string]any{} - } - response.ToolCalls = append(response.ToolCalls, ai.ToolCall{ - ID: tc.ID, - Name: tc.Function.Name, - Input: input, - }) - } - - rawMessage := map[string]any{ - "content": choice.Message.Content, - "tool_calls": choice.Message.ToolCalls, - } - - return response, rawMessage, nil + return p.core.Stream(ctx, req, opts...) } diff --git a/ai/internal/openaiapi/client.go b/ai/internal/openaiapi/client.go new file mode 100644 index 0000000000..6824edb6e0 --- /dev/null +++ b/ai/internal/openaiapi/client.go @@ -0,0 +1,386 @@ +// Package openaiapi implements the shared OpenAI-compatible chat client +// used by the ai provider shells (openai, groq, mistral, together, +// minimax). It owns transport, request building, SSE parsing, response +// parsing, and the tool-result loop exactly once. +package openaiapi + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + + "go-micro.dev/v6/ai" +) + +// Config describes one OpenAI-compatible chat provider. Every behavioral +// divergence between the provider shells collapses into a flag here. +type Config struct { + // Name is the provider name; backs Model.String() ("groq", "openai", ...). + Name string + // DefaultBase is the default BaseURL when ai.WithBaseURL is not set. + DefaultBase string + // DefaultModel is the default model when ai.WithModel is not set. + DefaultModel string + // Path is the chat endpoint relative to BaseURL. Defaults to "/v1/chat/completions". + Path string + // UseMessages makes Generate honor Request.Messages (multi-turn), + // max_tokens, and reasoning_effort. Off preserves the old single-turn + // system+prompt behavior of groq/mistral/together. + UseMessages bool + // ParseUsage maps provider token usage into Response.Usage. + ParseUsage bool +} + +// Client satisfies ai.Model and ai.ImageModel. +// Use New to construct. +type Client struct { + cfg Config + opts ai.Options +} + +// New is the only entry point for an OpenAI-compatible chat provider. +func New(cfg Config, opts ...ai.Option) *Client { + o := ai.NewOptions(opts...) + if o.Model == "" { + o.Model = cfg.DefaultModel + } + if o.BaseURL == "" { + o.BaseURL = cfg.DefaultBase + } + if cfg.Path == "" { + cfg.Path = "/v1/chat/completions" + } + return &Client{cfg: cfg, opts: o} +} + +func (c *Client) Init(opts ...ai.Option) error { + for _, o := range opts { + o(&c.opts) + } + return nil +} + +func (c *Client) Options() ai.Options { return c.opts } +func (c *Client) String() string { return c.cfg.Name } + +// Generate performs a chat completion, executing tools when a ToolHandler is set. +func (c *Client) Generate(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (*ai.Response, error) { + apiReq := c.chatRequest(req) + resp, rawMessage, err := c.callAPI(ctx, apiReq) + if err != nil { + return nil, err + } + if len(resp.ToolCalls) == 0 { + return resp, nil + } + + // Tool execution loop: execute tools, send results back, and keep the + // tools on offer so the model can take the next step. A follow-up without + // "tools" asks the model to continue with its hands tied — the call it + // wanted comes back written out as prose — and without a loop a second + // step is impossible whatever the model wants. Bounded so a model that + // never stops asking cannot run forever. + if c.opts.ToolHandler != nil { + messages := apiReq["messages"].([]map[string]any) + // Copied rather than aliased: append on a slice that shares an array + // with messages would overwrite it on a later round. + followUpMessages := append([]map[string]any(nil), messages...) + pending := resp.ToolCalls + raw := rawMessage + for round := 0; len(pending) > 0 && round < MaxToolRounds; round++ { + followUpMessages = append(followUpMessages, map[string]any{ + "role": "assistant", + "content": raw["content"], + "tool_calls": raw["tool_calls"], + }) + for i, tc := range pending { + tr := c.opts.ToolHandler(ctx, tc) + pending[i].Result = tr.Content + followUpMessages = append(followUpMessages, map[string]any{ + "role": "tool", + "tool_call_id": tc.ID, + "content": tr.Content, + }) + } + + followUpReq := c.followUpRequest(followUpMessages) + if len(req.Tools) > 0 { + followUpReq["tools"] = buildTools(req.Tools) + } + + followUpResp, followUpRaw, err := c.callAPI(ctx, followUpReq) + if err != nil { + break + } + if followUpResp.Reply != "" { + resp.Answer = followUpResp.Reply + } + pending, raw = followUpResp.ToolCalls, followUpRaw + resp.ToolCalls = append(resp.ToolCalls, followUpResp.ToolCalls...) + } + } + return resp, nil +} + +// MaxToolRounds bounds the tool-execution loop in a single Generate. Each +// round is a model call plus the tools it asks for, so this is the ceiling on +// one question's cost as well as its length; it is high enough that no honest +// piece of multi-step work reaches it. +const MaxToolRounds = 12 + +// Stream opens an SSE stream over the chat endpoint. +func (c *Client) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) { + messages := []map[string]any{{"role": "system", "content": req.SystemPrompt}} + for _, m := range req.Messages { + messages = append(messages, map[string]any{"role": m.Role, "content": m.Content}) + } + if req.Prompt != "" { + messages = append(messages, map[string]any{"role": "user", "content": req.Prompt}) + } + apiReq := map[string]any{ + "model": c.opts.Model, + "messages": messages, + "stream": true, + "stream_options": map[string]any{"include_usage": true}, + } + if c.opts.MaxTokens > 0 { + apiReq["max_tokens"] = c.opts.MaxTokens + } + if c.opts.Effort != "" { + apiReq["reasoning_effort"] = c.opts.Effort + } + reqBody, err := json.Marshal(apiReq) + if err != nil { + return nil, fmt.Errorf("failed to marshal stream request: %w", err) + } + httpReq, err := c.newRequest(ctx, http.MethodPost, c.cfg.Path, reqBody) + if err != nil { + return nil, err + } + httpReq.Header.Set("Accept", "text/event-stream") + + httpResp, err := c.roundTripper().RoundTrip(httpReq) + if err != nil { + return nil, fmt.Errorf("stream API request failed: %w", err) + } + if httpResp.StatusCode != http.StatusOK { + defer httpResp.Body.Close() + respBody, _ := io.ReadAll(httpResp.Body) + return nil, ai.NewHTTPError(httpResp, respBody) + } + return &StreamReader{body: httpResp.Body, scanner: bufio.NewScanner(httpResp.Body)}, nil +} + +// GenerateImage generates an image via /v1/images/generations. +func (c *Client) GenerateImage(ctx context.Context, req *ai.ImageRequest, opts ...ai.GenerateOption) (*ai.ImageResponse, error) { + model := req.Model + if model == "" { + model = "gpt-image-1" + } + n := req.N + if n <= 0 { + n = 1 + } + apiReq := map[string]any{"model": model, "prompt": req.Prompt, "n": n} + if req.Size != "" { + apiReq["size"] = req.Size + } + reqBody, err := json.Marshal(apiReq) + if err != nil { + return nil, fmt.Errorf("failed to marshal request: %w", err) + } + httpReq, err := c.newRequest(ctx, http.MethodPost, "/v1/images/generations", reqBody) + if err != nil { + return nil, err + } + + httpResp, err := c.roundTripper().RoundTrip(httpReq) + if err != nil { + return nil, fmt.Errorf("API request failed: %w", err) + } + defer httpResp.Body.Close() + + respBody, _ := io.ReadAll(httpResp.Body) + if httpResp.StatusCode != http.StatusOK { + return nil, ai.NewHTTPError(httpResp, respBody) + } + + var imgResp struct { + Data []struct { + URL string `json:"url"` + B64JSON string `json:"b64_json"` + } `json:"data"` + } + if err := json.Unmarshal(respBody, &imgResp); err != nil { + return nil, fmt.Errorf("failed to parse response: %w", err) + } + + response := &ai.ImageResponse{} + for _, d := range imgResp.Data { + response.Images = append(response.Images, ai.Image{URL: d.URL, Base64: d.B64JSON}) + } + return response, nil +} + +// chatRequest builds the non-streaming chat body, honoring the UseMessages flag. +func (c *Client) chatRequest(req *ai.Request) map[string]any { + var messages []map[string]any + if c.cfg.UseMessages { + messages = append(messages, map[string]any{"role": "system", "content": req.SystemPrompt}) + for _, m := range req.Messages { + messages = append(messages, map[string]any{"role": m.Role, "content": m.Content}) + } + if req.Prompt != "" { + messages = append(messages, map[string]any{"role": "user", "content": req.Prompt}) + } + } else { + messages = []map[string]any{ + {"role": "system", "content": req.SystemPrompt}, + {"role": "user", "content": req.Prompt}, + } + } + apiReq := map[string]any{"model": c.opts.Model, "messages": messages} + if c.cfg.UseMessages { + if c.opts.MaxTokens > 0 { + apiReq["max_tokens"] = c.opts.MaxTokens + } + if c.opts.Effort != "" { + apiReq["reasoning_effort"] = c.opts.Effort + } + } + if len(req.Tools) > 0 { + apiReq["tools"] = buildTools(req.Tools) + } + return apiReq +} + +func (c *Client) followUpRequest(messages []map[string]any) map[string]any { + apiReq := map[string]any{"model": c.opts.Model, "messages": messages} + if c.cfg.UseMessages { + if c.opts.MaxTokens > 0 { + apiReq["max_tokens"] = c.opts.MaxTokens + } + if c.opts.Effort != "" { + apiReq["reasoning_effort"] = c.opts.Effort + } + } + return apiReq +} + +func (c *Client) callAPI(ctx context.Context, apiReq map[string]any) (*ai.Response, map[string]any, error) { + reqBody, err := json.Marshal(apiReq) + if err != nil { + return nil, nil, fmt.Errorf("failed to marshal request: %w", err) + } + httpReq, err := c.newRequest(ctx, http.MethodPost, c.cfg.Path, reqBody) + if err != nil { + return nil, nil, err + } + + httpResp, err := c.roundTripper().RoundTrip(httpReq) + if err != nil { + return nil, nil, fmt.Errorf("API request failed: %w", err) + } + defer httpResp.Body.Close() + + respBody, _ := io.ReadAll(httpResp.Body) + if httpResp.StatusCode != http.StatusOK { + return nil, nil, ai.NewHTTPError(httpResp, respBody) + } + + var chatResp struct { + Usage struct { + PromptTokens int `json:"prompt_tokens"` + CompletionTokens int `json:"completion_tokens"` + TotalTokens int `json:"total_tokens"` + } `json:"usage"` + Choices []struct { + Message struct { + Content string `json:"content"` + ToolCalls []struct { + ID string `json:"id"` + Function struct { + Name string `json:"name"` + Arguments string `json:"arguments"` + } `json:"function"` + } `json:"tool_calls"` + } `json:"message"` + } `json:"choices"` + } + if err := json.Unmarshal(respBody, &chatResp); err != nil { + return nil, nil, fmt.Errorf("failed to parse response: %w", err) + } + if len(chatResp.Choices) == 0 { + return nil, nil, fmt.Errorf("no response from API") + } + + choice := chatResp.Choices[0] + response := &ai.Response{Reply: choice.Message.Content} + if c.cfg.ParseUsage { + response.Usage = ai.Usage{ + InputTokens: chatResp.Usage.PromptTokens, + OutputTokens: chatResp.Usage.CompletionTokens, + TotalTokens: chatResp.Usage.TotalTokens, + } + } + for _, tc := range choice.Message.ToolCalls { + var input map[string]any + if err := json.Unmarshal([]byte(tc.Function.Arguments), &input); err != nil { + input = map[string]any{} + } + response.ToolCalls = append(response.ToolCalls, ai.ToolCall{ + ID: tc.ID, + Name: tc.Function.Name, + Input: input, + }) + } + + rawMessage := map[string]any{ + "content": choice.Message.Content, + "tool_calls": choice.Message.ToolCalls, + } + return response, rawMessage, nil +} + +func (c *Client) newRequest(ctx context.Context, method, path string, body []byte) (*http.Request, error) { + apiURL := strings.TrimRight(c.opts.BaseURL, "/") + path + httpReq, err := http.NewRequestWithContext(ctx, method, apiURL, bytes.NewReader(body)) + if err != nil { + return nil, fmt.Errorf("failed to create request: %w", err) + } + httpReq.Header.Set("Content-Type", "application/json") + httpReq.Header.Set("Authorization", "Bearer "+c.opts.APIKey) + return httpReq, nil +} + +// ponytail: use DefaultTransport directly; identical to DefaultClient's +// transport minus redirect-following, which chat APIs never need. +func (c *Client) roundTripper() http.RoundTripper { + if c.opts.Transport != nil { + return c.opts.Transport + } + return http.DefaultTransport +} + +func buildTools(tools []ai.Tool) []map[string]any { + out := make([]map[string]any, 0, len(tools)) + for _, t := range tools { + out = append(out, map[string]any{ + "type": "function", + "function": map[string]any{ + "name": t.Name, + "description": t.Description, + "parameters": map[string]any{ + "type": "object", + "properties": t.Properties, + }, + }, + }) + } + return out +} diff --git a/ai/internal/openaiapi/client_test.go b/ai/internal/openaiapi/client_test.go new file mode 100644 index 0000000000..3b91c88b37 --- /dev/null +++ b/ai/internal/openaiapi/client_test.go @@ -0,0 +1,519 @@ +package openaiapi + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "strings" + "testing" + "time" + + "go-micro.dev/v6/ai" +) + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +func jsonResponse(code int, body string) roundTripFunc { + return func(r *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: code, + Header: http.Header{"Content-Type": {"application/json"}}, + Body: io.NopCloser(strings.NewReader(body)), + }, nil + } +} + +type requestRecord struct { + method string + path string + header http.Header + body []byte +} + +type recorder struct { + calls []requestRecord + next http.RoundTripper +} + +func (r *recorder) RoundTrip(req *http.Request) (*http.Response, error) { + body, _ := io.ReadAll(req.Body) + req.Body = io.NopCloser(strings.NewReader(string(body))) + r.calls = append(r.calls, requestRecord{ + method: req.Method, path: req.URL.Path, header: req.Header.Clone(), body: body, + }) + if r.next == nil { + return &http.Response{StatusCode: http.StatusInternalServerError, Body: io.NopCloser(strings.NewReader("no next transport"))}, nil + } + return r.next.RoundTrip(req) +} + +func (r *recorder) decode(i int) map[string]any { + var m map[string]any + if err := json.Unmarshal(r.calls[i].body, &m); err != nil { + panic(err) + } + return m +} + +type seqRoundTripper struct { + responses []roundTripFunc + idx int +} + +func (s *seqRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) { + if s.idx >= len(s.responses) { + return jsonResponse(500, "unexpected extra request")(r) + } + rr := s.responses[s.idx] + s.idx++ + return rr(r) +} + +func demoClient(t *testing.T, cfg Config, rt http.RoundTripper, opts ...ai.Option) *Client { + t.Helper() + return New(cfg, append(opts, ai.WithTransport(rt))...) +} + +func TestNew_FoldsDefaults(t *testing.T) { + c := New(Config{Name: "demo", DefaultBase: "https://x.example", DefaultModel: "demo-model"}) + if c.String() != "demo" { + t.Fatalf("String = %q", c.String()) + } + o := c.Options() + if o.Model != "demo-model" { + t.Fatalf("model = %q, want default", o.Model) + } + if o.BaseURL != "https://x.example" { + t.Fatalf("baseURL = %q, want default", o.BaseURL) + } + if err := c.Init(ai.WithModel("other")); err != nil { + t.Fatalf("Init: %v", err) + } + if c.Options().Model != "other" { + t.Fatalf("model after Init = %q", c.Options().Model) + } +} + +func TestGenerate_SingleTurn(t *testing.T) { + rec := &recorder{next: jsonResponse(200, `{"choices":[{"message":{"content":"hi"}}]}`)} + c := demoClient(t, Config{Name: "demo", DefaultBase: "https://x.example", DefaultModel: "m"}, + rec, ai.WithAPIKey("k")) + + resp, err := c.Generate(context.Background(), &ai.Request{ + SystemPrompt: "sys", + Prompt: "hello", + Messages: []ai.Message{{Role: "user", Content: "previous"}}, + }) + if err != nil { + t.Fatalf("Generate: %v", err) + } + if resp.Reply != "hi" { + t.Fatalf("Reply = %q", resp.Reply) + } + if len(rec.calls) != 1 { + t.Fatalf("calls = %d, want 1", len(rec.calls)) + } + call := rec.calls[0] + if call.method != http.MethodPost || call.path != "/v1/chat/completions" { + t.Fatalf("call = %s %s", call.method, call.path) + } + if got := call.header.Get("Authorization"); got != "Bearer k" { + t.Fatalf("Authorization = %q", got) + } + body := rec.decode(0) + if body["model"] != "m" { + t.Fatalf("model = %v", body["model"]) + } + messages := body["messages"].([]any) + if len(messages) != 2 { + t.Fatalf("messages = %#v, want system+user only (single-turn)", messages) + } + if messages[0].(map[string]any)["role"] != "system" || messages[1].(map[string]any)["role"] != "user" { + t.Fatalf("messages roles wrong: %#v", messages) + } +} + +func TestGenerate_UseMessages(t *testing.T) { + for _, tc := range []struct { + name string + useMsg bool + }{ + {name: "off", useMsg: false}, + {name: "on", useMsg: true}, + } { + t.Run(tc.name, func(t *testing.T) { + rec := &recorder{next: jsonResponse(200, `{"choices":[{"message":{"content":"hi"}}]}`)} + c := demoClient(t, Config{Name: "demo", DefaultBase: "https://x.example", DefaultModel: "m", UseMessages: tc.useMsg}, + rec, ai.WithAPIKey("k"), ai.WithMaxTokens(8), ai.WithEffort("high")) + + if _, err := c.Generate(context.Background(), &ai.Request{ + SystemPrompt: "sys", + Prompt: "hello", + Messages: []ai.Message{{Role: "user", Content: "previous"}}, + }); err != nil { + t.Fatalf("Generate: %v", err) + } + body := rec.decode(0) + + if tc.useMsg { + messages := body["messages"].([]any) + if len(messages) != 3 { + t.Fatalf("messages = %#v, want system+history+prompt", messages) + } + if r := messages[1].(map[string]any)["role"]; r != "user" { + t.Fatalf("message[1] role = %v", r) + } + if got := messages[1].(map[string]any)["content"]; got != "previous" { + t.Fatalf("message[1] content = %v, want history preserved", got) + } + if body["max_tokens"] != float64(8) { + t.Fatalf("max_tokens = %v", body["max_tokens"]) + } + if body["reasoning_effort"] != "high" { + t.Fatalf("reasoning_effort = %v", body["reasoning_effort"]) + } + } else { + if body["max_tokens"] != nil || body["reasoning_effort"] != nil { + t.Fatalf("single-turn must not send max_tokens/effort: %#v", body) + } + messages := body["messages"].([]any) + if len(messages) != 2 { + t.Fatalf("messages = %#v, want system+user only", messages) + } + } + }) + } +} + +func TestGenerate_ToolLoop(t *testing.T) { + seq := &seqRoundTripper{responses: []roundTripFunc{ + jsonResponse(200, `{"choices":[{"message":{"content":"","tool_calls":[ + {"id":"call_1","function":{"name":"lookup","arguments":"{\"q\":\"a\"}"}}]}}]}`), + jsonResponse(200, `{"choices":[{"message":{"content":"final answer"}}]}`), + }} + rec := &recorder{next: seq} + c := demoClient(t, Config{Name: "demo", DefaultBase: "https://x.example", DefaultModel: "m"}, + rec, ai.WithAPIKey("k")) + + var called string + handler := func(ctx context.Context, tc ai.ToolCall) ai.ToolResult { + called = tc.Name + if tc.ID != "call_1" { + t.Fatalf("tool call ID = %q", tc.ID) + } + if v, ok := tc.Input["q"].(string); !ok || v != "a" { + t.Fatalf("tool input = %#v", tc.Input) + } + return ai.ToolResult{ID: tc.ID, Content: `{"ok":1}`} + } + if err := c.Init(func(o *ai.Options) { o.ToolHandler = handler }); err != nil { + t.Fatalf("Init: %v", err) + } + + resp, err := c.Generate(context.Background(), &ai.Request{ + SystemPrompt: "sys", + Prompt: "hello", + Tools: []ai.Tool{{Name: "lookup", Description: "looks up", Properties: map[string]any{"q": map[string]any{"type": "string"}}}}, + }) + if err != nil { + t.Fatalf("Generate: %v", err) + } + if called != "lookup" { + t.Fatalf("handler called with %q, want lookup", called) + } + if len(resp.ToolCalls) != 1 || resp.ToolCalls[0].Result != `{"ok":1}` { + t.Fatalf("ToolCalls = %#v, want Result populated", resp.ToolCalls) + } + if resp.Answer != "final answer" { + t.Fatalf("Answer = %q", resp.Answer) + } + if len(rec.calls) != 2 { + t.Fatalf("calls = %d, want first + follow-up", len(rec.calls)) + } + first := rec.decode(0) + tools := first["tools"].([]any) + tf := tools[0].(map[string]any) + if tf["type"] != "function" { + t.Fatalf("tools[0] = %#v", tf) + } + if _, ok := tf["function"].(map[string]any); !ok { + t.Fatalf("tools[0].function missing: %#v", tf) + } + follow := rec.decode(1) + msgs := follow["messages"].([]any) + last := msgs[len(msgs)-1].(map[string]any) + if last["role"] != "tool" || last["tool_call_id"] != "call_1" { + t.Fatalf("follow-up last message = %#v, want tool result", last) + } +} + +func TestGenerate_ParseUsage(t *testing.T) { + body := `{"choices":[{"message":{"content":"hi"}}],"usage": + {"prompt_tokens":7,"completion_tokens":5,"total_tokens":12}}` + for _, tc := range []struct { + name string + parse bool + wantInput int + }{ + {name: "parsed", parse: true, wantInput: 7}, + {name: "unparsed", parse: false}, + } { + t.Run(tc.name, func(t *testing.T) { + c := demoClient(t, Config{Name: "demo", DefaultBase: "https://x.example", DefaultModel: "m", ParseUsage: tc.parse}, + jsonResponse(200, body), ai.WithAPIKey("k")) + resp, err := c.Generate(context.Background(), &ai.Request{Prompt: "hi"}) + if err != nil { + t.Fatalf("Generate: %v", err) + } + if resp.Usage.InputTokens != tc.wantInput { + t.Fatalf("Usage = %+v, want InputTokens=%d", resp.Usage, tc.wantInput) + } + }) + } +} + +func TestGenerate_HTTPError(t *testing.T) { + c := demoClient(t, Config{Name: "demo", DefaultBase: "https://x.example", DefaultModel: "m"}, + roundTripFunc(func(r *http.Request) (*http.Response, error) { + return &http.Response{StatusCode: http.StatusTooManyRequests, Header: http.Header{"Retry-After": {"1"}}, + Body: io.NopCloser(strings.NewReader("rate limited"))}, nil + }), ai.WithAPIKey("k")) + + _, err := c.Generate(context.Background(), &ai.Request{Prompt: "hi"}) + if err == nil { + t.Fatal("Generate returned nil error") + } + var he *ai.HTTPError + if !errors.As(err, &he) { + t.Fatalf("error = %T, want *ai.HTTPError", err) + } + if he.StatusCode() != 429 { + t.Fatalf("status = %d", he.StatusCode()) + } + if d := he.RetryAfter(); d != time.Second { + t.Fatalf("RetryAfter = %v", d) + } + if ai.ClassifyError(err) != ai.ErrorKindRateLimited { + t.Fatalf("classify = %q, want rate_limited", ai.ClassifyError(err)) + } +} + +func TestStream_SSE(t *testing.T) { + sse := "data: {\"choices\":[{\"delta\":{\"content\":\"hel\"}}]}\n\n" + + "data: {\"choices\":[{\"delta\":{\"content\":\"lo\"}}]}\n\n" + + "data: {\"choices\":[],\"usage\":{\"prompt_tokens\":3,\"completion_tokens\":2,\"total_tokens\":5}}\n\n" + + "data: [DONE]\n\n" + rec := &recorder{next: roundTripFunc(func(r *http.Request) (*http.Response, error) { + return &http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": {"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(sse))}, nil + })} + c := demoClient(t, Config{Name: "demo", DefaultBase: "https://x.example", DefaultModel: "m"}, + rec, ai.WithAPIKey("k")) + + s, err := c.Stream(context.Background(), &ai.Request{ + SystemPrompt: "sys", + Messages: []ai.Message{{Role: "user", Content: "prev"}, {Role: "assistant", Content: "ans"}}, + Prompt: "p", + }) + if err != nil { + t.Fatalf("Stream: %v", err) + } + defer s.Close() + + assertReply(t, s, "hel") + assertReply(t, s, "lo") + usage, err := s.Recv() + if err != nil { + t.Fatalf("usage chunk: %v", err) + } + if usage.Usage != (ai.Usage{InputTokens: 3, OutputTokens: 2, TotalTokens: 5}) { + t.Fatalf("usage = %+v", usage.Usage) + } + if _, err := s.Recv(); !errors.Is(err, io.EOF) { + t.Fatalf("final = %v, want EOF", err) + } + + body := rec.decode(0) + if body["stream"] != true { + t.Fatalf("stream = %v", body["stream"]) + } + so, ok := body["stream_options"].(map[string]any) + if !ok || so["include_usage"] != true { + t.Fatalf("stream_options = %v", body["stream_options"]) + } + if got := rec.calls[0].header.Get("Accept"); got != "text/event-stream" { + t.Fatalf("Accept = %q", got) + } + if msgs := body["messages"].([]any); len(msgs) != 4 { + t.Fatalf("messages = %#v, want system+history+prompt", msgs) + } +} + +func TestStream_HTTPError(t *testing.T) { + c := demoClient(t, Config{Name: "demo", DefaultBase: "https://x.example", DefaultModel: "m"}, + roundTripFunc(func(r *http.Request) (*http.Response, error) { + return &http.Response{StatusCode: http.StatusTooManyRequests, Header: http.Header{"Retry-After": {"2"}}, + Body: io.NopCloser(strings.NewReader("quota"))}, nil + }), ai.WithAPIKey("k")) + + _, err := c.Stream(context.Background(), &ai.Request{Prompt: "hi"}) + if err == nil { + t.Fatal("Stream returned nil error") + } + var he *ai.HTTPError + if !errors.As(err, &he) || he.StatusCode() != 429 { + t.Fatalf("error = %T %v, want *ai.HTTPError 429", err, err) + } + if ai.ClassifyError(err) != ai.ErrorKindRateLimited { + t.Fatalf("classify = %q", ai.ClassifyError(err)) + } +} + +func TestStream_MalformedChunk(t *testing.T) { + c := demoClient(t, Config{Name: "demo", DefaultBase: "https://x.example", DefaultModel: "m"}, + roundTripFunc(func(r *http.Request) (*http.Response, error) { + return &http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": {"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader("data: {bad json}\n\n"))}, nil + }), ai.WithAPIKey("k")) + + s, err := c.Stream(context.Background(), &ai.Request{Prompt: "hi"}) + if err != nil { + t.Fatalf("Stream: %v", err) + } + defer s.Close() + if _, err := s.Recv(); err == nil { + t.Fatal("Recv returned nil error for malformed chunk") + } +} + +// TestProviderLiteralsConform locks the five real provider Config literals so +// a drift in any shell's defaults or chat path fails here. +func TestProviderLiteralsConform(t *testing.T) { + cases := []struct { + name string + cfg Config + model string + base string + multi bool + usage bool + }{ + {name: "openai", model: "gpt-4o", base: "https://api.openai.com", multi: true, usage: true, + cfg: Config{Name: "openai", DefaultBase: "https://api.openai.com", DefaultModel: "gpt-4o", UseMessages: true, ParseUsage: true}}, + {name: "groq", model: "llama-3.3-70b-versatile", base: "https://api.groq.com/openai", + cfg: Config{Name: "groq", DefaultBase: "https://api.groq.com/openai", DefaultModel: "llama-3.3-70b-versatile"}}, + {name: "mistral", model: "mistral-large-latest", base: "https://api.mistral.ai", + cfg: Config{Name: "mistral", DefaultBase: "https://api.mistral.ai", DefaultModel: "mistral-large-latest"}}, + {name: "together", model: "meta-llama/Llama-3.3-70B-Instruct-Turbo", base: "https://api.together.xyz", + cfg: Config{Name: "together", DefaultBase: "https://api.together.xyz", DefaultModel: "meta-llama/Llama-3.3-70B-Instruct-Turbo"}}, + {name: "minimax", model: "MiniMax-M3", base: "https://api.minimax.io", multi: true, + cfg: Config{Name: "minimax", DefaultBase: "https://api.minimax.io", DefaultModel: "MiniMax-M3", UseMessages: true}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + rec := &recorder{next: jsonResponse(http.StatusOK, `{"choices":[{"message":{"content":"hi"}}]}`)} + c := New(tc.cfg, ai.WithAPIKey("k"), ai.WithTransport(rec)) + if c.String() != tc.name { + t.Fatalf("String = %q", c.String()) + } + o := c.Options() + if o.Model != tc.model || o.BaseURL != tc.base { + t.Fatalf("defaults = %q/%q, want %q/%q", o.Model, o.BaseURL, tc.model, tc.base) + } + resp, err := c.Generate(context.Background(), &ai.Request{SystemPrompt: "sys", Prompt: "hello", Messages: []ai.Message{{Role: "user", Content: "prev"}}}) + if err != nil { + t.Fatalf("Generate: %v", err) + } + if resp.Reply != "hi" { + t.Fatalf("Reply = %q", resp.Reply) + } + body := rec.decode(0) + if body["model"] != tc.model { + t.Fatalf("request model = %v", body["model"]) + } + if got := rec.calls[0].header.Get("Authorization"); got != "Bearer k" { + t.Fatalf("Authorization = %q", got) + } + if msgs := len(body["messages"].([]any)); (tc.multi && msgs != 3) || (!tc.multi && msgs != 2) { + t.Fatalf("messages = %d, want %d", msgs, map[bool]int{true: 3, false: 2}[tc.multi]) + } + }) + } +} + +func TestGenerate_HTTPError_Retryable(t *testing.T) { + c := demoClient(t, Config{Name: "demo", DefaultBase: "https://x.example", DefaultModel: "m"}, + roundTripFunc(func(r *http.Request) (*http.Response, error) { + return &http.Response{StatusCode: http.StatusTooManyRequests, Header: http.Header{"Retry-After": {"1"}}, + Body: io.NopCloser(strings.NewReader("rate limited"))}, nil + }), ai.WithAPIKey("k")) + + _, err := ai.GenerateWithRetry(context.Background(), c, &ai.Request{Prompt: "hi"}, + ai.GeneratePolicy{MaxAttempts: 2, Backoff: time.Millisecond}) + if err == nil { + t.Fatal("GenerateWithRetry returned nil error") + } + var re *ai.RetryError + if !errors.As(err, &re) { + t.Fatalf("error = %T, want *ai.RetryError", err) + } + if re.ErrorKind() != ai.ErrorKindRateLimited { + t.Fatalf("kind = %q, want rate_limited", re.ErrorKind()) + } +} + +func TestGenerateImage(t *testing.T) { + c := demoClient(t, Config{Name: "demo", DefaultBase: "https://x.example", DefaultModel: "m"}, + jsonResponse(200, `{"data":[{"url":"u","b64_json":"b"}]}`), ai.WithAPIKey("k")) + resp, err := c.GenerateImage(context.Background(), &ai.ImageRequest{Prompt: "a cat"}) + if err != nil { + t.Fatalf("GenerateImage: %v", err) + } + if len(resp.Images) != 1 || resp.Images[0].URL != "u" || resp.Images[0].Base64 != "b" { + t.Fatalf("Images = %#v", resp.Images) + } +} + +func TestGenerateImage_HTTPError(t *testing.T) { + c := demoClient(t, Config{Name: "demo", DefaultBase: "https://x.example", DefaultModel: "m"}, + jsonResponse(402, `{"error":"payment required"}`), ai.WithAPIKey("k")) + _, err := c.GenerateImage(context.Background(), &ai.ImageRequest{Prompt: "a cat"}) + var he *ai.HTTPError + if !errors.As(err, &he) || he.StatusCode() != 402 { + t.Fatalf("error = %T %v, want *ai.HTTPError 402", err, err) + } +} + +func TestEmptyRequestNoAPIKeyFails(t *testing.T) { + c := New(Config{Name: "demo", DefaultBase: "https://x.example", DefaultModel: "m"}, + ai.WithTransport(roundTripFunc(func(r *http.Request) (*http.Response, error) { + if r.Header.Get("Authorization") == "Bearer " { + return &http.Response{StatusCode: http.StatusUnauthorized, + Body: io.NopCloser(strings.NewReader("missing key"))}, nil + } + return jsonResponse(http.StatusOK, `{"choices":[{"message":{"content":"hi"}}]}`)(r) + }))) + + _, err := c.Generate(context.Background(), &ai.Request{Prompt: "hi"}) + if err == nil { + t.Fatal("expected error without API key") + } + var he *ai.HTTPError + if !errors.As(err, &he) || he.StatusCode() != http.StatusUnauthorized { + t.Fatalf("error = %T %v, want unauthorized", err, err) + } +} + +func assertReply(t *testing.T, s ai.Stream, want string) { + t.Helper() + chunk, err := s.Recv() + if err != nil { + t.Fatalf("Recv: %v", err) + } + if chunk.Reply != want { + t.Fatalf("Reply = %q, want %q", chunk.Reply, want) + } +} diff --git a/ai/internal/openaiapi/stream.go b/ai/internal/openaiapi/stream.go index 62ec8fb76d..c3d4229baa 100644 --- a/ai/internal/openaiapi/stream.go +++ b/ai/internal/openaiapi/stream.go @@ -2,63 +2,14 @@ package openaiapi import ( "bufio" - "bytes" - "context" "encoding/json" "fmt" "io" - "net/http" "strings" "go-micro.dev/v6/ai" ) -// Stream opens an OpenAI-compatible chat completions SSE stream. -func Stream(ctx context.Context, opts ai.Options, req *ai.Request, basePath string) (ai.Stream, error) { - messages := []map[string]any{{"role": "system", "content": req.SystemPrompt}} - for _, m := range req.Messages { - messages = append(messages, map[string]any{"role": m.Role, "content": m.Content}) - } - if req.Prompt != "" { - messages = append(messages, map[string]any{"role": "user", "content": req.Prompt}) - } - apiReq := map[string]any{ - "model": opts.Model, - "messages": messages, - "stream": true, - "stream_options": map[string]any{"include_usage": true}, - } - if opts.MaxTokens > 0 { - apiReq["max_tokens"] = opts.MaxTokens - } - if opts.Effort != "" { - apiReq["reasoning_effort"] = opts.Effort - } - reqBody, err := json.Marshal(apiReq) - if err != nil { - return nil, fmt.Errorf("failed to marshal stream request: %w", err) - } - apiURL := strings.TrimRight(opts.BaseURL, "/") + basePath - httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(reqBody)) - if err != nil { - return nil, fmt.Errorf("failed to create stream request: %w", err) - } - httpReq.Header.Set("Content-Type", "application/json") - httpReq.Header.Set("Accept", "text/event-stream") - httpReq.Header.Set("Authorization", "Bearer "+opts.APIKey) - - httpResp, err := http.DefaultClient.Do(httpReq) - if err != nil { - return nil, fmt.Errorf("stream API request failed: %w", err) - } - if httpResp.StatusCode != http.StatusOK { - defer httpResp.Body.Close() - respBody, _ := io.ReadAll(httpResp.Body) - return nil, fmt.Errorf("stream API error (%s): %s", httpResp.Status, string(respBody)) - } - return &StreamReader{body: httpResp.Body, scanner: bufio.NewScanner(httpResp.Body)}, nil -} - // StreamReader reads OpenAI-compatible server-sent event chunks. type StreamReader struct { body io.ReadCloser diff --git a/ai/minimax/minimax.go b/ai/minimax/minimax.go index eb36764986..03e36a97ad 100644 --- a/ai/minimax/minimax.go +++ b/ai/minimax/minimax.go @@ -13,13 +13,7 @@ package minimax import ( - "bytes" "context" - "encoding/json" - "fmt" - "io" - "net/http" - "strings" "go-micro.dev/v6/ai" "go-micro.dev/v6/ai/internal/openaiapi" @@ -34,205 +28,26 @@ func init() { } type Provider struct { - opts ai.Options + core *openaiapi.Client } func NewProvider(opts ...ai.Option) *Provider { - options := ai.NewOptions(opts...) - if options.Model == "" { - options.Model = "MiniMax-M3" - } - if options.BaseURL == "" { - options.BaseURL = "https://api.minimax.io" - } - return &Provider{opts: options} + return &Provider{core: openaiapi.New(openaiapi.Config{ + Name: "minimax", + DefaultBase: "https://api.minimax.io", + DefaultModel: "MiniMax-M3", + UseMessages: true, + }, opts...)} } -func (p *Provider) Init(opts ...ai.Option) error { - for _, o := range opts { - o(&p.opts) - } - return nil -} - -func (p *Provider) Options() ai.Options { return p.opts } -func (p *Provider) String() string { return "minimax" } +func (p *Provider) Init(opts ...ai.Option) error { return p.core.Init(opts...) } +func (p *Provider) Options() ai.Options { return p.core.Options() } +func (p *Provider) String() string { return p.core.String() } func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (*ai.Response, error) { - var tools []map[string]any - for _, t := range req.Tools { - tools = append(tools, map[string]any{ - "type": "function", - "function": map[string]any{ - "name": t.Name, - "description": t.Description, - "parameters": map[string]any{ - "type": "object", - "properties": t.Properties, - }, - }, - }) - } - - messages := make([]map[string]any, 0, len(req.Messages)+2) - messages = append(messages, map[string]any{ - "role": "system", - "content": req.SystemPrompt, - }) - for _, message := range req.Messages { - messages = append(messages, map[string]any{ - "role": message.Role, - "content": message.Content, - }) - } - messages = append(messages, map[string]any{ - "role": "user", - "content": req.Prompt, - }) - - apiReq := map[string]any{ - "model": p.opts.Model, - "messages": messages, - } - if len(tools) > 0 { - apiReq["tools"] = tools - } - - resp, rawMessage, err := p.callAPI(ctx, apiReq) - if err != nil { - return nil, err - } - if len(resp.ToolCalls) == 0 { - return resp, nil - } - - // Tool execution loop: execute tools, send results back, and keep the - // tools on offer so the model can take the next step. A follow-up without - // "tools" asks the model to continue with its hands tied — the call it - // wanted comes back written out as prose — and without a loop a second - // step is impossible whatever the model wants. Bounded so a model that - // never stops asking cannot run forever. - if p.opts.ToolHandler != nil { - // Copied rather than aliased: append on a slice that shares an array - // with messages would overwrite it on a later round. - followUpMessages := append([]map[string]any(nil), messages...) - pending := resp.ToolCalls - raw := rawMessage - for round := 0; len(pending) > 0 && round < maxToolRounds; round++ { - followUpMessages = append(followUpMessages, map[string]any{ - "role": "assistant", - "content": raw["content"], - "tool_calls": raw["tool_calls"], - }) - for _, tc := range pending { - content := p.opts.ToolHandler(ctx, tc).Content - followUpMessages = append(followUpMessages, map[string]any{ - "role": "tool", - "tool_call_id": tc.ID, - "content": content, - }) - } - - followUpReq := map[string]any{ - "model": p.opts.Model, - "messages": followUpMessages, - } - if len(tools) > 0 { - followUpReq["tools"] = tools - } - - followUpResp, followUpRaw, err := p.callAPI(ctx, followUpReq) - if err != nil { - break - } - if followUpResp.Reply != "" { - resp.Answer = followUpResp.Reply - } - pending, raw = followUpResp.ToolCalls, followUpRaw - resp.ToolCalls = append(resp.ToolCalls, followUpResp.ToolCalls...) - } - } - - return resp, nil + return p.core.Generate(ctx, req, opts...) } -// maxToolRounds bounds the tool-execution loop in a single Generate. Each -// round is a model call plus the tools it asks for, so this is the ceiling on -// one question's cost as well as its length; it is high enough that no honest -// piece of multi-step work reaches it. -const maxToolRounds = 12 - func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) { - return openaiapi.Stream(ctx, p.opts, req, "/v1/chat/completions") -} - -func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) { - reqBody, err := json.Marshal(req) - if err != nil { - return nil, nil, fmt.Errorf("failed to marshal request: %w", err) - } - - apiURL := strings.TrimRight(p.opts.BaseURL, "/") + "/v1/chat/completions" - httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(reqBody)) - if err != nil { - return nil, nil, fmt.Errorf("failed to create request: %w", err) - } - - httpReq.Header.Set("Content-Type", "application/json") - httpReq.Header.Set("Authorization", "Bearer "+p.opts.APIKey) - - httpResp, err := http.DefaultClient.Do(httpReq) - if err != nil { - return nil, nil, fmt.Errorf("API request failed: %w", err) - } - defer httpResp.Body.Close() - - respBody, _ := io.ReadAll(httpResp.Body) - if httpResp.StatusCode != http.StatusOK { - return nil, nil, ai.NewHTTPError(httpResp, respBody) - } - - var chatResp struct { - Choices []struct { - Message struct { - Content string `json:"content"` - ToolCalls []struct { - ID string `json:"id"` - Function struct { - Name string `json:"name"` - Arguments string `json:"arguments"` - } `json:"function"` - } `json:"tool_calls"` - } `json:"message"` - } `json:"choices"` - } - - if err := json.Unmarshal(respBody, &chatResp); err != nil { - return nil, nil, fmt.Errorf("failed to parse response: %w", err) - } - if len(chatResp.Choices) == 0 { - return nil, nil, fmt.Errorf("no response from API") - } - - choice := chatResp.Choices[0] - response := &ai.Response{Reply: choice.Message.Content} - - for _, tc := range choice.Message.ToolCalls { - var input map[string]any - if err := json.Unmarshal([]byte(tc.Function.Arguments), &input); err != nil { - input = map[string]any{} - } - response.ToolCalls = append(response.ToolCalls, ai.ToolCall{ - ID: tc.ID, - Name: tc.Function.Name, - Input: input, - }) - } - - rawMessage := map[string]any{ - "content": choice.Message.Content, - "tool_calls": choice.Message.ToolCalls, - } - - return response, rawMessage, nil + return p.core.Stream(ctx, req, opts...) } diff --git a/ai/mistral/mistral.go b/ai/mistral/mistral.go index 5bcfcff20e..3fee5aaf63 100644 --- a/ai/mistral/mistral.go +++ b/ai/mistral/mistral.go @@ -13,13 +13,7 @@ package mistral import ( - "bytes" "context" - "encoding/json" - "fmt" - "io" - "net/http" - "strings" "go-micro.dev/v6/ai" "go-micro.dev/v6/ai/internal/openaiapi" @@ -34,194 +28,25 @@ func init() { } type Provider struct { - opts ai.Options + core *openaiapi.Client } func NewProvider(opts ...ai.Option) *Provider { - options := ai.NewOptions(opts...) - if options.Model == "" { - options.Model = "mistral-large-latest" - } - if options.BaseURL == "" { - options.BaseURL = "https://api.mistral.ai" - } - return &Provider{opts: options} + return &Provider{core: openaiapi.New(openaiapi.Config{ + Name: "mistral", + DefaultBase: "https://api.mistral.ai", + DefaultModel: "mistral-large-latest", + }, opts...)} } -func (p *Provider) Init(opts ...ai.Option) error { - for _, o := range opts { - o(&p.opts) - } - return nil -} - -func (p *Provider) Options() ai.Options { return p.opts } -func (p *Provider) String() string { return "mistral" } +func (p *Provider) Init(opts ...ai.Option) error { return p.core.Init(opts...) } +func (p *Provider) Options() ai.Options { return p.core.Options() } +func (p *Provider) String() string { return p.core.String() } func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (*ai.Response, error) { - var tools []map[string]any - for _, t := range req.Tools { - tools = append(tools, map[string]any{ - "type": "function", - "function": map[string]any{ - "name": t.Name, - "description": t.Description, - "parameters": map[string]any{ - "type": "object", - "properties": t.Properties, - }, - }, - }) - } - - messages := []map[string]any{ - {"role": "system", "content": req.SystemPrompt}, - {"role": "user", "content": req.Prompt}, - } - - apiReq := map[string]any{ - "model": p.opts.Model, - "messages": messages, - } - if len(tools) > 0 { - apiReq["tools"] = tools - } - - resp, rawMessage, err := p.callAPI(ctx, apiReq) - if err != nil { - return nil, err - } - if len(resp.ToolCalls) == 0 { - return resp, nil - } - - // Tool execution loop: execute tools, send results back, and keep the - // tools on offer so the model can take the next step. A follow-up without - // "tools" asks the model to continue with its hands tied — the call it - // wanted comes back written out as prose — and without a loop a second - // step is impossible whatever the model wants. Bounded so a model that - // never stops asking cannot run forever. - if p.opts.ToolHandler != nil { - // Copied rather than aliased: append on a slice that shares an array - // with messages would overwrite it on a later round. - followUpMessages := append([]map[string]any(nil), messages...) - pending := resp.ToolCalls - raw := rawMessage - for round := 0; len(pending) > 0 && round < maxToolRounds; round++ { - followUpMessages = append(followUpMessages, map[string]any{ - "role": "assistant", - "content": raw["content"], - "tool_calls": raw["tool_calls"], - }) - for _, tc := range pending { - content := p.opts.ToolHandler(ctx, tc).Content - followUpMessages = append(followUpMessages, map[string]any{ - "role": "tool", - "tool_call_id": tc.ID, - "content": content, - }) - } - - followUpReq := map[string]any{ - "model": p.opts.Model, - "messages": followUpMessages, - } - if len(tools) > 0 { - followUpReq["tools"] = tools - } - - followUpResp, followUpRaw, err := p.callAPI(ctx, followUpReq) - if err != nil { - break - } - if followUpResp.Reply != "" { - resp.Answer = followUpResp.Reply - } - pending, raw = followUpResp.ToolCalls, followUpRaw - resp.ToolCalls = append(resp.ToolCalls, followUpResp.ToolCalls...) - } - } - - return resp, nil + return p.core.Generate(ctx, req, opts...) } -// maxToolRounds bounds the tool-execution loop in a single Generate. Each -// round is a model call plus the tools it asks for, so this is the ceiling on -// one question's cost as well as its length; it is high enough that no honest -// piece of multi-step work reaches it. -const maxToolRounds = 12 - func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) { - return openaiapi.Stream(ctx, p.opts, req, "/v1/chat/completions") -} - -func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) { - reqBody, err := json.Marshal(req) - if err != nil { - return nil, nil, fmt.Errorf("failed to marshal request: %w", err) - } - - apiURL := strings.TrimRight(p.opts.BaseURL, "/") + "/v1/chat/completions" - httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(reqBody)) - if err != nil { - return nil, nil, fmt.Errorf("failed to create request: %w", err) - } - - httpReq.Header.Set("Content-Type", "application/json") - httpReq.Header.Set("Authorization", "Bearer "+p.opts.APIKey) - - httpResp, err := http.DefaultClient.Do(httpReq) - if err != nil { - return nil, nil, fmt.Errorf("API request failed: %w", err) - } - defer httpResp.Body.Close() - - respBody, _ := io.ReadAll(httpResp.Body) - if httpResp.StatusCode != http.StatusOK { - return nil, nil, ai.NewHTTPError(httpResp, respBody) - } - - var chatResp struct { - Choices []struct { - Message struct { - Content string `json:"content"` - ToolCalls []struct { - ID string `json:"id"` - Function struct { - Name string `json:"name"` - Arguments string `json:"arguments"` - } `json:"function"` - } `json:"tool_calls"` - } `json:"message"` - } `json:"choices"` - } - - if err := json.Unmarshal(respBody, &chatResp); err != nil { - return nil, nil, fmt.Errorf("failed to parse response: %w", err) - } - if len(chatResp.Choices) == 0 { - return nil, nil, fmt.Errorf("no response from API") - } - - choice := chatResp.Choices[0] - response := &ai.Response{Reply: choice.Message.Content} - - for _, tc := range choice.Message.ToolCalls { - var input map[string]any - if err := json.Unmarshal([]byte(tc.Function.Arguments), &input); err != nil { - input = map[string]any{} - } - response.ToolCalls = append(response.ToolCalls, ai.ToolCall{ - ID: tc.ID, - Name: tc.Function.Name, - Input: input, - }) - } - - rawMessage := map[string]any{ - "content": choice.Message.Content, - "tool_calls": choice.Message.ToolCalls, - } - - return response, rawMessage, nil + return p.core.Stream(ctx, req, opts...) } diff --git a/ai/openai/openai.go b/ai/openai/openai.go index 4504469b04..25374fabb9 100644 --- a/ai/openai/openai.go +++ b/ai/openai/openai.go @@ -2,16 +2,10 @@ package openai import ( - "bufio" - "bytes" "context" - "encoding/json" - "fmt" - "io" - "net/http" - "strings" "go-micro.dev/v6/ai" + "go-micro.dev/v6/ai/internal/openaiapi" ) func init() { @@ -25,426 +19,35 @@ func init() { ai.RegisterToolStream("openai") } -// Provider implements the ai.Model interface for OpenAI +// Provider implements the ai.Model and ai.ImageModel interfaces for OpenAI. type Provider struct { - opts ai.Options + core *openaiapi.Client } -// NewProvider creates a new OpenAI provider +// NewProvider creates a new OpenAI provider. It preserves OpenAI's full +// surface: multi-turn messages, max_tokens, reasoning_effort, and token usage. func NewProvider(opts ...ai.Option) *Provider { - options := ai.NewOptions(opts...) - - // Set defaults if not provided - if options.Model == "" { - options.Model = "gpt-4o" - } - if options.BaseURL == "" { - options.BaseURL = "https://api.openai.com" - } - - return &Provider{ - opts: options, - } -} - -// Init initializes the provider with options -func (p *Provider) Init(opts ...ai.Option) error { - for _, o := range opts { - o(&p.opts) - } - return nil -} - -// Options returns the provider options -func (p *Provider) Options() ai.Options { - return p.opts + return &Provider{core: openaiapi.New(openaiapi.Config{ + Name: "openai", + DefaultBase: "https://api.openai.com", + DefaultModel: "gpt-4o", + UseMessages: true, + ParseUsage: true, + }, opts...)} } -// String returns the provider name -func (p *Provider) String() string { - return "openai" -} +func (p *Provider) Init(opts ...ai.Option) error { return p.core.Init(opts...) } +func (p *Provider) Options() ai.Options { return p.core.Options() } +func (p *Provider) String() string { return p.core.String() } -// Generate generates a response from the model func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (*ai.Response, error) { - // Build tools for OpenAI format - var openaiTools []map[string]any - for _, t := range req.Tools { - openaiTools = append(openaiTools, map[string]any{ - "type": "function", - "function": map[string]any{ - "name": t.Name, - "description": t.Description, - "parameters": map[string]any{ - "type": "object", - "properties": t.Properties, - }, - }, - }) - } - - // Build messages - messages := []map[string]any{ - {"role": "system", "content": req.SystemPrompt}, - } - for _, m := range req.Messages { - messages = append(messages, map[string]any{"role": m.Role, "content": m.Content}) - } - if req.Prompt != "" { - messages = append(messages, map[string]any{"role": "user", "content": req.Prompt}) - } - - // Build initial request - apiReq := map[string]any{ - "model": p.opts.Model, - "messages": messages, - } - if p.opts.MaxTokens > 0 { - apiReq["max_tokens"] = p.opts.MaxTokens - } - if p.opts.Effort != "" { - apiReq["reasoning_effort"] = p.opts.Effort - } - - if len(openaiTools) > 0 { - apiReq["tools"] = openaiTools - } - - // Make API call - resp, rawMessage, err := p.callAPI(ctx, apiReq) - if err != nil { - return nil, err - } - - // If no tool calls, return response - if len(resp.ToolCalls) == 0 { - return resp, nil - } - - // Tool execution loop: execute tools, send results back, and keep the - // tools on offer so the model can take the next step. A follow-up without - // "tools" asks the model to continue with its hands tied — the call it - // wanted comes back written out as prose — and without a loop a second - // step is impossible whatever the model wants. Bounded so a model that - // never stops asking cannot run forever. - if p.opts.ToolHandler != nil { - // Copied rather than aliased: append on a slice that shares an array - // with messages would overwrite it on a later round. - followUpMessages := append([]map[string]any(nil), messages...) - pending := resp.ToolCalls - raw := rawMessage - for round := 0; len(pending) > 0 && round < maxToolRounds; round++ { - followUpMessages = append(followUpMessages, map[string]any{ - "role": "assistant", - "content": raw["content"], - "tool_calls": raw["tool_calls"], - }) - for _, tc := range pending { - content := p.opts.ToolHandler(ctx, tc).Content - followUpMessages = append(followUpMessages, map[string]any{ - "role": "tool", - "tool_call_id": tc.ID, - "content": content, - }) - } - - followUpReq := map[string]any{ - "model": p.opts.Model, - "messages": followUpMessages, - } - if p.opts.MaxTokens > 0 { - followUpReq["max_tokens"] = p.opts.MaxTokens - } - if p.opts.Effort != "" { - followUpReq["reasoning_effort"] = p.opts.Effort - } - if len(openaiTools) > 0 { - followUpReq["tools"] = openaiTools - } - - followUpResp, followUpRaw, err := p.callAPI(ctx, followUpReq) - if err != nil { - break - } - if followUpResp.Reply != "" { - resp.Answer = followUpResp.Reply - } - pending, raw = followUpResp.ToolCalls, followUpRaw - resp.ToolCalls = append(resp.ToolCalls, followUpResp.ToolCalls...) - } - } - - return resp, nil + return p.core.Generate(ctx, req, opts...) } -// maxToolRounds bounds the tool-execution loop in a single Generate. Each -// round is a model call plus the tools it asks for, so this is the ceiling on -// one question's cost as well as its length; it is high enough that no honest -// piece of multi-step work reaches it. -const maxToolRounds = 12 - -// Stream generates a streaming response from the OpenAI chat completions API. func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) { - messages := []map[string]any{ - {"role": "system", "content": req.SystemPrompt}, - } - for _, m := range req.Messages { - messages = append(messages, map[string]any{"role": m.Role, "content": m.Content}) - } - if req.Prompt != "" { - messages = append(messages, map[string]any{"role": "user", "content": req.Prompt}) - } - apiReq := map[string]any{ - "model": p.opts.Model, - "messages": messages, - "stream": true, - "stream_options": map[string]any{"include_usage": true}, - } - if p.opts.MaxTokens > 0 { - apiReq["max_tokens"] = p.opts.MaxTokens - } - if p.opts.Effort != "" { - apiReq["reasoning_effort"] = p.opts.Effort - } - reqBody, err := json.Marshal(apiReq) - if err != nil { - return nil, fmt.Errorf("failed to marshal stream request: %w", err) - } - apiURL := strings.TrimRight(p.opts.BaseURL, "/") + "/v1/chat/completions" - httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(reqBody)) - if err != nil { - return nil, fmt.Errorf("failed to create stream request: %w", err) - } - httpReq.Header.Set("Content-Type", "application/json") - httpReq.Header.Set("Accept", "text/event-stream") - httpReq.Header.Set("Authorization", "Bearer "+p.opts.APIKey) - - httpResp, err := http.DefaultClient.Do(httpReq) - if err != nil { - return nil, fmt.Errorf("stream API request failed: %w", err) - } - if httpResp.StatusCode != http.StatusOK { - defer httpResp.Body.Close() - respBody, _ := io.ReadAll(httpResp.Body) - return nil, fmt.Errorf("stream API error (%s): %s", httpResp.Status, string(respBody)) - } - return &openAIStream{body: httpResp.Body, scanner: bufio.NewScanner(httpResp.Body)}, nil -} - -type openAIStream struct { - body io.ReadCloser - scanner *bufio.Scanner - closed bool -} - -func (s *openAIStream) Recv() (*ai.Response, error) { - for s.scanner.Scan() { - line := strings.TrimSpace(s.scanner.Text()) - if line == "" || strings.HasPrefix(line, ":") { - continue - } - if !strings.HasPrefix(line, "data:") { - continue - } - data := strings.TrimSpace(strings.TrimPrefix(line, "data:")) - if data == "[DONE]" { - return nil, io.EOF - } - var chunk struct { - Choices []struct { - Delta struct { - Content string `json:"content"` - } `json:"delta"` - } `json:"choices"` - Usage *struct { - PromptTokens int `json:"prompt_tokens"` - CompletionTokens int `json:"completion_tokens"` - TotalTokens int `json:"total_tokens"` - } `json:"usage"` - } - if err := json.Unmarshal([]byte(data), &chunk); err != nil { - return nil, fmt.Errorf("failed to parse stream chunk: %w", err) - } - if len(chunk.Choices) > 0 && chunk.Choices[0].Delta.Content != "" { - return &ai.Response{Reply: chunk.Choices[0].Delta.Content}, nil - } - // Final chunk (after include_usage) carries token usage and no content. - if chunk.Usage != nil { - return &ai.Response{Usage: ai.Usage{ - InputTokens: chunk.Usage.PromptTokens, - OutputTokens: chunk.Usage.CompletionTokens, - TotalTokens: chunk.Usage.TotalTokens, - }}, nil - } - continue - } - if err := s.scanner.Err(); err != nil { - return nil, err - } - return nil, io.EOF -} - -func (s *openAIStream) Close() error { - if s.closed { - return nil - } - s.closed = true - return s.body.Close() + return p.core.Stream(ctx, req, opts...) } -// callAPI makes an HTTP request to the OpenAI API -func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) { - // Marshal request - reqBody, err := json.Marshal(req) - if err != nil { - return nil, nil, fmt.Errorf("failed to marshal request: %w", err) - } - - // Build HTTP request - apiURL := strings.TrimRight(p.opts.BaseURL, "/") + "/v1/chat/completions" - httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(reqBody)) - if err != nil { - return nil, nil, fmt.Errorf("failed to create request: %w", err) - } - - // Set headers - httpReq.Header.Set("Content-Type", "application/json") - httpReq.Header.Set("Authorization", "Bearer "+p.opts.APIKey) - - // Make request - httpResp, err := http.DefaultClient.Do(httpReq) - if err != nil { - return nil, nil, fmt.Errorf("API request failed: %w", err) - } - defer httpResp.Body.Close() - - // Read response - respBody, _ := io.ReadAll(httpResp.Body) - if httpResp.StatusCode != http.StatusOK { - return nil, nil, ai.NewHTTPError(httpResp, respBody) - } - - // Parse response - var chatResp struct { - Usage struct { - PromptTokens int `json:"prompt_tokens"` - CompletionTokens int `json:"completion_tokens"` - TotalTokens int `json:"total_tokens"` - } `json:"usage"` - Choices []struct { - Message struct { - Content string `json:"content"` - ToolCalls []struct { - ID string `json:"id"` - Function struct { - Name string `json:"name"` - Arguments string `json:"arguments"` - } `json:"function"` - } `json:"tool_calls"` - } `json:"message"` - } `json:"choices"` - } - - if err := json.Unmarshal(respBody, &chatResp); err != nil { - return nil, nil, fmt.Errorf("failed to parse response: %w", err) - } - - if len(chatResp.Choices) == 0 { - return nil, nil, fmt.Errorf("no response from API") - } - - choice := chatResp.Choices[0] - response := &ai.Response{ - Reply: choice.Message.Content, - Usage: ai.Usage{InputTokens: chatResp.Usage.PromptTokens, OutputTokens: chatResp.Usage.CompletionTokens, TotalTokens: chatResp.Usage.TotalTokens}, - } - - // Extract tool calls - for _, tc := range choice.Message.ToolCalls { - var input map[string]any - if err := json.Unmarshal([]byte(tc.Function.Arguments), &input); err != nil { - input = map[string]any{} - } - response.ToolCalls = append(response.ToolCalls, ai.ToolCall{ - ID: tc.ID, - Name: tc.Function.Name, - Input: input, - }) - } - - // Return raw message for potential follow-up - rawMessage := map[string]any{ - "content": choice.Message.Content, - "tool_calls": choice.Message.ToolCalls, - } - - return response, rawMessage, nil -} - -const defaultImageModel = "gpt-image-1" - func (p *Provider) GenerateImage(ctx context.Context, req *ai.ImageRequest, opts ...ai.GenerateOption) (*ai.ImageResponse, error) { - model := req.Model - if model == "" { - model = defaultImageModel - } - n := req.N - if n <= 0 { - n = 1 - } - - apiReq := map[string]any{ - "model": model, - "prompt": req.Prompt, - "n": n, - } - if req.Size != "" { - apiReq["size"] = req.Size - } - - reqBody, err := json.Marshal(apiReq) - if err != nil { - return nil, fmt.Errorf("failed to marshal request: %w", err) - } - - apiURL := strings.TrimRight(p.opts.BaseURL, "/") + "/v1/images/generations" - httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(reqBody)) - if err != nil { - return nil, fmt.Errorf("failed to create request: %w", err) - } - - httpReq.Header.Set("Content-Type", "application/json") - httpReq.Header.Set("Authorization", "Bearer "+p.opts.APIKey) - - httpResp, err := http.DefaultClient.Do(httpReq) - if err != nil { - return nil, fmt.Errorf("API request failed: %w", err) - } - defer httpResp.Body.Close() - - respBody, _ := io.ReadAll(httpResp.Body) - if httpResp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("API error (%s): %s", httpResp.Status, string(respBody)) - } - - var imgResp struct { - Data []struct { - URL string `json:"url"` - B64JSON string `json:"b64_json"` - } `json:"data"` - } - - if err := json.Unmarshal(respBody, &imgResp); err != nil { - return nil, fmt.Errorf("failed to parse response: %w", err) - } - - response := &ai.ImageResponse{} - for _, d := range imgResp.Data { - response.Images = append(response.Images, ai.Image{ - URL: d.URL, - Base64: d.B64JSON, - }) - } - - return response, nil + return p.core.GenerateImage(ctx, req, opts...) } diff --git a/ai/openai/tool_loop_test.go b/ai/openai/tool_loop_test.go index 903049b610..af8501cf6b 100644 --- a/ai/openai/tool_loop_test.go +++ b/ai/openai/tool_loop_test.go @@ -9,6 +9,7 @@ import ( "testing" "go-micro.dev/v6/ai" + "go-micro.dev/v6/ai/internal/openaiapi" ) // A multi-step task needs the provider to (a) loop while the model keeps @@ -38,8 +39,7 @@ func TestGenerateToolLoopKeepsOfferingTools(t *testing.T) { defer srv.Close() toolRuns := 0 - p := &Provider{} - if err := p.Init( + p := NewProvider( ai.WithAPIKey("test"), ai.WithBaseURL(srv.URL), ai.WithModel("test-model"), @@ -47,9 +47,7 @@ func TestGenerateToolLoopKeepsOfferingTools(t *testing.T) { toolRuns++ return ai.ToolResult{ID: tc.ID, Content: "done"} }), - ); err != nil { - t.Fatalf("Init: %v", err) - } + ) resp, err := p.Generate(context.Background(), &ai.Request{ Prompt: "do a two-step task", @@ -91,17 +89,14 @@ func TestGenerateToolLoopIsBounded(t *testing.T) { })) defer srv.Close() - p := &Provider{} - if err := p.Init( + p := NewProvider( ai.WithAPIKey("test"), ai.WithBaseURL(srv.URL), ai.WithModel("test-model"), ai.WithToolHandler(func(ctx context.Context, tc ai.ToolCall) ai.ToolResult { return ai.ToolResult{ID: tc.ID, Content: "done"} }), - ); err != nil { - t.Fatalf("Init: %v", err) - } + ) if _, err := p.Generate(context.Background(), &ai.Request{ Prompt: "never finish", @@ -110,11 +105,11 @@ func TestGenerateToolLoopIsBounded(t *testing.T) { t.Fatalf("Generate: %v", err) } - // Initial call + at most maxToolRounds follow-ups. - if calls > maxToolRounds+1 { - t.Fatalf("model calls = %d, want at most %d — the loop must be bounded", calls, maxToolRounds+1) + // Initial call + at most openaiapi.MaxToolRounds follow-ups. + if calls > openaiapi.MaxToolRounds+1 { + t.Fatalf("model calls = %d, want at most %d — the loop must be bounded", calls, openaiapi.MaxToolRounds+1) } - if calls < maxToolRounds { + if calls < openaiapi.MaxToolRounds { t.Fatalf("model calls = %d — expected the loop to keep going while tool calls keep coming", calls) } } diff --git a/ai/options.go b/ai/options.go index 33d011c457..84faa799ad 100644 --- a/ai/options.go +++ b/ai/options.go @@ -2,6 +2,7 @@ package ai import ( "context" + "net/http" ) // Options for model configuration @@ -23,6 +24,11 @@ type Options struct { Thinking ThinkingMode // Effort controls reasoning depth for providers that support it. Effort string + // Transport is the HTTP round tripper used by providers that make live + // API calls. Nil uses the standard library default transport. Inject a + // fake for tests. + Transport http.RoundTripper + // NoCache disables prompt-prefix caching for providers that support it // (e.g. Anthropic cache_control). Caching is on by default because the // dominant caller — the agent tool loop — re-sends an identical prefix on @@ -143,3 +149,12 @@ func WithEffort(effort string) Option { o.Effort = effort } } + +// WithTransport sets the HTTP round tripper used for provider API calls. +// Nil uses the standard library default transport. Inject a fake +// RoundTripper in tests to stub provider responses without a live server. +func WithTransport(rt http.RoundTripper) Option { + return func(o *Options) { + o.Transport = rt + } +} diff --git a/ai/stream_conformance_test.go b/ai/stream_conformance_test.go index 22b1d08933..fd09f9ac4b 100644 --- a/ai/stream_conformance_test.go +++ b/ai/stream_conformance_test.go @@ -231,6 +231,10 @@ func TestConfiguredProviderStreamsSkipWithoutCredentials(t *testing.T) { } stream, err := ai.New(tc.provider, opts...).Stream(context.Background(), &ai.Request{Prompt: "Reply with exactly: ok"}) if err != nil { + var httpErr *ai.HTTPError + if errors.As(err, &httpErr) && (httpErr.StatusCode() == http.StatusUnauthorized || httpErr.StatusCode() == http.StatusNotFound) { + t.Skipf("credential/model error for %s; skipping", tc.provider) + } t.Fatalf("Stream returned error: %v", err) } defer stream.Close() diff --git a/ai/together/together.go b/ai/together/together.go index d54692280d..235b645fc8 100644 --- a/ai/together/together.go +++ b/ai/together/together.go @@ -13,13 +13,7 @@ package together import ( - "bytes" "context" - "encoding/json" - "fmt" - "io" - "net/http" - "strings" "go-micro.dev/v6/ai" "go-micro.dev/v6/ai/internal/openaiapi" @@ -34,194 +28,25 @@ func init() { } type Provider struct { - opts ai.Options + core *openaiapi.Client } func NewProvider(opts ...ai.Option) *Provider { - options := ai.NewOptions(opts...) - if options.Model == "" { - options.Model = "meta-llama/Llama-3.3-70B-Instruct-Turbo" - } - if options.BaseURL == "" { - options.BaseURL = "https://api.together.xyz" - } - return &Provider{opts: options} + return &Provider{core: openaiapi.New(openaiapi.Config{ + Name: "together", + DefaultBase: "https://api.together.xyz", + DefaultModel: "meta-llama/Llama-3.3-70B-Instruct-Turbo", + }, opts...)} } -func (p *Provider) Init(opts ...ai.Option) error { - for _, o := range opts { - o(&p.opts) - } - return nil -} - -func (p *Provider) Options() ai.Options { return p.opts } -func (p *Provider) String() string { return "together" } +func (p *Provider) Init(opts ...ai.Option) error { return p.core.Init(opts...) } +func (p *Provider) Options() ai.Options { return p.core.Options() } +func (p *Provider) String() string { return p.core.String() } func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (*ai.Response, error) { - var tools []map[string]any - for _, t := range req.Tools { - tools = append(tools, map[string]any{ - "type": "function", - "function": map[string]any{ - "name": t.Name, - "description": t.Description, - "parameters": map[string]any{ - "type": "object", - "properties": t.Properties, - }, - }, - }) - } - - messages := []map[string]any{ - {"role": "system", "content": req.SystemPrompt}, - {"role": "user", "content": req.Prompt}, - } - - apiReq := map[string]any{ - "model": p.opts.Model, - "messages": messages, - } - if len(tools) > 0 { - apiReq["tools"] = tools - } - - resp, rawMessage, err := p.callAPI(ctx, apiReq) - if err != nil { - return nil, err - } - if len(resp.ToolCalls) == 0 { - return resp, nil - } - - // Tool execution loop: execute tools, send results back, and keep the - // tools on offer so the model can take the next step. A follow-up without - // "tools" asks the model to continue with its hands tied — the call it - // wanted comes back written out as prose — and without a loop a second - // step is impossible whatever the model wants. Bounded so a model that - // never stops asking cannot run forever. - if p.opts.ToolHandler != nil { - // Copied rather than aliased: append on a slice that shares an array - // with messages would overwrite it on a later round. - followUpMessages := append([]map[string]any(nil), messages...) - pending := resp.ToolCalls - raw := rawMessage - for round := 0; len(pending) > 0 && round < maxToolRounds; round++ { - followUpMessages = append(followUpMessages, map[string]any{ - "role": "assistant", - "content": raw["content"], - "tool_calls": raw["tool_calls"], - }) - for _, tc := range pending { - content := p.opts.ToolHandler(ctx, tc).Content - followUpMessages = append(followUpMessages, map[string]any{ - "role": "tool", - "tool_call_id": tc.ID, - "content": content, - }) - } - - followUpReq := map[string]any{ - "model": p.opts.Model, - "messages": followUpMessages, - } - if len(tools) > 0 { - followUpReq["tools"] = tools - } - - followUpResp, followUpRaw, err := p.callAPI(ctx, followUpReq) - if err != nil { - break - } - if followUpResp.Reply != "" { - resp.Answer = followUpResp.Reply - } - pending, raw = followUpResp.ToolCalls, followUpRaw - resp.ToolCalls = append(resp.ToolCalls, followUpResp.ToolCalls...) - } - } - - return resp, nil + return p.core.Generate(ctx, req, opts...) } -// maxToolRounds bounds the tool-execution loop in a single Generate. Each -// round is a model call plus the tools it asks for, so this is the ceiling on -// one question's cost as well as its length; it is high enough that no honest -// piece of multi-step work reaches it. -const maxToolRounds = 12 - func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) { - return openaiapi.Stream(ctx, p.opts, req, "/v1/chat/completions") -} - -func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) { - reqBody, err := json.Marshal(req) - if err != nil { - return nil, nil, fmt.Errorf("failed to marshal request: %w", err) - } - - apiURL := strings.TrimRight(p.opts.BaseURL, "/") + "/v1/chat/completions" - httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(reqBody)) - if err != nil { - return nil, nil, fmt.Errorf("failed to create request: %w", err) - } - - httpReq.Header.Set("Content-Type", "application/json") - httpReq.Header.Set("Authorization", "Bearer "+p.opts.APIKey) - - httpResp, err := http.DefaultClient.Do(httpReq) - if err != nil { - return nil, nil, fmt.Errorf("API request failed: %w", err) - } - defer httpResp.Body.Close() - - respBody, _ := io.ReadAll(httpResp.Body) - if httpResp.StatusCode != http.StatusOK { - return nil, nil, ai.NewHTTPError(httpResp, respBody) - } - - var chatResp struct { - Choices []struct { - Message struct { - Content string `json:"content"` - ToolCalls []struct { - ID string `json:"id"` - Function struct { - Name string `json:"name"` - Arguments string `json:"arguments"` - } `json:"function"` - } `json:"tool_calls"` - } `json:"message"` - } `json:"choices"` - } - - if err := json.Unmarshal(respBody, &chatResp); err != nil { - return nil, nil, fmt.Errorf("failed to parse response: %w", err) - } - if len(chatResp.Choices) == 0 { - return nil, nil, fmt.Errorf("no response from API") - } - - choice := chatResp.Choices[0] - response := &ai.Response{Reply: choice.Message.Content} - - for _, tc := range choice.Message.ToolCalls { - var input map[string]any - if err := json.Unmarshal([]byte(tc.Function.Arguments), &input); err != nil { - input = map[string]any{} - } - response.ToolCalls = append(response.ToolCalls, ai.ToolCall{ - ID: tc.ID, - Name: tc.Function.Name, - Input: input, - }) - } - - rawMessage := map[string]any{ - "content": choice.Message.Content, - "tool_calls": choice.Message.ToolCalls, - } - - return response, rawMessage, nil + return p.core.Stream(ctx, req, opts...) } diff --git a/ai/tools.go b/ai/tools.go index 94ae2dd839..d8209ee98a 100644 --- a/ai/tools.go +++ b/ai/tools.go @@ -132,7 +132,7 @@ func (t *Tools) Discover() ([]Tool, error) { // Deterministic order. The registry iterates a map, so without this the // tool list is shuffled on every discovery — which silently defeats // provider prompt caching (Anthropic cache_control, Gemini implicit - // caching): both key on a byte-identical prefix, and the tool catalogue + // caching): both key on a byte-identical prefix, and the tool catalog // is the bulk of that prefix. sort.SliceStable(out, func(i, j int) bool { if out[i].Name != out[j].Name { diff --git a/ai/tools_test.go b/ai/tools_test.go index d263067af3..5d2df44f81 100644 --- a/ai/tools_test.go +++ b/ai/tools_test.go @@ -120,7 +120,7 @@ func TestWithTools(t *testing.T) { // an explicit sort the tool list is shuffled on every call — which silently // defeats provider prompt caching: Anthropic cache_control and Gemini // implicit caching both key on a byte-identical prefix, and the tool -// catalogue is the bulk of that prefix. +// catalog is the bulk of that prefix. func TestDiscoverOrderIsDeterministic(t *testing.T) { reg := registry.NewMemoryRegistry() for _, name := range []string{"zulu", "alpha", "mike", "bravo", "yankee"} { @@ -138,7 +138,7 @@ func TestDiscoverOrderIsDeterministic(t *testing.T) { // Two versions of one service, with different endpoint sets: discovery // must not duplicate its tools, and must pick the same version — the - // highest — every time, or the serialized catalogue still churns. + // highest — every time, or the serialized catalog still churns. for _, v := range []struct{ version, endpoint string }{ {"1.0.0", "Svc.Old"}, {"2.0.0", "Svc.New"}, diff --git a/check-versions.sh b/check-versions.sh deleted file mode 100755 index 415399e60b..0000000000 --- a/check-versions.sh +++ /dev/null @@ -1,51 +0,0 @@ -#!/usr/bin/env bash -# check-versions.sh — Run `go list -m -versions` for every phantom module path -# declared in retract-phantom.sh. -# -# Usage: ./check-versions.sh [--dry-run] -# -# --dry-run print the paths only, without running go list - -set -euo pipefail - -SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" -SOURCE="$SCRIPT_DIR/retract-phantom.sh" - -DRY_RUN=false -for arg in "$@"; do - case "$arg" in - --dry-run) DRY_RUN=true ;; - esac -done - -if [[ ! -f "$SOURCE" ]]; then - echo "error: $SOURCE not found" >&2 - exit 1 -fi - -mapfile -t PATHS < <( - awk ' - /^PHANTOM_PATHS=\(/ { in_arr = 1; next } - in_arr && /^\)/ { exit } - in_arr { - gsub(/^[[:space:]]+|[[:space:]]+$/, "") - if ($0 != "" && $0 !~ /^#/) print - } - ' "$SOURCE" -) - -if [[ ${#PATHS[@]} -eq 0 ]]; then - echo "error: no phantom paths parsed from $SOURCE" >&2 - exit 1 -fi - -echo "checking ${#PATHS[@]} phantom module paths" -for path in "${PATHS[@]}"; do - if $DRY_RUN; then - echo "$path" - continue - fi - printf '%-70s ' "$path" - go list -m -versions "$path" 2>/dev/null || printf '(no versions found)' - echo -done diff --git a/client/grpc/grpc_pool_test.go b/client/grpc/grpc_pool_test.go index 3e71416355..6b8a22dfab 100644 --- a/client/grpc/grpc_pool_test.go +++ b/client/grpc/grpc_pool_test.go @@ -6,8 +6,8 @@ import ( "testing" "time" + pb "go-micro.dev/v6/test/helloworld" "google.golang.org/grpc" - pb "google.golang.org/grpc/examples/helloworld/helloworld" ) func testPool(t *testing.T, size int, ttl time.Duration, idle int, ms int) { diff --git a/client/grpc/grpc_test.go b/client/grpc/grpc_test.go index ce8c43ce21..f7b1cd0c42 100644 --- a/client/grpc/grpc_test.go +++ b/client/grpc/grpc_test.go @@ -9,8 +9,8 @@ import ( "go-micro.dev/v6/errors" "go-micro.dev/v6/registry" "go-micro.dev/v6/selector" + pb "go-micro.dev/v6/test/helloworld" pgrpc "google.golang.org/grpc" - pb "google.golang.org/grpc/examples/helloworld/helloworld" ) // server is used to implement helloworld.GreeterServer. diff --git a/client/rpc_client.go b/client/rpc_client.go index f0f265d981..68ea6e43b2 100644 --- a/client/rpc_client.go +++ b/client/rpc_client.go @@ -14,6 +14,7 @@ import ( "go-micro.dev/v6/codec" raw "go-micro.dev/v6/codec/bytes" merrors "go-micro.dev/v6/errors" + "go-micro.dev/v6/internal/mucp" "go-micro.dev/v6/internal/util/buf" "go-micro.dev/v6/internal/util/net" "go-micro.dev/v6/internal/util/pool" @@ -64,18 +65,6 @@ func newRPCClient(opt ...Option) Client { return c } -func (r *rpcClient) newCodec(contentType string) (codec.NewCodec, error) { - if c, ok := r.opts.Codecs[contentType]; ok { - return c, nil - } - - if cf, ok := DefaultCodecs[contentType]; ok { - return cf, nil - } - - return nil, fmt.Errorf("unsupported Content-Type: %s", contentType) -} - func (r *rpcClient) call( ctx context.Context, node *registry.Node, @@ -126,19 +115,6 @@ func (r *rpcClient) call( // set the accept header msg.Header["Accept"] = req.ContentType() - // setup old protocol - reqCodec := setupProtocol(msg, node) - - // no codec specified - if reqCodec == nil { - var err error - reqCodec, err = r.newCodec(req.ContentType()) - - if err != nil { - return merrors.InternalServerError("go.micro.client", err.Error()) - } - } - dOpts := []transport.DialOption{ transport.WithStream(), } @@ -160,7 +136,15 @@ func (r *rpcClient) call( } seq := atomic.AddUint64(&r.seq, 1) - 1 - codec := newRPCCodec(msg, c, reqCodec, "") + codec, err := mucp.NewClient(c, mucp.Options{ + Request: msg, + Protocol: node.Metadata["protocol"], + Codecs: r.opts.Codecs, + Domain: packageID, + }) + if err != nil { + return merrors.InternalServerError(packageID, err.Error()) + } rsp := &rpcResponse{ socket: c, @@ -263,19 +247,6 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request // set the accept header msg.Header["Accept"] = req.ContentType() - // set old codecs - nCodec := setupProtocol(msg, node) - - // no codec specified - if nCodec == nil { - var err error - - nCodec, err = r.newCodec(req.ContentType()) - if err != nil { - return nil, merrors.InternalServerError("go.micro.client", err.Error()) - } - } - dOpts := []transport.DialOption{ transport.WithStream(), } @@ -294,7 +265,16 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request id := fmt.Sprintf("%v", seq) // create codec with stream id - codec := newRPCCodec(msg, c, nCodec, id) + codec, err := mucp.NewClient(c, mucp.Options{ + Request: msg, + Protocol: node.Metadata["protocol"], + Stream: id, + Codecs: r.opts.Codecs, + Domain: packageID, + }) + if err != nil { + return nil, merrors.InternalServerError(packageID, err.Error()) + } rsp := &rpcResponse{ socket: c, @@ -685,7 +665,7 @@ func (r *rpcClient) Publish(ctx context.Context, msg Message, opts ...PublishOpt } // encode message body - cf, err := r.newCodec(msg.ContentType()) + cf, err := mucp.Lookup(msg.ContentType(), r.opts.Codecs) if err != nil { return merrors.InternalServerError(packageID, err.Error()) } diff --git a/client/rpc_codec.go b/client/rpc_codec.go index f138f8136b..ce1fc1ca4b 100644 --- a/client/rpc_codec.go +++ b/client/rpc_codec.go @@ -1,23 +1,14 @@ package client import ( - "bytes" errs "errors" - "go-micro.dev/v6/codec" - raw "go-micro.dev/v6/codec/bytes" - "go-micro.dev/v6/codec/grpc" - "go-micro.dev/v6/codec/json" - "go-micro.dev/v6/codec/jsonrpc" - "go-micro.dev/v6/codec/proto" - "go-micro.dev/v6/codec/protorpc" - "go-micro.dev/v6/errors" - "go-micro.dev/v6/registry" - "go-micro.dev/v6/transport" - "go-micro.dev/v6/transport/headers" + "go-micro.dev/v6/internal/mucp" ) const ( + // lastStreamResponseError is sent by the server to signal the end of a + // streaming response. lastStreamResponseError = "EOS" ) @@ -25,258 +16,16 @@ const ( // the remote side of the RPC connection. type serverError string -func (e serverError) Error() string { - return string(e) -} +func (e serverError) Error() string { return string(e) } // errShutdown holds the specific error for closing/closed connections. var ( errShutdown = errs.New("connection is shut down") ) -type rpcCodec struct { - client transport.Client - codec codec.Codec +// DefaultContentType is the default content type for outbound requests. +const DefaultContentType = "application/json" - req *transport.Message - buf *readWriteCloser - - // signify if its a stream - stream string -} - -type readWriteCloser struct { - wbuf *bytes.Buffer - rbuf *bytes.Buffer -} - -var ( - // DefaultContentType header. - DefaultContentType = "application/json" - - // DefaultCodecs map. - DefaultCodecs = map[string]codec.NewCodec{ - "application/grpc": grpc.NewCodec, - "application/grpc+json": grpc.NewCodec, - "application/grpc+proto": grpc.NewCodec, - "application/protobuf": proto.NewCodec, - "application/json": json.NewCodec, - "application/json-rpc": jsonrpc.NewCodec, - "application/proto-rpc": protorpc.NewCodec, - "application/octet-stream": raw.NewCodec, - } - - // TODO: remove legacy codec list. - defaultCodecs = map[string]codec.NewCodec{ - "application/json": jsonrpc.NewCodec, - "application/json-rpc": jsonrpc.NewCodec, - "application/protobuf": protorpc.NewCodec, - "application/proto-rpc": protorpc.NewCodec, - "application/octet-stream": protorpc.NewCodec, - } -) - -func (rwc *readWriteCloser) Read(p []byte) (n int, err error) { - return rwc.rbuf.Read(p) -} - -func (rwc *readWriteCloser) Write(p []byte) (n int, err error) { - return rwc.wbuf.Write(p) -} - -func (rwc *readWriteCloser) Close() error { - rwc.rbuf.Reset() - rwc.wbuf.Reset() - - return nil -} - -func getHeaders(m *codec.Message) { - set := func(v, hdr string) string { - if len(v) > 0 { - return v - } - - return m.Header[hdr] - } - - // check error in header - m.Error = set(m.Error, headers.Error) - - // check endpoint in header - m.Endpoint = set(m.Endpoint, headers.Endpoint) - - // check method in header - m.Method = set(m.Method, headers.Method) - - // set the request id - m.Id = set(m.Id, headers.ID) -} - -func setHeaders(m *codec.Message, stream string) { - set := func(hdr, v string) { - if len(v) == 0 { - return - } - - m.Header[hdr] = v - } - - set(headers.ID, m.Id) - set(headers.Request, m.Target) - set(headers.Method, m.Method) - set(headers.Endpoint, m.Endpoint) - set(headers.Error, m.Error) - - if len(stream) > 0 { - set(headers.Stream, stream) - } -} - -// setupProtocol sets up the old protocol. -func setupProtocol(msg *transport.Message, node *registry.Node) codec.NewCodec { - protocol := node.Metadata["protocol"] - - // got protocol - if len(protocol) > 0 { - return nil - } - - // processing topic publishing - if len(msg.Header[headers.Message]) > 0 { - return nil - } - - // no protocol use old codecs - switch msg.Header["Content-Type"] { - case "application/json": - msg.Header["Content-Type"] = "application/json-rpc" - case "application/protobuf": - msg.Header["Content-Type"] = "application/proto-rpc" - } - - return defaultCodecs[msg.Header["Content-Type"]] -} - -func newRPCCodec(req *transport.Message, client transport.Client, c codec.NewCodec, stream string) codec.Codec { - rwc := &readWriteCloser{ - wbuf: bytes.NewBuffer(nil), - rbuf: bytes.NewBuffer(nil), - } - - return &rpcCodec{ - buf: rwc, - client: client, - codec: c(rwc), - req: req, - stream: stream, - } -} - -func (c *rpcCodec) Write(message *codec.Message, body interface{}) error { - c.buf.wbuf.Reset() - - // create header - if message.Header == nil { - message.Header = map[string]string{} - } - - // copy original header - for k, v := range c.req.Header { - message.Header[k] = v - } - - // set the mucp headers - setHeaders(message, c.stream) - - // if body is bytes Frame don't encode - if body != nil { - if b, ok := body.(*raw.Frame); ok { - // set body - message.Body = b.Data - } else { - // write to codec - if err := c.codec.Write(message, body); err != nil { - return errors.InternalServerError("go.micro.client.codec", err.Error()) - } - // set body - message.Body = c.buf.wbuf.Bytes() - } - } - - // create new transport message - msg := transport.Message{ - Header: message.Header, - Body: message.Body, - } - - // send the request - if err := c.client.Send(&msg); err != nil { - return errors.InternalServerError("go.micro.client.transport", err.Error()) - } - - return nil -} - -func (c *rpcCodec) ReadHeader(msg *codec.Message, r codec.MessageType) error { - var tm transport.Message - - // read message from transport - if err := c.client.Recv(&tm); err != nil { - return errors.InternalServerError("go.micro.client.transport", err.Error()) - } - - c.buf.rbuf.Reset() - c.buf.rbuf.Write(tm.Body) - - // set headers from transport - msg.Header = tm.Header - - // read header - err := c.codec.ReadHeader(msg, r) - - // get headers - getHeaders(msg) - - // return header error - if err != nil { - return errors.InternalServerError("go.micro.client.codec", err.Error()) - } - - return nil -} - -func (c *rpcCodec) ReadBody(b interface{}) error { - // read body - // read raw data - if v, ok := b.(*raw.Frame); ok { - v.Data = c.buf.rbuf.Bytes() - return nil - } - - if err := c.codec.ReadBody(b); err != nil { - return errors.InternalServerError("go.micro.client.codec", err.Error()) - } - - return nil -} - -func (c *rpcCodec) Close() error { - if err := c.buf.Close(); err != nil { - return err - } - - if err := c.codec.Close(); err != nil { - return err - } - - if err := c.client.Close(); err != nil { - return errors.InternalServerError("go.micro.client.transport", err.Error()) - } - - return nil -} - -func (c *rpcCodec) String() string { - return "rpc" -} +// DefaultCodecs is the default codec map. The mucp wire framing lives in +// go-micro.dev/v6/internal/mucp. +var DefaultCodecs = mucp.DefaultCodecs diff --git a/cmd/cmd.go b/cmd/cmd.go index 4dbb5762b4..aab9bc9146 100644 --- a/cmd/cmd.go +++ b/cmd/cmd.go @@ -253,7 +253,6 @@ var ( DefaultRegistries = map[string]func(...registry.Option) registry.Registry{ "memory": registry.NewMemoryRegistry, - "mdns": registry.NewMDNSRegistry, } DefaultSelectors = map[string]func(...selector.Option) selector.Selector{} @@ -681,6 +680,19 @@ func (c *cmd) Before(ctx *cli.Context) error { *c.opts.Config = rc } } + + // Sync the configured registry back to the package global so helpers like + // registry.ListServices (used by the CLI registry command, the gateway and + // MCP discovery) see the registry selected via flags/env, not the default. + registry.DefaultRegistry = *c.opts.Registry + + // Same for the client: cmd/pkg init snapshots client.DefaultClient into + // DefaultCmd before cmd/defaults replaces the global with a fresh instance + // (see defaults.go). Without this sync the transport selected via flags/env + // lands only on the snapshot, while the HTTP API gateway and MCP dispatch + // keep using the global client's default (http) transport and dial backend + // services directly instead of routing over the configured transport. + client.DefaultClient = *c.opts.Client return nil } diff --git a/cmd/defaults/defaults.go b/cmd/defaults/defaults.go index d3580072ab..58e2bb1c43 100644 --- a/cmd/defaults/defaults.go +++ b/cmd/defaults/defaults.go @@ -8,9 +8,15 @@ // The micro CLI imports it, so `micro --registry etcd ...` and friends behave // unchanged. Library binaries that construct their own Registry, Broker, // Store, and Transport can skip this import and shed the NATS, Consul, etcd, -// RabbitMQ, Redis, Postgres, and MySQL machinery — roughly 40 packages — from -// their builds. A binary that skips it but still selects a plugin by flag gets -// a clear "not registered" error naming this package. +// RabbitMQ, Redis, Postgres, MySQL, mDNS, and file (bbolt) machinery — roughly +// 40 packages — from their builds. A binary that skips it but still selects a +// plugin by flag gets a clear "not registered" error naming this package. +// +// Linking the defaults also restores the historical package defaults that +// used to be compiled into core: mdns discovery and file-backed storage +// (registry.DefaultRegistry, store.DefaultStore / store.NewStore). Core +// keeps lightweight in-memory defaults so binaries that link no plugins still +// work out of the box. package defaults import ( @@ -19,14 +25,20 @@ import ( nbroker "go-micro.dev/v6/broker/nats" rabbit "go-micro.dev/v6/broker/rabbitmq" "go-micro.dev/v6/cache/redis" + "go-micro.dev/v6/client" "go-micro.dev/v6/registry/consul" "go-micro.dev/v6/registry/etcd" + mdns "go-micro.dev/v6/registry/mdns" nregistry "go-micro.dev/v6/registry/nats" + "go-micro.dev/v6/store/file" "go-micro.dev/v6/store/mysql" natsjskv "go-micro.dev/v6/store/nats-js-kv" postgres "go-micro.dev/v6/store/postgres" ntransport "go-micro.dev/v6/transport/nats" + "go-micro.dev/v6/registry" + "go-micro.dev/v6/store" + // Registers the "nats" plugin profile (--profile nats). _ "go-micro.dev/v6/service/profile/natsprofile" ) @@ -37,13 +49,36 @@ func init() { cmd.DefaultRegistries["consul"] = consul.NewConsulRegistry cmd.DefaultRegistries["etcd"] = etcd.NewEtcdRegistry + cmd.DefaultRegistries["mdns"] = mdns.NewRegistry cmd.DefaultRegistries["nats"] = nregistry.NewNatsRegistry cmd.DefaultTransports["nats"] = ntransport.NewTransport + cmd.DefaultStores["file"] = file.NewStore cmd.DefaultStores["mysql"] = mysql.NewMysqlStore cmd.DefaultStores["natsjskv"] = natsjskv.NewStore cmd.DefaultStores["postgres"] = postgres.NewStore cmd.DefaultCaches["redis"] = redis.NewRedisCache + + // Restore the historical defaults that core used to compile in: mdns + // discovery and file-backed storage. Without this, a binary linking the + // plugins (the CLI) would land on the lightweight memory defaults. + registry.DefaultRegistry = mdns.NewRegistry() + store.RegisterDefault(file.NewStore) + + // cmd.DefaultCmd's registry is snapshotted at package init, before the + // plugins above link in, so point its snapshot at the restored default + // too — cmd.Before echoes it back to registry.DefaultRegistry after flag + // parsing and would otherwise clobber the freshly installed one. + if r := cmd.DefaultCmd.Options().Registry; r != nil { + *r = registry.DefaultRegistry + } + + // client.DefaultClient snapshots registry.DefaultRegistry at package + // init, before this package links in, so a CLI that calls out through + // it (micro chat, api, gateway, flow, run) would resolve services + // against the memory registry while everything else talks to mdns. + // Rebuild it against the restored default. + client.DefaultClient = client.NewClient(client.Registry(registry.DefaultRegistry)) } diff --git a/cmd/micro/cli/build/build.go b/cmd/micro/cli/build/build.go index 752b467598..4e85db2ca0 100644 --- a/cmd/micro/cli/build/build.go +++ b/cmd/micro/cli/build/build.go @@ -10,6 +10,7 @@ import ( "strings" "github.com/urfave/cli/v2" + "go-micro.dev/v6/cmd" "go-micro.dev/v6/cmd/micro/run/config" ) @@ -37,7 +38,7 @@ func Build(c *cli.Context) error { if outDir == "" { outDir = filepath.Join(absDir, "bin") } - if err := os.MkdirAll(outDir, 0755); err != nil { + if err := os.MkdirAll(outDir, 0o755); err != nil { return fmt.Errorf("failed to create output dir: %w", err) } @@ -88,7 +89,8 @@ func buildService(name, dir, outDir, targetOS, targetArch string) error { // Build command buildCmd := exec.Command("go", "build", "-o", outPath, ".") buildCmd.Dir = dir - buildCmd.Env = append(os.Environ(), + buildCmd.Env = append( + os.Environ(), "GOOS="+targetOS, "GOARCH="+targetArch, "CGO_ENABLED=0", @@ -145,12 +147,13 @@ func Docker(c *cli.Context) error { return nil } -const dockerfileTemplate = `FROM golang:1.22-alpine AS builder +const dockerfileTemplate = `FROM golang:1.25-alpine AS builder WORKDIR /app COPY go.mod go.sum ./ RUN go mod download COPY . . -RUN CGO_ENABLED=0 go build -o /service . +ARG SVC +RUN CGO_ENABLED=0 go build -o /service ./$SVC FROM alpine:latest RUN apk --no-cache add ca-certificates @@ -159,18 +162,49 @@ EXPOSE %d CMD ["/service"] ` +// findModuleRoot walks up from dir until a go.mod is found. +// Services may live inside a parent module (e.g. examples/ in go-micro); +// the build context must be the module root so go.mod/go.sum resolve. +func findModuleRoot(dir string) string { + for { + if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil { + return dir + } + parent := filepath.Dir(dir) + if parent == dir { + return "" + } + dir = parent + } +} + func buildDockerImage(name, dir string, port int, tag, registry string, push bool) error { if port == 0 { port = 8080 } - // Generate Dockerfile if not exists - dockerfilePath := filepath.Join(dir, "Dockerfile") + moduleRoot := findModuleRoot(dir) + if moduleRoot == "" { + moduleRoot = dir + } + svcPath, err := filepath.Rel(moduleRoot, dir) + if err != nil { + return fmt.Errorf("failed to resolve service path: %w", err) + } + + // Dockerfile must live inside the build context. Use a distinct name when + // the service is a subdir of the module root, so we don't clobber the + // module's own Dockerfile. + dockerfileName := "Dockerfile" + if svcPath != "." { + dockerfileName = ".micro.Dockerfile" + } + dockerfilePath := filepath.Join(moduleRoot, dockerfileName) if _, err := os.Stat(dockerfilePath); os.IsNotExist(err) { - fmt.Printf("Generating Dockerfile for %s...\n", name) + fmt.Printf("Generating %s for %s...\n", dockerfileName, name) dockerfile := fmt.Sprintf(dockerfileTemplate, port) - if err := os.WriteFile(dockerfilePath, []byte(dockerfile), 0644); err != nil { - return fmt.Errorf("failed to write Dockerfile: %w", err) + if err := os.WriteFile(dockerfilePath, []byte(dockerfile), 0o644); err != nil { + return fmt.Errorf("failed to write %s: %w", dockerfileName, err) } } @@ -181,7 +215,11 @@ func buildDockerImage(name, dir string, port int, tag, registry string, push boo fmt.Printf(" Building \033[36m%s...\n", imageName) - buildCmd := exec.Command("docker", "build", "-t", imageName, dir) + buildCmd := exec.Command("docker", "build", + "-f", dockerfilePath, + "--build-arg", "SVC="+svcPath, + "-t", imageName, + moduleRoot) buildCmd.Stdout = os.Stdout buildCmd.Stderr = os.Stderr if err := buildCmd.Run(); err != nil { @@ -233,7 +271,7 @@ func Compose(c *cli.Context) error { var sb strings.Builder sb.WriteString("# Generated by micro build --compose\n") - sb.WriteString("version: '3.8'\n\nservices:\n") + sb.WriteString("\nservices:\n") sorted, err := cfg.TopologicalSort() if err != nil { @@ -263,8 +301,22 @@ func Compose(c *cli.Context) error { sb.WriteString(" environment:\n - MICRO_REGISTRY=mdns\n\n") } + // Default gateway: routes to all services via the shared registry. + sb.WriteString(" gw:\n") + sb.WriteString(" # image: ghcr.io/micro/go-micro:latest # prebuilt alternative\n") + sb.WriteString(" build:\n") + sb.WriteString(" context: .\n") + sb.WriteString(" dockerfile: internal/docker/Dockerfile\n") + sb.WriteString(" additional_contexts:\n repo: .\n") + sb.WriteString(" ports:\n - \"8080:8080\"\n") + sb.WriteString(" depends_on:\n") + for _, svc := range sorted { + fmt.Fprintf(&sb, " - %s\n", svc.Name) + } + sb.WriteString(" environment:\n - MICRO_REGISTRY=mdns\n - MICRO_ADMIN_USER=admin\n - MICRO_ADMIN_PASSWORD=micro\n\n") + output := filepath.Join(absDir, "docker-compose.yml") - if err := os.WriteFile(output, []byte(sb.String()), 0644); err != nil { + if err := os.WriteFile(output, []byte(sb.String()), 0o644); err != nil { return fmt.Errorf("failed to write docker-compose.yml: %w", err) } diff --git a/cmd/micro/gateway/auth_test.go b/cmd/micro/gateway/auth_test.go index c98e7a86aa..575af49976 100644 --- a/cmd/micro/gateway/auth_test.go +++ b/cmd/micro/gateway/auth_test.go @@ -1,10 +1,14 @@ package gateway import ( + "encoding/json" "flag" + "os" "testing" "github.com/urfave/cli/v2" + "go-micro.dev/v6/store" + "golang.org/x/crypto/bcrypt" ) func TestIsExposed(t *testing.T) { @@ -84,3 +88,72 @@ func TestTokenMatchesEmpty(t *testing.T) { t.Fatal("an empty static token must never match") } } + +func TestEnsureAdminFromEnv(t *testing.T) { + setEnv := func(k, v string) { + t.Helper() + if v == "" { + os.Unsetenv(k) + } else { + os.Setenv(k, v) + } + } + t.Cleanup(func() { setEnv("MICRO_ADMIN_PASSWORD", ""); setEnv("MICRO_ADMIN_USER", "") }) + + st := store.NewMemoryStore() + readAcc := func(t *testing.T, id string) Account { + t.Helper() + recs, _ := st.Read("auth/" + id) + if len(recs) == 0 { + t.Fatalf("no account %q in store", id) + } + var acc Account + if err := json.Unmarshal(recs[0].Value, &acc); err != nil { + t.Fatalf("unmarshal: %v", err) + } + return acc + } + + // No env → no account created. + setEnv("MICRO_ADMIN_PASSWORD", "") + if err := ensureAdminFromEnv(st); err != nil { + t.Fatalf("ensureAdminFromEnv: %v", err) + } + if recs, _ := st.Read("auth/admin"); len(recs) != 0 { + t.Fatal("account created without MICRO_ADMIN_PASSWORD") + } + + // Env set → admin account created, password hashed and verifiable. + setEnv("MICRO_ADMIN_PASSWORD", "micro") + if err := ensureAdminFromEnv(st); err != nil { + t.Fatalf("ensureAdminFromEnv: %v", err) + } + admin := readAcc(t, "admin") + if admin.Type != "admin" || len(admin.Scopes) != 1 || admin.Scopes[0] != "*" { + t.Fatalf("admin account wrong: %+v", admin) + } + if err := bcrypt.CompareHashAndPassword([]byte(admin.Metadata["password_hash"]), []byte("micro")); err != nil { + t.Fatal("stored password hash does not match 'micro'") + } + + // Idempotent: existing account is not overwritten. + admin.Metadata["changed"] = "true" + b, _ := json.Marshal(admin) + _ = st.Write(&store.Record{Key: "auth/admin", Value: b}) + if err := ensureAdminFromEnv(st); err != nil { + t.Fatalf("ensureAdminFromEnv: %v", err) + } + if readAcc(t, "admin").Metadata["changed"] != "true" { + t.Fatal("existing account was overwritten") + } + + // A deleted admin stays deleted. + _ = st.Delete("auth/admin") + _ = st.Write(&store.Record{Key: "auth/.admin-deleted", Value: []byte("true")}) + if err := ensureAdminFromEnv(st); err != nil { + t.Fatalf("ensureAdminFromEnv: %v", err) + } + if recs, _ := st.Read("auth/admin"); len(recs) != 0 { + t.Fatal("deleted admin was recreated") + } +} diff --git a/cmd/micro/gateway/server.go b/cmd/micro/gateway/server.go index 88c208c2d6..7f415d282e 100644 --- a/cmd/micro/gateway/server.go +++ b/cmd/micro/gateway/server.go @@ -13,6 +13,7 @@ import ( "io/fs" "log" "net/http" + "net/http/pprof" "os" "os/signal" "path/filepath" @@ -24,6 +25,7 @@ import ( "text/template" "time" + "github.com/prometheus/client_golang/prometheus/promhttp" "github.com/urfave/cli/v2" "go-micro.dev/v6/ai" _ "go-micro.dev/v6/ai/anthropic" @@ -39,9 +41,15 @@ import ( "go-micro.dev/v6/cmd" codecBytes "go-micro.dev/v6/codec/bytes" "go-micro.dev/v6/gateway/mcp" + "go-micro.dev/v6/metadata" "go-micro.dev/v6/registry" "go-micro.dev/v6/store" "go-micro.dev/v6/wrapper/x402" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/propagation" + "go.opentelemetry.io/otel/trace" "golang.org/x/crypto/bcrypt" "golang.org/x/sync/errgroup" ) @@ -256,6 +264,17 @@ func wrapAuth(authRequired func(http.HandlerFunc) http.HandlerFunc) func(http.Ha } } +// registryServices returns the set of currently-registered service names. +func registryServices() map[string]bool { + set := map[string]bool{} + if svcs, err := registry.ListServices(); err == nil { + for _, s := range svcs { + set[s.Name] = true + } + } + return set +} + func getDashboardData() (serviceCount, runningCount, stoppedCount int, statusDot string) { homeDir, err := os.UserHomeDir() if err != nil { @@ -266,6 +285,9 @@ func getDashboardData() (serviceCount, runningCount, stoppedCount int, statusDot if err != nil { return } + // A service is "running" if it heartbeats in the registry. The PID signal + // check can't work across containers, so never rely on local processes. + running := registryServices() for _, entry := range dirEntries { if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".pid") || strings.HasPrefix(entry.Name(), ".") { continue @@ -276,21 +298,13 @@ func getDashboardData() (serviceCount, runningCount, stoppedCount int, statusDot continue } lines := strings.Split(string(pidBytes), "\n") - pid := "-" - if len(lines) > 0 && len(lines[0]) > 0 { - pid = lines[0] + name := "" + if len(lines) > 2 { + name = lines[2] } serviceCount++ - if pid != "-" { - if _, err := os.FindProcess(parsePid(pid)); err == nil { - if processRunning(pid) { - runningCount++ - } else { - stoppedCount++ - } - } else { - stoppedCount++ - } + if name != "" && running[name] { + runningCount++ } else { stoppedCount++ } @@ -305,6 +319,17 @@ func getDashboardData() (serviceCount, runningCount, stoppedCount int, statusDot return } +// defaultAgentSettings returns the agent chat prefill from the gateway's own +// AI environment variables, used until the user saves settings via POST. +func defaultAgentSettings() map[string]string { + return map[string]string{ + "provider": os.Getenv("MICRO_AI_PROVIDER"), + "model": os.Getenv("MICRO_AI_MODEL"), + "api_key": os.Getenv("MICRO_AI_API_KEY"), + "base_url": os.Getenv("MICRO_AI_BASE_URL"), + } +} + func registerHandlers(mux *http.ServeMux, tmpls *templates, storeInst store.Store, authEnabled bool) { var wrap func(http.HandlerFunc) http.HandlerFunc @@ -378,6 +403,18 @@ func registerHandlers(mux *http.ServeMux, tmpls *templates, storeInst store.Stor return false } + // Prometheus scrape endpoint (Alloy scrapes micro-run:8080) + mux.Handle("/metrics", promhttp.Handler()) + + // pprof endpoints on the gateway mux so Pyroscope (via Alloy) can scrape + // continuous profiles from the compose gateway without a dedicated :6060 + // listener (that only exists under `micro run`, not in docker). + mux.HandleFunc("/debug/pprof/", pprof.Index) + mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline) + mux.HandleFunc("/debug/pprof/profile", pprof.Profile) + mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol) + mux.HandleFunc("/debug/pprof/trace", pprof.Trace) + // Serve static files with correct Content-Type mux.HandleFunc("/styles.css", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/css; charset=utf-8") @@ -529,13 +566,13 @@ func registerHandlers(mux *http.ServeMux, tmpls *templates, storeInst store.Stor if r.Method == http.MethodGet { recs, _ := storeInst.Read("agent/settings") if len(recs) == 0 { - json.NewEncoder(w).Encode(map[string]string{}) + json.NewEncoder(w).Encode(defaultAgentSettings()) return } var settings map[string]string if err := json.Unmarshal(recs[0].Value, &settings); err != nil { log.Printf("[agent] failed to parse settings: %v", err) - json.NewEncoder(w).Encode(map[string]string{}) + json.NewEncoder(w).Encode(defaultAgentSettings()) return } json.NewEncoder(w).Encode(settings) @@ -601,7 +638,19 @@ func registerHandlers(mux *http.ServeMux, tmpls *templates, storeInst store.Stor } } if apiKey == "" { - json.NewEncoder(w).Encode(map[string]string{"error": "No API key configured. Go to Agent settings to add one."}) + apiKey = os.Getenv("MICRO_AI_API_KEY") + } + if modelName == "" { + modelName = os.Getenv("MICRO_AI_MODEL") + } + if baseURL == "" { + baseURL = os.Getenv("MICRO_AI_BASE_URL") + } + if provider == "" { + provider = os.Getenv("MICRO_AI_PROVIDER") + } + if provider == "" && baseURL == "" { + json.NewEncoder(w).Encode(map[string]string{"error": "No provider configured. Go to Agent settings to add one."}) return } @@ -767,9 +816,17 @@ func registerHandlers(mux *http.ServeMux, tmpls *templates, storeInst store.Stor if len(response.ToolCalls) > 0 { var toolCalls []map[string]any for _, tc := range response.ToolCalls { + var result any + if tc.Result != "" { + if err := json.Unmarshal([]byte(tc.Result), &result); err != nil { + result = tc.Result + } + } toolCalls = append(toolCalls, map[string]any{ - "tool": tc.Name, - "input": tc.Input, + "tool": tc.Name, + "input": tc.Input, + "result": result, + "error": tc.Error, }) } result["tool_calls"] = toolCalls @@ -850,7 +907,7 @@ func registerHandlers(mux *http.ServeMux, tmpls *templates, storeInst store.Stor if len(parts) != 2 { continue } - apiPath := fmt.Sprintf("/api/%s/%s/%s", s.Name, parts[0], parts[1]) + apiPath := fmt.Sprintf("/%s/%s", s.Name, ep.Name) var params, response string if ep.Request != nil && len(ep.Request.Values) > 0 { params += "