AI 应用API网关LLM 网关后端前端【免费下载链接】axonhub⚡️ Open-source AI Gateway — Use any SDK to call 100 LLMs. Built-in failover, load balancing, cost control end-to-end tracing.项目地址https://gitcode.com/gh_mirrors/ax/axonhub点击查看免费下载AxonHub 是一个开源的 AI 网关提供与 Anthropic Messages API 原生兼容的接入层让现有 Anthropic SDK 客户端无需改动即可访问 Claude 及其他模型。本文聚焦 integration_test/anthropic/streaming/README.md 所描述的流式测试套件结合 streaming_test.go 的完整实现系统讲解流式响应的接收、聚合、工具调用、事件处理与错误恢复并延伸到 AxonHub 网关侧对流式事件的转换与聚合原理。读完本文你将掌握用 Anthropic Go SDK 编写健壮流式客户端代码的完整套路以及如何在 AxonHub 兼容层上完成流式功能的端到端验证。一、为什么需要流式测试AxonHub 的 Anthropic 兼容层AxonHub 支持原生 Anthropic Messages API路由端点为POST /anthropic/v1/messages与可选的POST /v1/messages详见 docs/zh/api-reference/anthropic-api.md。其核心价值在于自动转换客户端用 Anthropic 格式发来的请求会被 AxonHub 转换成 OpenAI、Gemini 等其他协议后转发给上游模型响应再被转换回 Anthropic 格式。这意味着你可以在 internal/server/api/anthropic.go 看到Anthropic 处理器复用同一套ChatCompletionOrchestrator编排管线只是注入了一个anthropic.NewInboundTransformer()用于入站格式转换。流式Streaming是这套转换链路中最复杂的部分事件必须按 Anthropic 的 SSE 事件规范逐条重建message_start、content_block_delta、message_delta、message_stop等事件类型一个都不能少。因此用真实的 Anthropic Go SDK 编写覆盖各场景的集成测试是验证网关流式兼容性的关键手段——这正是integration_test/anthropic/streaming目录存在的意义。二、测试环境准备流式测试位于独立 Go module 中模块声明为github.com/looplj/axonhub/anthropic_test依赖github.com/anthropics/anthropic-sdk-go v1.26.0Go 版本要求 1.25见 integration_test/anthropic/go.mod。测试运行前需要配置以下环境变量默认值见 config.go# 必填AxonHub 或 Anthropic 的 API Key export TEST_AXONHUB_API_KEYyour-api-key # 可选网关地址默认 http://localhost:8090/anthropic即 AxonHub 的 Anthropic 兼容端点 export TEST_ANTHROPIC_BASE_URLhttp://localhost:8090/anthropic # 可选测试模型默认 deepseek-chat也可以换成 claude-3-5-sonnet-20241022 等 export TEST_MODELdeepseek-chat # 可选显式指定 Trace / Thread ID默认会自动生成 export TEST_TRACE_IDtest-trace-123 export TEST_THREAD_IDtest-thread-456 # 可选关闭追踪/线程头 export TEST_DISABLE_TRACEfalse export TEST_DISABLE_THREADfalse注意两点实现细节其一当TEST_AXONHUB_API_KEY未设置时NewTestHelper会通过t.Skipf跳过测试而不是直接失败见 helper.go便于在未配置密钥的环境中安全执行其二CreateMessageStreamWithHeaders在调用点把AH-Trace-Id、AH-Thread-Id请求头注入到每个流式请求中helper.go这些头部正是 AxonHub 构建追踪所需。三、六个流式测试用例详解streaming_test.go 中定义了 6 个测试函数覆盖了流式功能的全部关键场景。3.1 TestBasicStreamingChatCompletion基本流式聊天补全该用例验证最基础的流式能力向模型发送问题后逐块接收流式事件边接收边聚合出完整文本最后校验内容。params : anthropic.MessageNewParams{ Model: helper.GetModel(), Messages: []anthropic.MessageParam{anthropic.NewUserMessage(anthropic.NewTextBlock(question))}, MaxTokens: 1024, } stream : helper.CreateMessageStreamWithHeaders(ctx, params) defer stream.Close() var message anthropic.Message var fullResponse string for stream.Next() { event : stream.Current() err : message.Accumulate(event) // 官方 SDK 的累积器按事件类型归并到 Message helper.AssertNoError(t, err, Failed to accumulate streaming event) switch event : event.AsAny().(type) { case anthropic.ContentBlockDeltaEvent: // 文本增量 if textDelta, ok : event.Delta.AsAny().(anthropic.TextDelta); ok { fullResponse textDelta.Text } case anthropic.MessageDeltaEvent: // 流结束信息stop_reason if event.Delta.StopReason ! { t.Logf(Stream completed with stop reason: %s, event.Delta.StopReason) } case anthropic.MessageStartEvent: // 流开始含 message ID t.Logf(Stream started with message ID: %s, message.ID) } } helper.AssertNoError(t, stream.Err(), Stream encountered an error)这里展示了流式客户端最重要的两条并行路径message.Accumulate(event)让官方 SDK 把分散的事件还原成完整的anthropic.Message对象类型断言分支则在 UI 场景下实现实时反馈比如边接收边打印fullResponse。测试最后会从累积出的message.Content中重新提取文本并校验其包含提问中的关键词robot/paint保证流式聚合结果与一次性的非流式响应在语义上一致。3.2 TestLongResponseStreaming长响应流式长响应会生成大量 chunk是对流式聚合与性能的考验。该用例要求模型写出光合作用photosynthesis的详细解释并设置了两条硬性校验聚合后的响应文本长度必须 ≥ 100 字符响应必须至少包含photosynthesis、light、chlorophyll、carbon dioxide、oxygen五个关键词中的两个。expectedTerms : []string{photosynthesis, light, chlorophyll, carbon dioxide, oxygen} foundTerms : 0 for _, term : range expectedTerms { if testutil.ContainsCaseInsensitive(responseText, term) { foundTerms } } if foundTerms 2 { t.Errorf(Expected explanation to contain key terms, found %d/%d, foundTerms, len(expectedTerms)) }这组断言的意义在于长文本跨多个 chunk 传输时如果网关在格式转换中出现丢块、乱序或截断关键词统计会立刻暴露问题。这也是生产环境中评估流式感知速度与完整性之间的平衡点。3.3 TestStreamingResponseWithTools流式模式下的工具调用这是最有代表性的场景流式模式下模型可能在生成过程中发起工具调用客户端需要识别tool_use块、执行工具、把结果回传并继续第二轮流式对话。测试定义了两个工具——calculate数学计算与get_current_weather查天气通过anthropic.ToolParam声明 JSON SchemacalculatorTool : anthropic.ToolParam{ Name: calculate, Description: anthropic.String(Perform mathematical calculations), InputSchema: anthropic.ToolInputSchemaParam{ Type: object, Properties: map[string]interface{}{ expression: map[string]string{type: string}, }, Required: []string{expression}, }, }提问What is 25 * 4 and what is the weather in Tokyo?会诱导模型先输出工具调用。流式响应累积完成后测试检查message.StopReason anthropic.StopReasonToolUse遍历message.Content中的AsToolUse()块拿到工具名与调用参数模拟工具执行结果calculate返回100天气工具返回东京天气把message.ToParam()即带工具调用的助手消息与anthropic.NewToolResultBlock(toolUseBlock.ID, toolResult, false)包装的用户消息追加到对话历史重新发起第二轮流式请求取回最终自然语言答复并校验非空。这段代码演示了流式工具调用的完整闭环流式识别 → 工具执行 → 结果注入 → 二次流式。注意工具结果块与工具调用 ID 的关联toolUseBlock.ID这是多轮对话中正确配对工具结果的依据。3.4 TestStreamingErrorHandling流式错误处理该用例故意传入非法参数MaxTokens: -1负值验证流式请求在出错时不会被 panic而是通过stream.Err()暴露错误params : anthropic.MessageNewParams{ Model: helper.GetModel(), Messages: []anthropic.MessageParam{anthropic.NewUserMessage(anthropic.NewTextBlock(question))}, MaxTokens: -1, // Invalid negative value } stream : helper.CreateMessageStreamWithHeaders(ctx, params) defer stream.Close() for stream.Next() { // Just consume events until error } err : stream.Err() if err nil { t.Log(Streaming request accepted despite invalid parameters) } else { t.Logf(Correctly caught streaming error: %v, err) }测试对两种结果都持宽容态度记录日志而非硬性失败因为不同模型/网关对负值参数的处理策略不同。对开发者而言关键教训是流式消费循环结束后必须检查stream.Err()不能只依赖stream.Next()的返回值因为传输中途断流、网络错误等都会在Err()中体现。AxonHub 侧对应的错误响应格式为{type: error, error: {type: invalid_request_error, message: ...}}见 docs/zh/api-reference/anthropic-api.md。3.5 TestStreamingEventHandling流式事件类型统计Anthropic 流式协议由多种事件构成该用例逐一计数验证要求至少收到 1 个message_start、1 个message_stop以及若干个content_block_delta文本增量事件var eventCount, textDeltaCount, messageStartCount, messageStopCount int for stream.Next() { eventCount event : stream.Current() err : message.Accumulate(event) helper.AssertNoError(t, err, Failed to accumulate streaming event) switch event.AsAny().(type) { case anthropic.ContentBlockDeltaEvent: textDeltaCount case anthropic.MessageStartEvent: messageStartCount case anthropic.MessageStopEvent: messageStopCount } } if eventCount 0 || textDeltaCount 0 || messageStartCount 0 || messageStopCount 0 { t.Error(Expected to receive streaming events) }这些断言直接对应当前仓库中 Anthropic 流式协议的事件骨架message_start携带 message ID 与初始 usagecontent_block_delta以text_delta形式逐片推送文本message_delta携带stop_reason与最终 usagemessage_stop标记流结束。网关侧的聚合器正是按这套事件序列工作的详见第五节。3.6 TestStreamingWithSystemPrompt系统提示词流式最后验证系统提示词在流式链路中的传递与上下文保持。测试让模型扮演量子物理教授要求用恰当的技术术语解释量子计算systemPrompt : You are a quantum physics professor. Explain concepts clearly but use technical terms appropriately. params : anthropic.MessageNewParams{ Model: helper.GetModel(), Messages: []anthropic.MessageParam{anthropic.NewUserMessage(anthropic.NewTextBlock(question))}, MaxTokens: 1024, System: []anthropic.TextBlockParam{{Type: text, Text: systemPrompt}}, }校验逻辑与长响应用例类似聚合出的回答必须命中quantum、superposition、entanglement、qubit、algorithm中的至少两个术语以此确认系统提示词确实参与并约束了流式生成内容。四、运行测试的三种方式4.1 直接运行 go test原文档给出的命令均可直接使用在integration_test/anthropic/streaming目录下# 运行全部流式测试 go test -v . # 运行单个用例 go test -v -run TestBasicStreamingChatCompletion go test -v -run TestStreamingResponseWithTools go test -v -run TestLongResponseStreaming go test -v -run TestStreamingEventHandling go test -v -run TestStreamingWithSystemPrompt # 错误处理用例 go test -v -run TestStreamingErrorHandling4.2 通过 Makefile 运行integration_test/anthropic/Makefile 提供了专门的test-streaming目标它会先执行go mod tidy再运行整个 streaming 目录make test-streaming其他相关目标make test-all跑全部集成测试、make check-env检查环境变量是否就绪、make quick做快速冒烟验证仅 qa_simple。4.3 借助 -count1 规避缓存若怀疑 Go 测试缓存干扰了结果可在命令中加入-count1强制重新执行例如go test -count1 -v -run TestStreamingWithSystemPrompt .五、深入原理AxonHub 网关侧的流式聚合与转换客户端测试通过后值得理解网关另一侧发生了什么。AxonHub 把上游OpenAI、Gemini 等的流式 chunk 转换成 Anthropic SSE 事件流核心逻辑位于 llm/transformer/anthropic/aggregator.go 的AggregateStreamChunks它按事件类型分派处理message_start保存消息元数据与初始 usageinput_tokenscontent_block_start新建内容块对tool_use类型会把 Input 置空等待后续input_json_delta增量拼装 JSON 参数content_block_delta按index定位内容块把text_delta的文本增量追加到对应块aggregator.go把input_json_delta的PartialJSON拼到工具参数上message_delta提取stop_reason并把message_delta中的 usage 与message_start的 usage合并——保留 input_tokens、更新 output_tokens同时叠加CachedTokens、CacheReadInputTokens、CacheCreationInputTokens等缓存计数字段aggregator.gocontent_block_stop对工具参数做 JSON 合法性校验非法时用jsonrepair修复message_stop终态事件。而 llm/transformer/anthropic/inbound_stream.go 的TransformStream则实现了有状态的流式转换anthropicInboundStream结构体维护工具调用映射、thinking 内容缓冲、pending signature、文本引用citations等状态确保多轮、多内容块交织的事件流被正确地重新编排为 Anthropic 客户端可识别的顺序。这套机制在 llm/pipeline/streaming_integration_test.go 中有端到端验证例如TestPipeline_Streaming_Anthropic_to_OpenAIAnthropic 请求 → OpenAI 上游与TestPipeline_Streaming_OpenAI_to_AnthropicOpenAI 请求 → Anthropic 上游四条转换路径均有对应测试。这解释了为什么我们的客户端测试可以用统一代码同时对接真实 Anthropic 与 AxonHub 网关事件语义在转换前后保持一致。六、追踪与线程头部让流式请求可观测所有流式测试都通过 config.go 注入两个 AxonHub 专属请求头AH-Trace-Id代表一条用户消息及其触发的所有请求AH-Thread-Id代表一个完整对话会话用于把多次 Trace 关联起来。流式场景下这两个头部尤其重要一次流式调用在网关侧会被记录为一次 Request包含请求/响应体、耗时与 Token 使用量而工具调用的多轮流式请求通过同一 Thread ID 串联成完整会话便于在管理后台 Traces 页面还原全貌。头部名称可在config.yml的server.trace段自定义详见 docs/zh/guides/tracing.md。若希望测试不产生追踪可设置TEST_DISABLE_TRACEtrue关闭自动生成。七、注意事项与最佳实践原文档 Notes 部分总结了三条重要经验结合源码可进一步展开Anthropic 流式与 OpenAI 存在差异Anthropic 协议采用message_start / content_block_delta / message_delta / message_stop事件模型工具参数通过input_json_delta分片到达与 OpenAI 的choices[].delta结构不同。跨协议开发时不要假设两者事件结构一致——这正是 AxonHub 转换层存在的原因。流式工具调用支持交互式对话工具调用可以在流尚未结束时发出客户端需要边接收边判断并准备好在同一会话中发起二次流式请求见 3.3 的闭环模式。错误检查不可省略无论正常消费还是异常中断循环结束后都要调用stream.Err()并defer stream.Close()避免泄漏连接与遗漏错误。此外从实现细节可提炼的编码规范包括用message.Accumulate(event)维护官方状态而非手工拼装、用event.AsAny()做类型安全的断言、通过关键词/长度断言验证聚合完整性以及用t.Logf输出stop_reason、message ID 等调试信息方便失败复现。八、小结integration_test/anthropic/streaming是一套麻雀虽小、五脏俱全的流式验证套件6 个用例分别覆盖基本补全、长文本、工具调用、错误恢复、事件完整性与系统提示词配合 Makefile 的test-streaming目标即可一键执行。理解这些用例等于同时掌握了 Anthropic Go SDK 的流式编程范式与 AxonHub 网关的流式转换链路——无论你是要在网关兼容层上做回归验证还是要为自己的应用编写流式客户端这套测试模式都可以直接复用。赞分享AI 应用API网关LLM 网关后端前端【免费下载链接】axonhub⚡️ Open-source AI Gateway — Use any SDK to call 100 LLMs. Built-in failover, load balancing, cost control end-to-end tracing.项目地址https://gitcode.com/gh_mirrors/ax/axonhub点击查看免费下载相关推荐基于 AxonHub 的 Anthropic Go SDK 简单问答集成测试实战指南基于 AxonHub 的 Anthropic Go SDK 简单问答集成测试实战指南 本文以 AxonHub 仓库中 integration_test/anthAI 应用API网关LLM 网关后端前端HACS 仓库验证体系从规则骨架到 Action 集成的工作原理与自定义规则编写指南HACS 仓库验证体系从规则骨架到 Action 集成的工作原理与自定义规则编写指南 导读 HACSHome Assistant Community StoAI 应用API网关LLM 网关后端前端DataHub Form 实体实战指南结构化元数据采集与合规治理的模板化方案DataHub Form 实体实战指南结构化元数据采集与合规治理的模板化方案 Form表单是 DataHub 元数据模型中的核心实体它通过模板化问题AI 应用API网关LLM 网关后端前端上一篇Web-Dev-For-Beginners 精讲JavaScript 方法与函数 —— 从函数声明、参数与返回值到高阶回调的完整实战指南下一篇React Native Skia 阴影滤镜实战用 Shadow 组件实现投影、内阴影与拟物化设计创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考