AI Agent底层基石：单次模型请求的三种方式与选型实战

1. 为什么单次模型请求值得单独拎出来聊很多人一上来就搞多轮对话、工具调用、记忆管理结果连一次干净的模型请求都发不明白。我见过太多项目Agent 框架搭得花里胡哨最后排查问题发现是底层单次请求的参数传错了或者流式和非流式的处理逻辑混在一起导致整个链路雪崩。所以这次我想把“单次模型请求”这件事单独拆开讲透因为它是所有 AI Agent 的地基地基不平上面盖什么都会歪。所谓单次模型请求说白了就是你的程序向大模型服务端发一次调用拿回一次结果中间不涉及多轮上下文拼接、不涉及工具回调、不涉及状态保持。听起来简单但实际落地时有三种主流方式每种方式的适用场景、坑点和性能特征完全不一样。选错了方式轻则代码冗余难维护重则线上超时率飙升、费用翻倍。这篇文章适合谁看如果你正在写 Agent 的底层调用层或者你已经在用某些框架但想搞清楚它底层到底怎么发请求的再或者你单纯想优化一下自己项目里模型调用的稳定性和成本那这篇内容应该能帮你省下不少试错时间。我会把三种方式的原理、代码实现、参数计算、排查技巧全部摊开讲尽量做到你看完就能直接抄作业。2. 三种主流请求方式的全景拆解2.1 先搞清楚“单次请求”到底包含哪些环节在展开三种方式之前得先把一次完整的模型请求拆解清楚。不管用什么方式一次请求本质上都包含这几个阶段构造请求体、建立连接、发送数据、等待响应、解析响应、异常处理。区别在于不同方式对这些阶段的控制粒度不一样。构造请求体这一步核心是消息列表的组装。单次请求通常只包含一条 system 消息和一条 user 消息但即使是这么简单的结构也有讲究。比如 system 消息放什么内容、user 消息要不要做预处理、需不需要加 few-shot 示例这些都会直接影响返回质量。我个人的习惯是单次请求的 system 消息尽量精简把角色定义和输出格式约束写清楚就行不要把多轮对话才需要的东西塞进来。连接阶段涉及的是网络层面的东西。同步请求会阻塞当前线程直到拿到完整响应异步请求则把控制权交还出去流式请求会在连接建立后持续接收数据块。这三种模式对应到具体的请求方式上就衍生出了下面要讲的三种主流方案。2.2 三种方式的定义与核心差异第一种是原生 HTTP 同步请求。你直接用 HTTP 客户端库向模型服务的接口地址发一个 POST 请求body 里带上模型名称、消息列表、温度等参数然后等服务器返回完整的 JSON 响应。这种方式最原始但也最透明没有任何框架封装出问题了你能精确知道是哪一层的事。第二种是 SDK 异步请求。主流模型服务商都会提供官方 SDK里面封装了异步客户端。你用 async/await 语法发起调用底层走的是异步 IO不会阻塞事件循环。这种方式在 Python 的 asyncio 生态或者 Node.js 的 Promise 体系里非常常见适合需要并发发起多个单次请求的场景。第三种是流式请求。同样是发一次请求但服务端会以 Server-Sent Events 或者 chunked transfer 的方式逐块返回内容。你在客户端逐块接收、逐块处理用户能看到文字一个个蹦出来。这种方式严格来说也是单次请求只不过响应的接收方式从“一次性拿完”变成了“边拿边处理”。这三种方式没有绝对的优劣关键看你的场景。下面我逐个展开把每种方式的实现细节和踩坑经验都讲清楚。3. 原生 HTTP 同步请求最笨但最可控的方式3.1 适用场景与选型理由原生 HTTP 同步请求最适合什么场景我的经验是脚本类任务、离线批处理、调试阶段、以及对第三方依赖极度敏感的生产环境。为什么因为这种方式不依赖任何厂商 SDK你只需要一个通用的 HTTP 客户端库就行。哪怕服务商的 SDK 更新了、出 bug 了、或者你用的语言根本没有官方 SDK你照样能发请求。举个例子我之前做过一个数据标注清洗的小工具需要把几千条文本逐条发给模型做分类。这种场景下我不需要流式输出不需要并发因为要控制速率也不需要复杂的错误重试框架。直接用 HTTP 同步请求写个循环每条请求之间加个 sleep简单粗暴但极其稳定。跑了一晚上一条没丢。另一个典型场景是跨语言调用。比如你的主业务是 Java 写的但模型服务商的 SDK 只有 Python 和 Node.js 版本。这时候用 Java 的 HTTP 客户端直接发请求反而是最省事的方案。你只需要照着 API 文档拼 JSON 就行不需要为了调个模型再引入一个 Python 微服务。3.2 请求构造的关键参数与计算过程构造一个同步请求核心参数其实就那么几个但每个都值得细说。我拿一个典型的请求体来举例{ model: 某模型名称, messages: [ {role: system, content: 你是一个文本分类助手只输出类别标签。}, {role: user, content: 帮我判断这条评论的情感倾向这个产品用起来还不错。} ], temperature: 0.1, max_tokens: 64, top_p: 0.9 }temperature 设 0.1 是因为分类任务需要确定性输出温度越低模型越倾向于选概率最高的词。max_tokens 设 64 是因为分类结果只需要输出一个标签给太多 token 纯属浪费。top_p 设 0.9 是配合低温度做一点微调避免完全死板。这里有个计算过程值得展开max_tokens 到底设多少合适我的经验公式是预期输出长度的 1.5 倍再向上取整到 16 的倍数。比如你预期输出 20 个字那 20 乘以 1.5 等于 30向上取到 32。为什么要留 1.5 倍余量因为模型有时候会输出一些额外的解释性文字虽然你在 prompt 里约束了只输出标签但它偶尔还是会多嘴。留点余量避免被截断截断了的响应处理起来更麻烦。还有一个容易被忽略的参数是超时时间。同步请求必须设超时而且不能设太长。我一般设 30 秒超过 30 秒没返回就直接断开重试。为什么不设 60 秒因为模型服务端的排队时间加上推理时间正常情况下不会超过 30 秒。如果超过了说明服务端可能出问题了你等再久也没用不如早点断开走重试逻辑。3.3 完整代码实现与逐行说明下面是一个用 Python 的 requests 库实现的同步请求示例我逐段解释import requests import json import time def single_request_sync(prompt, system_msg你是一个助手。, max_retries3): url https://api.example.com/v1/chat/completions headers { Content-Type: application/json, Authorization: Bearer YOUR_API_KEY } payload { model: 某模型名称, messages: [ {role: system, content: system_msg}, {role: user, content: prompt} ], temperature: 0.1, max_tokens: 128 } for attempt in range(max_retries): try: resp requests.post(url, headersheaders, jsonpayload, timeout30) if resp.status_code 200: data resp.json() return data[choices][0][message][content] elif resp.status_code 429: wait 2 ** attempt time.sleep(wait) continue else: print(f请求失败状态码{resp.status_code}响应{resp.text}) return None except requests.exceptions.Timeout: print(f第 {attempt 1} 次请求超时) continue except requests.exceptions.RequestException as e: print(f网络异常{e}) continue return None这段代码有几个细节值得说。第一重试逻辑里对 429 状态码做了指数退避第一次等 1 秒第二次等 2 秒第三次等 4 秒。这是应对速率限制的标准做法比固定间隔重试要优雅得多。第二超时异常和网络异常分开捕获因为它们的处理策略可能不同。超时通常重试就行网络异常可能需要检查代理或者 DNS 配置。第三返回 None 而不是抛异常是因为调用方通常只需要知道“这次没拿到结果”不需要处理复杂的异常链。注意API Key 千万不要硬编码在代码里。我见过太多项目把 Key 写在源码里然后提交到了代码仓库结果被人扫到之后一夜之间跑掉几千块。正确做法是用环境变量或者密钥管理服务。3.4 实操心得与常见坑用同步请求最大的坑就是阻塞。如果你在一个 Web 服务的主线程里发同步请求那这个请求没返回之前整个线程就被卡住了。并发量一上来线程池瞬间打满服务直接不可用。所以同步请求只适合脚本、离线任务、或者你明确知道并发量很低的场景。第二个坑是连接复用。requests 库默认每次请求都新建连接如果你在循环里发几千次请求每次都要做 TCP 三次握手和 TLS 握手开销很大。解决办法是用 requests.Session()它会在底层维护一个连接池复用已经建立的连接。实测下来用 Session 之后同样数量的请求总耗时能降低 30% 到 40%。第三个坑是响应解析的健壮性。不要假设返回的 JSON 结构永远符合预期。我遇到过服务端在高峰期返回了一个缺少 choices 字段的响应结果代码直接 KeyError 崩了。后来我加了一层防御性解析先检查 choices 是否存在、是否为空、message 里是否有 content全部确认之后再取值。多写这几行判断能省掉很多半夜被叫起来修 bug 的时间。4. SDK 异步请求高并发场景下的效率利器4.1 异步请求的核心原理与优势异步请求的本质是把“等待网络响应”这段时间利用起来。同步模式下线程发出请求后就傻等什么也干不了。异步模式下事件循环在等待响应的同时可以去处理其他任务等响应回来了再切回来继续处理。对于单次模型请求这种 IO 密集型操作异步带来的吞吐量提升是非常明显的。我做过一个对比测试同样是用 100 个单次请求去处理 100 条文本同步方式串行执行耗时大约 180 秒异步方式并发执行耗时不到 8 秒。差了 20 多倍。当然这是在服务端不限制并发的前提下实际使用中还要考虑速率限制但异步的收益依然巨大。SDK 异步请求和原生 HTTP 异步请求的区别在于SDK 帮你封装了连接池管理、重试策略、参数校验、响应解析这些脏活累活。你只需要关注业务逻辑不用操心底层细节。代价是你得依赖这个 SDK如果 SDK 有 bug 或者更新不及时你就得等或者自己 fork 一份改。4.2 异步客户端的初始化与参数配置以 Python 生态中常见的异步 SDK 用法为例初始化通常长这样import asyncio from some_sdk import AsyncClient client AsyncClient( api_keyYOUR_API_KEY, base_urlhttps://api.example.com/v1, timeout30.0, max_retries3, connection_pool_size20 )这里有几个参数需要根据实际情况调整。connection_pool_size 是连接池大小设太小了并发上不去设太大了浪费资源。我的经验值是预期并发数乘以 1.5。比如你预计同时最多发 10 个请求那连接池设 15 就够了。timeout 和 max_retries 的含义和同步请求一样但异步环境下超时的处理更复杂一些因为可能涉及多个协程同时超时。还有一个容易忽略的参数是并发限制。很多 SDK 不直接提供并发限制参数你需要自己用信号量来控制。比如semaphore asyncio.Semaphore(10) async def limited_request(prompt): async with semaphore: return await client.chat.completions.create( model某模型名称, messages[{role: user, content: prompt}], temperature0.1, max_tokens128 )这个信号量保证了同时最多只有 10 个请求在飞超出的请求会排队等待。为什么要做这个限制因为服务端通常有并发上限你一次性发 100 个请求过去大部分会被拒绝或者排队反而降低了整体效率。用信号量控制在一个合理值让请求平稳地流过去总吞吐量反而更高。4.3 批量请求的编排与结果收集异步请求最典型的用法就是批量处理。假设你有 50 条文本需要分类可以这样编排async def batch_process(texts): tasks [limited_request(f分类这条文本{t}) for t in texts] results await asyncio.gather(*tasks, return_exceptionsTrue) final [] for i, r in enumerate(results): if isinstance(r, Exception): print(f第 {i} 条处理失败{r}) final.append(None) else: final.append(r.choices[0].message.content) return final这里用 gather 来并发执行所有任务return_exceptionsTrue 让单个任务的异常不会导致整个批次失败。结果收集的时候逐条判断是正常结果还是异常异常的就记 None后续可以单独重试。这种模式在处理大批量数据时非常实用不会因为一条失败就全军覆没。提示asyncio.gather 默认会等待所有任务完成如果你希望某个任务失败后立即取消其他任务可以用 asyncio.wait 配合 FIRST_EXCEPTION 模式。但大多数批量处理场景下我们更希望“能跑多少跑多少”所以 gather 加 return_exceptions 是更常见的选择。4.4 异步请求的常见问题与排查异步请求最常见的问题是“事件循环被阻塞”。什么意思就是你在异步函数里调用了同步的阻塞代码比如 time.sleep()、同步的数据库查询、同步的文件读写。这些操作会把整个事件循环卡住导致其他协程全部无法执行。解决办法是用 asyncio.sleep() 替代 time.sleep()用异步数据库驱动替代同步驱动用 aiofiles 替代普通文件操作。第二个问题是“协程泄漏”。你创建了协程但没有 await 它或者创建了任务但没有保存引用导致协程在后台悄悄运行然后报错而你完全不知道。排查方法是开启 asyncio 的调试模式它会在协程被垃圾回收时打印警告。具体做法是在程序入口加一行asyncio.get_event_loop().set_debug(True)。第三个问题是“连接池耗尽”。当并发量超过连接池大小时新的请求会等待可用连接。如果等待时间超过了超时设置就会抛异常。排查方法是监控连接池的使用率如果长期处于满载状态就需要调大连接池或者降低并发量。我一般会在日志里记录每次请求的等待时间如果发现等待时间明显增长就是连接池不够用的信号。5. 流式请求用户体验优先的交互方式5.1 流式请求的工作机制流式请求和前面两种的本质区别在于响应的接收方式。同步和异步都是“发一次请求等一个完整响应”流式是“发一次请求持续接收数据块”。服务端每生成一个 token 或者一小段文本就立即推送给客户端客户端收到后可以立刻展示给用户。这种机制在交互式场景下体验极好。用户问一个问题模型开始回答文字一个个蹦出来用户感觉系统反应很快。如果等完整响应再展示用户可能要盯着 loading 转圈好几秒体验差很多。所以但凡是对用户体验有要求的场景流式几乎是必选项。但流式也有代价。首先你不能像同步请求那样直接拿到完整的 JSON 做解析得自己处理数据块的拼接。其次流式请求的错误处理更复杂因为可能在传输中途断开。再次流式请求不太适合批量处理因为它的优势在于“边生成边展示”批量场景下没有展示需求用流式反而增加了处理复杂度。5.2 流式数据的接收与拼接实现流式请求的响应通常是一系列以特定格式分隔的数据块。以常见的 SSE 格式为例每个数据块长这样data: {choices:[{delta:{content:你}}]} data: {choices:[{delta:{content:好}}]} data: [DONE]你需要逐行读取遇到data:开头就解析 JSON取出 delta 里的 content 拼接到结果字符串上遇到[DONE]就结束。下面是一个简化的实现import requests import json def stream_request(prompt): url https://api.example.com/v1/chat/completions headers { Content-Type: application/json, Authorization: Bearer YOUR_API_KEY } payload { model: 某模型名称, messages: [{role: user, content: prompt}], stream: True, temperature: 0.7, max_tokens: 512 } full_content with requests.post(url, headersheaders, jsonpayload, streamTrue, timeout60) as resp: for line in resp.iter_lines(): if not line: continue line_str line.decode(utf-8) if line_str.startswith(data: ): data_str line_str[6:] if data_str.strip() [DONE]: break try: chunk json.loads(data_str) delta chunk[choices][0][delta] if content in delta: content_piece delta[content] full_content content_piece print(content_piece, end, flushTrue) except (json.JSONDecodeError, KeyError, IndexError): continue return full_content这段代码的关键点在于streamTrue 告诉 requests 不要一次性读取全部响应而是以流的方式逐行读取。iter_lines 会按行迭代每次拿到一行数据。解析的时候要处理各种异常情况因为流式数据块偶尔会出现格式不完整的情况比如网络抖动导致一行被截断。用 try-except 包住解析逻辑遇到坏数据就跳过不要让它中断整个流。5.3 流式场景下的超时与中断处理流式请求的超时设置和同步请求不一样。同步请求的超时是“从发出请求到收到完整响应”的总时间流式请求的超时通常分为两部分连接超时和读取超时。连接超时是指建立连接的时间读取超时是指两次数据块之间的最大间隔。我一般把连接超时设 10 秒读取超时设 30 秒。读取超时不能设太短因为模型在生成某些复杂内容时两个 token 之间的间隔可能会比较长。如果设太短正常生成过程中就被断开了用户看到一半内容没了体验很差。中断处理也是流式场景必须考虑的。用户可能在生成过程中点击“停止”按钮这时候你需要主动关闭连接。在 Python 里可以用 resp.close() 来关闭流式响应。关闭之后已经接收到的内容要保留不能丢掉。我见过一些实现用户点停止之后已经生成的内容也消失了这就是没有做好中断时的状态保存。还有一个坑是流式请求的重试。同步请求重试很简单重新发一次就行。流式请求如果中途断了重试的话是从头开始还是从断点继续大多数模型服务不支持断点续传所以只能从头开始。但从头开始意味着用户已经看到的内容会重复体验不好。我的做法是如果流式请求在生成早期就断了比如前 10 个 token 内直接重试如果已经生成了一大半才断就不重试了把已生成的内容展示给用户并提示“生成中断”。5.4 流式与非流式的性能对比实测我做过一组实测对比流式和非流式在相同任务下的表现。任务是用模型生成一段 300 字左右的文案分别用流式和非流式各跑 20 次记录首字节时间和总完成时间。指标非流式流式首字节时间2.8 秒0.6 秒总完成时间3.2 秒3.4 秒用户感知等待3.2 秒0.6 秒网络传输量略少略多从数据可以看出流式的总完成时间其实比非流式略长因为数据分块传输有额外的协议开销。但首字节时间从 2.8 秒降到了 0.6 秒用户感知的等待时间大幅缩短。这就是流式的核心价值不是让生成更快而是让用户感觉更快。注意流式请求的 max_tokens 设置要比非流式更谨慎。因为流式是边生成边展示如果 max_tokens 设得太大用户可能要等很久才能看到完整内容。我一般会根据预期输出长度来设留 20% 余量就够了不要像非流式那样留 50% 余量。6. 三种方式的选型决策与混合使用策略6.1 选型决策表与判断逻辑到底该用哪种方式我整理了一个决策表你可以对照自己的场景来选场景特征推荐方式理由脚本任务、离线批处理原生 HTTP 同步简单可控不依赖 SDK高并发批量处理SDK 异步吞吐量高SDK 封装完善交互式对话、实时展示流式首字节时间短用户体验好跨语言调用原生 HTTP 同步不受 SDK 语言限制需要精细控制请求细节原生 HTTP完全透明无框架黑盒快速原型开发SDK 异步代码量少开发效率高判断逻辑其实就三条第一有没有实时展示需求有就选流式。第二需不需要高并发需要就选异步。第三对第三方依赖敏不敏感敏感就选原生 HTTP。三条都不满足那就随便选哪个顺手用哪个。6.2 混合使用的最佳实践实际项目中这三种方式往往不是互斥的而是混合使用的。我拿一个典型的 Agent 系统举例用户在前端输入问题后端需要先做意图识别单次请求用同步或异步都行然后根据意图决定是直接回答还是调用工具。如果是直接回答用流式返回给用户如果是调用工具工具调用本身可能是异步批量请求等工具结果回来后再用流式生成最终回答。这种混合模式下关键是要统一请求层的接口。我的做法是定义一个统一的请求函数内部根据参数决定走同步、异步还是流式对外暴露一致的调用方式。这样上层业务逻辑不需要关心底层用的是哪种方式切换起来也方便。还有一个实践是“降级策略”。比如正常情况下用流式但如果流式连接建立失败自动降级到非流式同步请求。虽然用户体验差一点但至少能拿到结果。这种降级逻辑在线上环境非常有必要因为流式请求对网络稳定性的要求比非流式更高。6.3 成本与性能的平衡计算选型的时候不能只看技术还得算成本。三种方式在 token 消耗上其实是一样的因为计费是按输入输出 token 数来的和传输方式无关。但在重试场景下流式的成本可能更高因为流式中断后重试是从头开始已经生成的 token 白费了。非流式中断后重试之前那次请求可能根本没产生输出不消耗输出 token。我算过一笔账假设流式中断率是 5%每次中断平均浪费 200 个输出 token一天 10000 次请求那每天浪费的 token 是 10000 乘以 5% 乘以 200 等于 100000 个 token。按某模型的价格算一天多花几块钱一个月多花一百多。不算多但如果你的中断率更高或者请求量更大这个数字就会很可观。所以我的建议是流式请求一定要做好中断检测和快速失败。如果连接建立后 5 秒内没有收到任何数据块直接断开重试不要傻等。这样能把浪费的 token 降到最低。7. 常见问题排查与避坑经验实录7.1 请求超时与连接失败的排查思路超时和连接失败是最常见的问题排查思路可以按这个顺序来先确认网络连通性用 curl 或者 telnet 测试接口地址是否可达再确认 API Key 是否有效有时候 Key 过期了返回的错误信息不明显然后检查请求体格式特别是 JSON 的字段名和数据类型最后看服务端状态如果服务端在维护或者过载你这边怎么调都没用。我遇到过一个很隐蔽的问题请求偶尔超时但大部分时候正常。排查了半天发现是 DNS 解析不稳定有时候解析到了慢的节点。解决办法是在 HTTP 客户端里配置 DNS 缓存或者直接用 IP 地址加 Host 头的方式绕过 DNS。这个问题在容器化环境里特别常见因为容器的 DNS 配置有时候不太靠谱。7.2 返回结果异常的调试方法返回结果异常包括返回空内容、返回乱码、返回格式不符合预期、返回内容被截断。空内容通常是 max_tokens 设太小模型还没来得及输出就达到了上限。乱码一般是编码问题确认请求和响应都用 UTF-8。格式不符合预期可能是 prompt 约束不够强或者 temperature 太高导致模型自由发挥。截断就是 max_tokens 不够加大就行。调试的时候我习惯把完整的请求体和响应体都打到日志里但要注意脱敏别把 API Key 和用户隐私数据打进去。日志级别用 DEBUG只在排查问题时开启平时关掉避免日志膨胀。7.3 速率限制与重试策略的实战配置速率限制是模型服务商保护自己系统的机制你没法绕过只能适应。常见的限制维度有每分钟请求数、每分钟 token 数、并发连接数。应对策略就是指数退避加队列。我的标准配置是最大重试 3 次第一次等待 1 秒第二次 2 秒第三次 4 秒。如果三次都失败就把请求放入延迟队列过 30 秒再试。延迟队列里的请求不再重试只尝试一次失败就记录到死信队列人工介入处理。提示不要用固定间隔重试比如每次都等 1 秒。因为如果服务端是因为过载才限流你固定间隔重试只会加重它的负担。指数退避给服务端喘息的时间成功率更高。7.4 常见问题速查表问题现象可能原因排查方法解决方案请求超时网络不通、服务端过载、超时设太短用 curl 测试连通性查看服务端状态页调整超时时间增加重试返回 401API Key 无效或过期检查 Key 是否正确、是否过期更换有效 Key返回 429触发速率限制查看响应头中的限制信息降低并发增加退避等待返回空内容max_tokens 太小、prompt 有问题检查 max_tokens 设置简化 prompt增大 max_tokens优化 prompt流式中断网络抖动、服务端主动断开查看中断时的错误信息实现断线重连或降级到非流式响应乱码编码不一致检查请求和响应的 Content-Type统一使用 UTF-8连接池耗尽并发量超过连接池大小监控连接池使用率调大连接池或降低并发这张表里的每一行都是我实际踩过的坑。特别是流式中断那个我一开始没做降级处理结果网络一抖动用户就看到半截内容投诉了好几次。后来加了降级逻辑流式失败自动切非流式虽然慢一点但至少完整。7.5 我个人的避坑清单最后分享几个我总结的避坑要点都是血泪教训第一永远不要相信网络是稳定的。任何一次请求都可能失败所以重试逻辑是必须的不是可选的。第二永远不要假设响应格式是固定的。服务端可能升级、可能出 bug、可能返回意料之外的字段。防御性解析能帮你省掉很多深夜修 bug 的时间。第三永远不要在循环里发同步请求而不加延迟。你以为你在快速处理实际上你在快速触发速率限制。第四流式请求一定要设读取超时。我见过有人没设读取超时结果服务端挂了但连接没断客户端傻等了一个小时。第五日志要打全但要注意脱敏。请求 ID、耗时、状态码、token 用量这些都要记但 API Key、用户输入内容这些敏感信息不要记。第六上线前一定要做压力测试。你以为你的代码能扛住 100 并发实际可能 10 并发就崩了。提前测出来比线上出问题强。这些经验看起来简单但每一条背后都有真实的事故。希望你看完之后能少走点弯路。单次模型请求虽然只是 Agent 系统里的一小环但这一环的稳定性直接决定了整个系统的可用性。把这一环做扎实了上面的多轮对话、工具调用、记忆管理才有意义。