1. 为什么我会攒出这套 LLM 工具链先交代一下背景。在做 LLM 应用开发的这大半年里我几乎每天都要跟各种模型 API、提示词、上下文窗口、Token 统计打交道。最开始还能靠 Postman 和几个零散脚本硬撑后来随着项目变多、模型切换频繁慢慢发现自己在重复劳动上耗掉的时间比真正写业务逻辑的时间还多。于是我开始把高频操作沉淀成一个个小工具。这些工具都不复杂单个拉出来也许只有几十行代码但组合起来之后日常工作效率提升非常明显。今天这篇文章想分享的就是这套我自认为“离不开”的 LLM 工具集包括设计思路、完整代码、使用方法以及我在踩坑过程中总结出来的一些经验和教训。如果你正处于下面这几个阶段这篇文章应该会比较适合你刚入门 LLM 应用开发习惯用网页端聊天但想开始接 API 做自动化。已经在写调用大模型的代码但觉得每次都要重新拼 Prompt、管理上下文很烦。想要搭建一套本地化的“模型调用 知识库问答 Agent 编排”小工具箱。对 Token 消耗不敏感但想学习如何从工程角度系统性管理 LLM 调用。文章不会只贴代码我会把每个工具解决什么问题、为什么这样设计、换成其他方案时有什么取舍都讲清楚。你可以直接复制代码到本地跑起来也可以按自己需要改造成适合自己的版本。2. 核心概念与工具全景2.1 LLM 应用开发的基本组成在展开代码之前先理清一个概念所谓的 LLM 应用通常不是“往 API 里丢一句话然后得到回复”这么简单。一个稍微成熟一点的项目至少会涉及这几个模块模型接入层负责对接 OpenAI、Anthropic、国内厂商或本地部署模型的 API统一鉴权、超时、重试和错误处理。会话管理层负责维护多轮对话的上下文避免聊天记录越来越长导致 Token 超限。提示词管理把常用 Prompt 抽成模板避免每次在代码里拼接字符串。工具调用层当模型需要查数据库、调接口、搜索网页时通过 Function Calling 机制把外部能力注册给模型。知识检索层也就是 RAG检索增强生成先从本地文档中检索出相关内容再交给模型总结回答解决“模型不知道你本地资料”的问题。资源观测层记录每次请求的耗时、Token 数、费用方便做成本分析和限流。下面要介绍的工具本质上就是把这些模块落成一个个可以直接运行的小项目。2.2 这套工具集包含什么为了不让工具变成一个庞然大物我把它们拆成了相对独立的模块。每个模块都可以单独运行也可以通过一个统一的入口去调用工具作用适合人群多模型 CLI 聊天终端在终端里随时切换模型对话支持流式输出和会话保存日常调试、快速验证 PromptLLM API 统一网关统一汇聚多个模型供应商的 API提供路由、重试、限流后端服务集成、团队共享 Key本地 RAG 知识库问答把本地文档切分、向量化检索后交给大模型回答问题个人知识库、内部资料问答Agent 函数调用编排器注册自定义工具让模型自动决定调用哪个函数完成任务自动化任务、信息查询类应用这四个工具覆盖了我日常开发中 90% 以上的场景。下面我会按顺序讲解先不急着引入太重的框架尽量用最直观的方式把原理讲明白。2.3 为什么不做成一个 All-in-One 应用你可能会问为什么不把这四个工具合并成一个系统我的理由是职责分离更清晰一个脚本坏掉不影响其他工具。可以单独替换某一部分比如今天想换一个向量库不用动聊天终端。方便通过命令行快速验证而不是每次都要启动一个 Web 前端。当然如果你已经有一个成熟项目也可以把这些工具的代码直接作为模块集成进去。本文的重点是讲解每块能力的实现方式而不是推荐某种固定架构。3. 环境准备与项目结构3.1 运行环境说明本文中的代码以 Python 为例因为 Python 在处理 API 调用、文本切分、数据存储等任务时比较方便生态也最齐全。操作系统Windows 10/11、macOS、Linux 均可。Python 版本建议 3.10 及以上代码中使用了较新的类型注解语法。包管理工具推荐使用venv或conda创建独立环境。模型 API示例代码以兼容 OpenAI SDK 的接口为例如果你使用的是本地模型或其他云厂商只需调整base_url和鉴权参数。需要特别说明的是不同框架和 SDK 的版本更新速度很快本文不会写死某个具体版本号。你在复现时建议使用当前较新的稳定版本。如果遇到接口变动以官方文档为准。3.2 创建项目结构我们可以创建一个名为llm-toolkit的目录整体结构如下llm-toolkit/ ├── .env.example # 环境变量模板 ├── requirements.txt # 依赖清单 ├── chat_cli.py # 工具一CLI 聊天终端 ├── api_gateway.py # 工具二API 统一网关 ├── rag_qa.py # 工具三RAG 知识库问答 ├── agent_orchestrator.py # 工具四Agent 函数调用编排器 ├── llm_client.py # 公共的 LLM 客户端封装 ├── config.py # 公共配置读取 └── docs/ # 存放本地知识库文档先创建虚拟环境并安装基础依赖python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate pip install openai python-dotenv fastapi uvicorn requests如果你的场景还需要做向量检索可以再安装对应的向量库客户端。为了示例简单我后面会先用一种基于本地 JSON 文件的轻量方案替换成正式向量库也很容易。3.3 统一配置管理LLM 工具最怕把 API Key 写死在代码里。这里我们统一通过环境变量管理配置新建.env文件# .env OPENAI_API_KEYsk-xxxx OPENAI_BASE_URLhttps://api.openai.com/v1 OPENAI_MODELgpt-4o-mini # 如果使用本地模型则可以改成 # OPENAI_BASE_URLhttp://localhost:11434/v1 # OPENAI_MODELllama3.1然后在config.py中统一读取import os from dotenv import load_dotenv load_dotenv() def get_config(): return { api_key: os.getenv(OPENAI_API_KEY), base_url: os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1), model: os.getenv(OPENAI_MODEL, gpt-4o-mini), }这样做的目的是让代码与具体密钥、模型名解耦。你可以在本地用 OpenAI在公司内网切换到私有化部署模型只需要改环境变量不需要动业务代码。4. 工具一多模型 CLI 聊天终端4.1 设计思路这个工具解决的核心痛点是当你需要快速验证一个 Prompt、比较不同模型的表现时打开网页端聊天显然太低效。我更希望直接在终端里输入一句话马上得到回复而且能一键切换模型、保存会话记录。设计上需要满足四个点支持流式输出不用等到完整回复生成才能看到内容。支持多轮对话自动维护上下文列表。支持通过命令切换模型比如输入/model gpt-4o。支持把会话保存为 JSON 文件方便事后复盘 Prompt 效果。4.2 核心代码实现先写一个公共的llm_client.py负责统一创建 OpenAI 客户端# llm_client.py from openai import OpenAI from config import get_config _config get_config() def get_client(): return OpenAI( api_key_config[api_key], base_url_config[base_url], ) def get_model(): return _config[model]接着是chat_cli.py实现一个简单的交互式聊天终端# chat_cli.py import json import datetime from llm_client import get_client, get_model client get_client() current_model get_model() history [] def chat(user_input: str): 发送一轮对话以流式方式打印回复 history.append({role: user, content: user_input}) try: stream client.chat.completions.create( modelcurrent_model, messageshistory, streamTrue, # 开启流式输出 ) except Exception as e: print(f[请求失败] {e}) history.pop() # 请求失败时移除这次的 user 输入 return print(Assistant: , end, flushTrue) full_reply for chunk in stream: if chunk.choices[0].delta.content: content chunk.choices[0].delta.content print(content, end, flushTrue) full_reply content print() # 换行 history.append({role: assistant, content: full_reply}) def save_session(filename: str): 保存当前会话到本地文件 payload { time: datetime.datetime.now().isoformat(), model: current_model, history: history, } with open(filename, w, encodingutf-8) as f: json.dump(payload, f, ensure_asciiFalse, indent2) print(f会话已保存到 {filename}) def main(): global current_model print(LLM Chat CLI - 输入 /help 查看命令) while True: try: user_input input(You: ).strip() except (EOFError, KeyboardInterrupt): print(\n再见) break if not user_input: continue if user_input.startswith(/): parts user_input.split(maxsplit1) cmd parts[0].lower() if cmd /help: print(/model 模型名 切换模型) print(/save 文件名 保存会话) print(/clear 清空会话) print(/exit 退出) elif cmd /model and len(parts) 1: current_model parts[1] print(f已切换到模型: {current_model}) elif cmd /save and len(parts) 1: save_session(parts[1]) elif cmd /clear: history.clear() print(会话已清空) elif cmd /exit: print(再见) break else: print(未知命令输入 /help 查看帮助) else: chat(user_input) if __name__ __main__: main()4.3 运行演示启动工具python chat_cli.py运行效果类似LLM Chat CLI - 输入 /help 查看命令 You: 用一句话介绍什么是大语言模型 Assistant: 大语言模型是一种基于海量文本训练的人工智能模型能够理解和生成自然语言文本。 You: /model gpt-4o 已切换到模型: gpt-4o You: 现在用更专业的语气再介绍一次 Assistant: 大语言模型Large Language Model是一种通过自监督学习在大型语料库上预训练的深度神经网络模型其核心能力在于对自然语言进行概率建模与生成。这里有两个实现细节值得展开说说。第一是流式输出。如果不开启streamTrue很多模型需要等十几秒才能一次性返回结果用户体验很差。开启后我们可以边生成边打印体感上会快很多。第二是失败回滚。在chat()函数中如果请求抛出异常我们会把本轮追加的 user 消息从history中弹出去。这样做是避免把一次失败请求也算进上下文导致后续对话携带了无效信息。4.4 这个工具可以怎么扩展加入多会话管理启动时选择继续某个历史会话而不是每次都是从空上下文开始。加入 Token 统计每次请求结束后打印消耗的 Token 数方便估算成本。加入 Prompt 模板系统把常用 Prompt 放到一个prompts目录通过/prompt 模板名快速插入。5. 工具二LLM API 统一网关5.1 为什么需要网关当我开始写多个 LLM 工具后很快遇到一个问题每个工具都要自己处理 API Key、重试逻辑、超时设置、限流。如果三个服务同时调用同一个模型还要担心会不会因为并发过高被限流。API 网关的作用就是把这些公共逻辑收敛到一个服务里。所有内部服务只需要请求网关由网关去和真实的大模型 API 交互。这样一来业务方不需要持有真实 API Key降低了泄漏风险。可以在网关层做统一的请求日志、Token 统计和费用汇总。可以针对不同供应商做路由比如某些模型走 OpenAI某些模型走本地部署服务。5.2 网关的核心能力我实现的网关并不复杂核心是下面这几块接收客户端请求参数格式保持和 OpenAI API 兼容。根据配置的供应商路由规则转发到真实模型服务。统一处理超时、重试、限流。返回标准格式的响应同时在响应头里附带本次请求的 Token 消耗。5.3 网关代码实现这里使用 FastAPI 来搭建代码非常简洁# api_gateway.py import time import uuid from fastapi import FastAPI, HTTPException, Request from pydantic import BaseModel from llm_client import get_client app FastAPI(titleLLM API Gateway) # 简单的请求日志正式项目可以改为写入数据库或日志系统 request_log [] class ChatRequest(BaseModel): model: str messages: list temperature: float 0.7 stream: bool False def call_real_api(req: ChatRequest): 转发到真实模型服务 client get_client() start time.time() try: resp client.chat.completions.create( modelreq.model, messagesreq.messages, temperaturereq.temperature, streamreq.stream, ) cost_time round(time.time() - start, 3) return resp, cost_time except Exception as e: raise HTTPException(status_code502, detailf上游模型调用失败: {e}) app.post(/v1/chat/completions) async def chat_completions(req: ChatRequest, raw_request: Request): # 如果开启了流式这里简化处理直接透传上游结果 if req.stream: resp, cost_time call_real_api(req) # FastAPI 对流式响应需要额外处理示例中先返回普通 JSON return {stream: True, cost_time: cost_time, data: 流式模式需要对接 SSE} resp, cost_time call_real_api(req) content resp.choices[0].message.content # 记录日志 log_entry { request_id: str(uuid.uuid4()), model: req.model, prompt_tokens: resp.usage.prompt_tokens if resp.usage else 0, completion_tokens: resp.usage.completion_tokens if resp.usage else 0, cost_time: cost_time, client_ip: raw_request.client.host if raw_request.client else unknown, } request_log.append(log_entry) return { id: log_entry[request_id], model: req.model, choices: [{message: {role: assistant, content: content}}], usage: { prompt_tokens: log_entry[prompt_tokens], completion_tokens: log_entry[completion_tokens], }, } app.get(/logs) def get_logs(): 查看最近的请求日志 return request_log[-50:] if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)启动网关python api_gateway.py然后用 curl 测试curl -X POST http://localhost:8000/v1/chat/completions \ -H Content-Type: application/json \ -d { model: gpt-4o-mini, messages: [{role: user, content: 你好介绍一下你自己}] }返回结果中会包含usage字段方便我们统计 Token 消耗。5.4 网关的进阶方向上面这个版本只是最小可用实现。在实际项目中你大概率还需要补充Key 管理把多个 API Key 存成池轮询或按权重分配避免单个 Key 触发限流。限流策略基于令牌桶或滑动窗口限制每个业务方每分钟的请求次数。缓存对相同请求做语义级或文本级缓存减少重复调用费用。SSE 流式支持当客户端需要流式响应时网关也要能透传 Server-Sent Events。其中流式透传是很多初学网关的人容易卡住的地方。核心思路是使用StreamingResponse在拿到上游的流对象后逐块转发给客户端from fastapi.responses import StreamingResponse app.post(/v1/chat/completions/stream) async def chat_completions_stream(req: ChatRequest): client get_client() stream client.chat.completions.create( modelreq.model, messagesreq.messages, streamTrue, ) def generate(): for chunk in stream: if chunk.choices[0].delta.content: yield fdata: {chunk.choices[0].delta.content}\n\n return StreamingResponse(generate(), media_typetext/event-stream)这个模式下客户端可以像调用真实 OpenAI API 一样用官方的 SDK 配合streamTrue来消费网关返回的流式数据。6. 工具三RAG 本地知识库问答6.1 RAG 是什么RAG 全称是 Retrieval-Augmented Generation翻译过来就是“检索增强生成”。它的核心思想是不直接让大模型凭记忆回答问题而是先从用户指定的文档中检索出相关内容把检索结果组装进 Prompt再让模型基于这些材料回答。这样做的好处很明显回答内容以你提供的资料为依据减少凭空捏造。可以随时更新知识库不用重新训练模型。回答中可以引用来源方便人工核对。我日常用得最多的场景是把团队的技术文档、项目周报、产品说明书放到本地然后通过 RAG 工具做问答。6.2 文档切分与向量化RAG 的流程可以拆成“入库”和“查询”两个阶段。入库阶段读取本地文档。按段落或固定块大小切分成片段。每个片段通过 Embedding 模型转成向量存入向量数据库。查询阶段把用户问题转成向量。在向量库中做相似度检索找到最相关的几个片段。把片段和问题一起发给大模型生成最终回答。这里为了演示方便我用一个 JSON 文件充当向量库避免引入过重的依赖。实际项目中可以替换为 Chroma、FAISS、Milvus 或 Qdrant。6.3 代码实现首先实现一个文档入库函数把文档切块并计算 Embedding# rag_qa.py import json import os from llm_client import get_client client get_client() # 简单的向量缓存文件 VECTOR_FILE local_vectors.json def read_docs_from_dir(doc_dir: str): 读取目录下所有 txt 文档 docs [] for filename in os.listdir(doc_dir): if filename.endswith(.txt): filepath os.path.join(doc_dir, filename) with open(filepath, r, encodingutf-8) as f: content f.read() docs.append({source: filename, content: content}) return docs def chunk_text(text: str, chunk_size: int 300): 将文本切分成固定长度的片段注意保留段落边界 paragraphs [p.strip() for p in text.split(\n) if p.strip()] chunks [] current for p in paragraphs: if len(current) len(p) chunk_size: current p \n else: if current: chunks.append(current.strip()) current p \n if current: chunks.append(current.strip()) return chunks def get_embedding(text: str): 调用 Embedding 接口生成向量 resp client.embeddings.create( modeltext-embedding-3-small, inputtext, ) return resp.data[0].embedding def build_index(doc_dir: str): 入库把所有文档切块并保存向量 docs read_docs_from_dir(doc_dir) records [] for doc in docs: chunks chunk_text(doc[content]) for chunk in chunks: vector get_embedding(chunk) records.append({ source: doc[source], content: chunk, vector: vector, }) with open(VECTOR_FILE, w, encodingutf-8) as f: json.dump(records, f, ensure_asciiFalse) print(f文档入库完成共 {len(records)} 个片段) def search(query: str, top_k: int 3): 查询根据问题向量做余弦相似度检索 query_vec get_embedding(query) with open(VECTOR_FILE, r, encodingutf-8) as f: records json.load(f) scored [] for rec in records: score cosine_similarity(query_vec, rec[vector]) scored.append({score: score, content: rec[content], source: rec[source]}) scored.sort(keylambda x: x[score], reverseTrue) return scored[:top_k] def cosine_similarity(vec1, vec2): 计算两个向量的余弦相似度 dot sum(a * b for a, b in zip(vec1, vec2)) norm1 sum(a * a for a in vec1) ** 0.5 norm2 sum(a * b for a, b in zip(vec2, vec2)) ** 0.5 if norm1 0 or norm2 0: return 0 return dot / (norm1 * norm2) def rag_answer(question: str): RAG 问答主流程 results search(question) if not results: return 知识库中没有找到相关内容 context \n\n.join([ f【来源{r[source]}】\n{r[content]} for r in results ]) prompt f请基于以下资料回答问题。如果资料中没有相关内容请直接说明不知道不要编造。 资料 {context} 问题{question} resp client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: 你是一个严谨的知识库问答助手回答时尽量结合给定资料。}, {role: user, content: prompt}, ], ) return resp.choices[0].message.content运行入库python rag_qa.py # 你需要在脚本入口调用 build_index(docs)运行问答# 在脚本中调用 print(rag_answer(这篇文档里提到的最佳实践是什么))回答中会带上我们检索到的资料片段作为上下文模型会根据这些片段组织答案。你可以看到这个实现虽然简单但已经具备了 RAG 的核心闭环。6.4 RAG 落地的关键坑点在实际项目中RAG 的效果往往取决于三个细节第一切分策略。固定chunk_size的做法虽然简单但可能会切断一个完整的概念。更好的做法是结合标题结构、段落语义进行切分或者使用递归字符分割器。第二Embedding 模型选择。不同语言的 Embedding 模型差异很大中文场景建议选用对中文支持好的模型并且要评估平均相似度分数不要盲目用默认模型。第三检索后处理。不要把检索到的所有内容都塞进 Prompt一是浪费时间 Token二是无关内容会干扰模型。合理的做法是先做相关性过滤再按分数降序排列。7. 工具四Agent 函数调用编排器7.1 函数调用Function Calling基础如果说 RAG 解决的是“让模型知道更多信息”那么 Agent 解决的是“让模型能执行动作”。大模型本身只能生成文本不能真正查询数据库、调用接口、写入文件。Function Calling 机制提供了一个标准化的办法你定义好一组函数告诉模型有哪些函数可用模型在回答时如果发现需要外部能力就会返回一个“我希望调用某个函数参数是什么”的结构化结果。这样模型就从一个“聊天机器人”升级成了“能主动分发任务的调度器”。7.2 工具注册与调度我这里的实现思路是定义一组工具函数每个函数都有一个名字、描述、参数 schema。把工具列表传给模型。模型返回工具调用请求时本地执行真正的函数。把函数执行结果追加到消息历史中再让模型继续回答。7.3 完整示例我们先定义两个简单的工具一个是查询本地服务器时间一个是模拟查询天气# agent_orchestrator.py import json import datetime from llm_client import get_client client get_client() def get_current_time(): 获取当前服务器时间 return datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) def get_weather(city: str): 模拟查询城市天气 weather_map { 北京: 晴25°C, 上海: 多云28°C, 深圳: 阵雨30°C, } return weather_map.get(city, f{city} 天气数据暂未收录) TOOLS [ { type: function, function: { name: get_current_time, description: 获取当前服务器时间, parameters: { type: object, properties: {}, }, }, }, { type: function, function: { name: get_weather, description: 查询指定城市的天气, parameters: { type: object, properties: { city: { type: string, description: 城市名称例如北京, } }, required: [city], }, }, }, ] TOOL_MAP { get_current_time: get_current_time, get_weather: get_weather, } def run_agent(user_input: str): 运行一个简单的 Agent 循环 messages [{role: user, content: user_input}] # 第一次调用带工具列表 resp client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolsTOOLS, ) msg resp.choices[0].message # 模型要求调用工具 if msg.tool_calls: messages.append(msg) # 保存带 tool_calls 的 assistant 消息 for tool_call in msg.tool_calls: fn_name tool_call.function.name args json.loads(tool_call.function.arguments) fn TOOL_MAP.get(fn_name) if fn: result fn(**args) else: result f未找到工具: {fn_name} # 把工具执行结果追加为 tool 消息 messages.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps(result, ensure_asciiFalse), }) # 第二次调用让模型基于工具结果生成回答 second_resp client.chat.completions.create( modelgpt-4o-mini, messagesmessages, ) return second_resp.choices[0].message.content # 模型直接回答没有调用工具 return msg.content if __name__ __main__: print(run_agent(现在几点了)) print(run_agent(北京今天天气怎么样)) print(run_agent(帮我分别查一下上海和深圳的天气))运行后模型会自行决定调用哪些工具。比如第一个问题会触发get_current_time第二个问题会触发get_weather第三个问题甚至可能连续调用两次get_weather然后把结果汇总成一句回答。7.4 Agent 编排的复杂度控制函数调用本身不难难在处理复杂情况时的状态管理。以我自己的经验有几个问题比较常见工具调用循环深度模型可能调用工具之后还想再调用另一个工具形成多轮工具调用。代码里需要加一个最大循环次数限制否则可能死循环。并行工具调用新版模型支持一次返回多个tool_calls处理时要遍历全部调用而不是只处理第一个。错误处理工具执行时可能抛异常比如请求外部接口超时。这时应该把错误信息作为 tool 消息返回给模型让模型学会“承认失败”或换个方式尝试。如果你要做一个生产级 Agent建议引入状态机或工作流引擎来管理任务状态而不是只用简单的 while 循环。这一点在下一节的最佳实践中会再提到。8. 常见问题与排查思路在复现或使用上述代码时你很可能遇到下面这些报错。整理成表格方便快速定位。问题现象常见原因解决思路OpenAI客户端报认证失败API Key 未配置或失效检查.env文件确认OPENAI_API_KEY已正确加载使用print(get_config())调试模型返回内容经常被截断上下文长度超过模型最大限制开启对话前先截断或压缩历史消息使用max_tokens控制回复长度调用 RAG 时检索结果不相关文档切分太碎或 Embedding 模型不适合当前语言调整chunk_size尝试术语覆盖更全的向量化模型检查向量相似度分数分布Agent 调用工具时参数解析失败模型生成的 JSON 参数不规范在解析tool_call.function.arguments时增加异常捕获必要时引导模型重新生成参数流式响应时终端输出乱码编码问题Windows 终端默认编码不是 UTF-8在脚本开头执行sys.stdout.reconfigure(encodingutf-8)或修改终端代码页请求频繁超时网络连接不稳定或上游服务过载在客户端中设置合理的timeout并增加重试机制检查base_url是否可达所有请求都被限流单个 API Key 并发太高优先考虑在网关层做 Key 池轮询其次降低并发或者购买更高限额在排查这类问题时我建议遵循一个固定顺序先确认配置再确认依赖最后再看业务逻辑。不要一上来就怀疑模型出了问题。9. 最佳实践与工程建议9.1 Token 成本控制LLM 工具用多了之后最大的成本压力往往不是服务器而是 Token 费用。我总结了几个比较实用的控制技巧开启 Prompt 缓存很多云厂商支持对相同前缀做缓存如果系统 Prompt 很长合理固定前缀可以显著降本。压缩历史记录对话历史超过一定长度时可以对早期消息做摘要而不是全部保留原文。控制检索片段数量RAG 查询时top_k不是越大越好一般 3 到 5 个片段足够回答大多数问题。使用低成本模型处理简单任务比如意图识别可以用更小的模型只有复杂推理任务才用强模型。9.2 模型精度选择FP16、FP32 与 BF16很多刚接触 LLM 的开发者会纠结一个问题本地部署模型时到底应该用哪种精度这里简单展开一下。FP32单精度浮点数值精度最高但显存占用也最大。适合对精度要求极高的场景但在大模型推理中很少会直接用 FP32 推理。FP16半精度浮点显存占用比 FP32 少一半推理速度快但数值表示范围有限容易出现精度溢出。BF16BFloat16拥有和 FP32 相同的指数位因此数值范围更广在训练中更稳定。很多新硬件的推理引擎对 BF16 支持也比较好。如果要在本地部署模型通常更推荐优先尝试模型的量化版本或 BF16 版本。实际效果取决于你的 GPU 对哪种数据格式支持得更好。可以通过一个简单的压力测试脚本来对比生成质量和显存占用再决定生产环境用哪种精度。9.3 安全与权限边界这部分想认真提醒一下。LLM 工具很多时候会被赋予“调用外部 API”的能力这是很方便但也意味着风险。API Key 最小化不要把所有 Key 都放到同一个环境变量里。给不同的工具分配不同的 Key或者使用网关集中管理减少单个 Key 泄漏造成的影响。工具调用的白名单机制在 Agent 中注册工具时仔细审核函数可访问的资源范围避免模型被诱导去执行危险操作。输出校验模型生成的 SQL、HTTP 请求、文件路径等在执行前必须做合法性校验。不要直接信任模型输出的命令并执行。生产环境变更要留痕如果 Agent 会修改外部系统数据必须有权限控制、审计日志和回滚机制。测试环境和生产环境要隔离。9.4 可维护性与模块化如果你像我一样会持续往工具箱里添加新工具那么从第一天就需要注意可维护性每个工具保持单一职责不建议写一个随处可见的utils.py。配置通过环境变量管理代码里不出现明文密钥。函数调用、RAG 检索、模型调用分离成独立模块方便单元测试。为工具添加简单的命令行参数解析而不是每次都要修改代码才能切换配置。一个很值得做的改造是把四个工具的公共部分抽成一个BaseLLMClient类。比如重试机制、日志记录、Token 统计都可以在客户端类里统一完成而不需要每个工具各自实现一遍。10. 下一步可以往哪个方向深入到这里这套“离不开”的 LLM 工具集就介绍完了。我们来回顾一下每个工具的核心价值多模型 CLI 聊天终端解决 Prompt 调试和模型对比的效率问题。LLM API 统一网关收敛模型调用统一限流、日志和成本统计。本地 RAG 知识库问答让模型能回答领域私密资料。Agent 函数调用编排器让模型具备执行动作的能力。如果你之前只是用网页端聊天那你现在应该已经具备了自己搭建一套 LLM 开发工具箱的基础能力。如果你已经有项目经验也可以把我们讲的模块化设计思路融合进现有系统中。下一步可以关注的方向我觉得有这么几个学习主流 Agent 编排框架理解工具管理、状态持久化和多 Agent 协作的底层机制。深入 RAG 的进阶技巧比如混合检索、重排序、引用溯源。研究通过 MCP 协议标准化的工具接入方式让 Agent 可以更统一地连接外部数据源和应用系统。在动手实践时优先关注成本和安全这两个容易被忽略的维度。模型能力越来越强之后真正决定一个 LLM 工具能不能长期用下去的往往不是“能不能实现”而是“能不能稳定、安全、低成本地运行”。希望这份工具集对你也有帮助。你可以先复现其中一两个工具跑通后再逐步扩展成自己的版本。如果遇到问题欢迎在评论区描述现象和报错信息一起交流。 高效LLM开发工具箱：CLI、网关、RAG与Agent编排实战
1. 为什么我会攒出这套 LLM 工具链先交代一下背景。在做 LLM 应用开发的这大半年里我几乎每天都要跟各种模型 API、提示词、上下文窗口、Token 统计打交道。最开始还能靠 Postman 和几个零散脚本硬撑后来随着项目变多、模型切换频繁慢慢发现自己在重复劳动上耗掉的时间比真正写业务逻辑的时间还多。于是我开始把高频操作沉淀成一个个小工具。这些工具都不复杂单个拉出来也许只有几十行代码但组合起来之后日常工作效率提升非常明显。今天这篇文章想分享的就是这套我自认为“离不开”的 LLM 工具集包括设计思路、完整代码、使用方法以及我在踩坑过程中总结出来的一些经验和教训。如果你正处于下面这几个阶段这篇文章应该会比较适合你刚入门 LLM 应用开发习惯用网页端聊天但想开始接 API 做自动化。已经在写调用大模型的代码但觉得每次都要重新拼 Prompt、管理上下文很烦。想要搭建一套本地化的“模型调用 知识库问答 Agent 编排”小工具箱。对 Token 消耗不敏感但想学习如何从工程角度系统性管理 LLM 调用。文章不会只贴代码我会把每个工具解决什么问题、为什么这样设计、换成其他方案时有什么取舍都讲清楚。你可以直接复制代码到本地跑起来也可以按自己需要改造成适合自己的版本。2. 核心概念与工具全景2.1 LLM 应用开发的基本组成在展开代码之前先理清一个概念所谓的 LLM 应用通常不是“往 API 里丢一句话然后得到回复”这么简单。一个稍微成熟一点的项目至少会涉及这几个模块模型接入层负责对接 OpenAI、Anthropic、国内厂商或本地部署模型的 API统一鉴权、超时、重试和错误处理。会话管理层负责维护多轮对话的上下文避免聊天记录越来越长导致 Token 超限。提示词管理把常用 Prompt 抽成模板避免每次在代码里拼接字符串。工具调用层当模型需要查数据库、调接口、搜索网页时通过 Function Calling 机制把外部能力注册给模型。知识检索层也就是 RAG检索增强生成先从本地文档中检索出相关内容再交给模型总结回答解决“模型不知道你本地资料”的问题。资源观测层记录每次请求的耗时、Token 数、费用方便做成本分析和限流。下面要介绍的工具本质上就是把这些模块落成一个个可以直接运行的小项目。2.2 这套工具集包含什么为了不让工具变成一个庞然大物我把它们拆成了相对独立的模块。每个模块都可以单独运行也可以通过一个统一的入口去调用工具作用适合人群多模型 CLI 聊天终端在终端里随时切换模型对话支持流式输出和会话保存日常调试、快速验证 PromptLLM API 统一网关统一汇聚多个模型供应商的 API提供路由、重试、限流后端服务集成、团队共享 Key本地 RAG 知识库问答把本地文档切分、向量化检索后交给大模型回答问题个人知识库、内部资料问答Agent 函数调用编排器注册自定义工具让模型自动决定调用哪个函数完成任务自动化任务、信息查询类应用这四个工具覆盖了我日常开发中 90% 以上的场景。下面我会按顺序讲解先不急着引入太重的框架尽量用最直观的方式把原理讲明白。2.3 为什么不做成一个 All-in-One 应用你可能会问为什么不把这四个工具合并成一个系统我的理由是职责分离更清晰一个脚本坏掉不影响其他工具。可以单独替换某一部分比如今天想换一个向量库不用动聊天终端。方便通过命令行快速验证而不是每次都要启动一个 Web 前端。当然如果你已经有一个成熟项目也可以把这些工具的代码直接作为模块集成进去。本文的重点是讲解每块能力的实现方式而不是推荐某种固定架构。3. 环境准备与项目结构3.1 运行环境说明本文中的代码以 Python 为例因为 Python 在处理 API 调用、文本切分、数据存储等任务时比较方便生态也最齐全。操作系统Windows 10/11、macOS、Linux 均可。Python 版本建议 3.10 及以上代码中使用了较新的类型注解语法。包管理工具推荐使用venv或conda创建独立环境。模型 API示例代码以兼容 OpenAI SDK 的接口为例如果你使用的是本地模型或其他云厂商只需调整base_url和鉴权参数。需要特别说明的是不同框架和 SDK 的版本更新速度很快本文不会写死某个具体版本号。你在复现时建议使用当前较新的稳定版本。如果遇到接口变动以官方文档为准。3.2 创建项目结构我们可以创建一个名为llm-toolkit的目录整体结构如下llm-toolkit/ ├── .env.example # 环境变量模板 ├── requirements.txt # 依赖清单 ├── chat_cli.py # 工具一CLI 聊天终端 ├── api_gateway.py # 工具二API 统一网关 ├── rag_qa.py # 工具三RAG 知识库问答 ├── agent_orchestrator.py # 工具四Agent 函数调用编排器 ├── llm_client.py # 公共的 LLM 客户端封装 ├── config.py # 公共配置读取 └── docs/ # 存放本地知识库文档先创建虚拟环境并安装基础依赖python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate pip install openai python-dotenv fastapi uvicorn requests如果你的场景还需要做向量检索可以再安装对应的向量库客户端。为了示例简单我后面会先用一种基于本地 JSON 文件的轻量方案替换成正式向量库也很容易。3.3 统一配置管理LLM 工具最怕把 API Key 写死在代码里。这里我们统一通过环境变量管理配置新建.env文件# .env OPENAI_API_KEYsk-xxxx OPENAI_BASE_URLhttps://api.openai.com/v1 OPENAI_MODELgpt-4o-mini # 如果使用本地模型则可以改成 # OPENAI_BASE_URLhttp://localhost:11434/v1 # OPENAI_MODELllama3.1然后在config.py中统一读取import os from dotenv import load_dotenv load_dotenv() def get_config(): return { api_key: os.getenv(OPENAI_API_KEY), base_url: os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1), model: os.getenv(OPENAI_MODEL, gpt-4o-mini), }这样做的目的是让代码与具体密钥、模型名解耦。你可以在本地用 OpenAI在公司内网切换到私有化部署模型只需要改环境变量不需要动业务代码。4. 工具一多模型 CLI 聊天终端4.1 设计思路这个工具解决的核心痛点是当你需要快速验证一个 Prompt、比较不同模型的表现时打开网页端聊天显然太低效。我更希望直接在终端里输入一句话马上得到回复而且能一键切换模型、保存会话记录。设计上需要满足四个点支持流式输出不用等到完整回复生成才能看到内容。支持多轮对话自动维护上下文列表。支持通过命令切换模型比如输入/model gpt-4o。支持把会话保存为 JSON 文件方便事后复盘 Prompt 效果。4.2 核心代码实现先写一个公共的llm_client.py负责统一创建 OpenAI 客户端# llm_client.py from openai import OpenAI from config import get_config _config get_config() def get_client(): return OpenAI( api_key_config[api_key], base_url_config[base_url], ) def get_model(): return _config[model]接着是chat_cli.py实现一个简单的交互式聊天终端# chat_cli.py import json import datetime from llm_client import get_client, get_model client get_client() current_model get_model() history [] def chat(user_input: str): 发送一轮对话以流式方式打印回复 history.append({role: user, content: user_input}) try: stream client.chat.completions.create( modelcurrent_model, messageshistory, streamTrue, # 开启流式输出 ) except Exception as e: print(f[请求失败] {e}) history.pop() # 请求失败时移除这次的 user 输入 return print(Assistant: , end, flushTrue) full_reply for chunk in stream: if chunk.choices[0].delta.content: content chunk.choices[0].delta.content print(content, end, flushTrue) full_reply content print() # 换行 history.append({role: assistant, content: full_reply}) def save_session(filename: str): 保存当前会话到本地文件 payload { time: datetime.datetime.now().isoformat(), model: current_model, history: history, } with open(filename, w, encodingutf-8) as f: json.dump(payload, f, ensure_asciiFalse, indent2) print(f会话已保存到 {filename}) def main(): global current_model print(LLM Chat CLI - 输入 /help 查看命令) while True: try: user_input input(You: ).strip() except (EOFError, KeyboardInterrupt): print(\n再见) break if not user_input: continue if user_input.startswith(/): parts user_input.split(maxsplit1) cmd parts[0].lower() if cmd /help: print(/model 模型名 切换模型) print(/save 文件名 保存会话) print(/clear 清空会话) print(/exit 退出) elif cmd /model and len(parts) 1: current_model parts[1] print(f已切换到模型: {current_model}) elif cmd /save and len(parts) 1: save_session(parts[1]) elif cmd /clear: history.clear() print(会话已清空) elif cmd /exit: print(再见) break else: print(未知命令输入 /help 查看帮助) else: chat(user_input) if __name__ __main__: main()4.3 运行演示启动工具python chat_cli.py运行效果类似LLM Chat CLI - 输入 /help 查看命令 You: 用一句话介绍什么是大语言模型 Assistant: 大语言模型是一种基于海量文本训练的人工智能模型能够理解和生成自然语言文本。 You: /model gpt-4o 已切换到模型: gpt-4o You: 现在用更专业的语气再介绍一次 Assistant: 大语言模型Large Language Model是一种通过自监督学习在大型语料库上预训练的深度神经网络模型其核心能力在于对自然语言进行概率建模与生成。这里有两个实现细节值得展开说说。第一是流式输出。如果不开启streamTrue很多模型需要等十几秒才能一次性返回结果用户体验很差。开启后我们可以边生成边打印体感上会快很多。第二是失败回滚。在chat()函数中如果请求抛出异常我们会把本轮追加的 user 消息从history中弹出去。这样做是避免把一次失败请求也算进上下文导致后续对话携带了无效信息。4.4 这个工具可以怎么扩展加入多会话管理启动时选择继续某个历史会话而不是每次都是从空上下文开始。加入 Token 统计每次请求结束后打印消耗的 Token 数方便估算成本。加入 Prompt 模板系统把常用 Prompt 放到一个prompts目录通过/prompt 模板名快速插入。5. 工具二LLM API 统一网关5.1 为什么需要网关当我开始写多个 LLM 工具后很快遇到一个问题每个工具都要自己处理 API Key、重试逻辑、超时设置、限流。如果三个服务同时调用同一个模型还要担心会不会因为并发过高被限流。API 网关的作用就是把这些公共逻辑收敛到一个服务里。所有内部服务只需要请求网关由网关去和真实的大模型 API 交互。这样一来业务方不需要持有真实 API Key降低了泄漏风险。可以在网关层做统一的请求日志、Token 统计和费用汇总。可以针对不同供应商做路由比如某些模型走 OpenAI某些模型走本地部署服务。5.2 网关的核心能力我实现的网关并不复杂核心是下面这几块接收客户端请求参数格式保持和 OpenAI API 兼容。根据配置的供应商路由规则转发到真实模型服务。统一处理超时、重试、限流。返回标准格式的响应同时在响应头里附带本次请求的 Token 消耗。5.3 网关代码实现这里使用 FastAPI 来搭建代码非常简洁# api_gateway.py import time import uuid from fastapi import FastAPI, HTTPException, Request from pydantic import BaseModel from llm_client import get_client app FastAPI(titleLLM API Gateway) # 简单的请求日志正式项目可以改为写入数据库或日志系统 request_log [] class ChatRequest(BaseModel): model: str messages: list temperature: float 0.7 stream: bool False def call_real_api(req: ChatRequest): 转发到真实模型服务 client get_client() start time.time() try: resp client.chat.completions.create( modelreq.model, messagesreq.messages, temperaturereq.temperature, streamreq.stream, ) cost_time round(time.time() - start, 3) return resp, cost_time except Exception as e: raise HTTPException(status_code502, detailf上游模型调用失败: {e}) app.post(/v1/chat/completions) async def chat_completions(req: ChatRequest, raw_request: Request): # 如果开启了流式这里简化处理直接透传上游结果 if req.stream: resp, cost_time call_real_api(req) # FastAPI 对流式响应需要额外处理示例中先返回普通 JSON return {stream: True, cost_time: cost_time, data: 流式模式需要对接 SSE} resp, cost_time call_real_api(req) content resp.choices[0].message.content # 记录日志 log_entry { request_id: str(uuid.uuid4()), model: req.model, prompt_tokens: resp.usage.prompt_tokens if resp.usage else 0, completion_tokens: resp.usage.completion_tokens if resp.usage else 0, cost_time: cost_time, client_ip: raw_request.client.host if raw_request.client else unknown, } request_log.append(log_entry) return { id: log_entry[request_id], model: req.model, choices: [{message: {role: assistant, content: content}}], usage: { prompt_tokens: log_entry[prompt_tokens], completion_tokens: log_entry[completion_tokens], }, } app.get(/logs) def get_logs(): 查看最近的请求日志 return request_log[-50:] if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)启动网关python api_gateway.py然后用 curl 测试curl -X POST http://localhost:8000/v1/chat/completions \ -H Content-Type: application/json \ -d { model: gpt-4o-mini, messages: [{role: user, content: 你好介绍一下你自己}] }返回结果中会包含usage字段方便我们统计 Token 消耗。5.4 网关的进阶方向上面这个版本只是最小可用实现。在实际项目中你大概率还需要补充Key 管理把多个 API Key 存成池轮询或按权重分配避免单个 Key 触发限流。限流策略基于令牌桶或滑动窗口限制每个业务方每分钟的请求次数。缓存对相同请求做语义级或文本级缓存减少重复调用费用。SSE 流式支持当客户端需要流式响应时网关也要能透传 Server-Sent Events。其中流式透传是很多初学网关的人容易卡住的地方。核心思路是使用StreamingResponse在拿到上游的流对象后逐块转发给客户端from fastapi.responses import StreamingResponse app.post(/v1/chat/completions/stream) async def chat_completions_stream(req: ChatRequest): client get_client() stream client.chat.completions.create( modelreq.model, messagesreq.messages, streamTrue, ) def generate(): for chunk in stream: if chunk.choices[0].delta.content: yield fdata: {chunk.choices[0].delta.content}\n\n return StreamingResponse(generate(), media_typetext/event-stream)这个模式下客户端可以像调用真实 OpenAI API 一样用官方的 SDK 配合streamTrue来消费网关返回的流式数据。6. 工具三RAG 本地知识库问答6.1 RAG 是什么RAG 全称是 Retrieval-Augmented Generation翻译过来就是“检索增强生成”。它的核心思想是不直接让大模型凭记忆回答问题而是先从用户指定的文档中检索出相关内容把检索结果组装进 Prompt再让模型基于这些材料回答。这样做的好处很明显回答内容以你提供的资料为依据减少凭空捏造。可以随时更新知识库不用重新训练模型。回答中可以引用来源方便人工核对。我日常用得最多的场景是把团队的技术文档、项目周报、产品说明书放到本地然后通过 RAG 工具做问答。6.2 文档切分与向量化RAG 的流程可以拆成“入库”和“查询”两个阶段。入库阶段读取本地文档。按段落或固定块大小切分成片段。每个片段通过 Embedding 模型转成向量存入向量数据库。查询阶段把用户问题转成向量。在向量库中做相似度检索找到最相关的几个片段。把片段和问题一起发给大模型生成最终回答。这里为了演示方便我用一个 JSON 文件充当向量库避免引入过重的依赖。实际项目中可以替换为 Chroma、FAISS、Milvus 或 Qdrant。6.3 代码实现首先实现一个文档入库函数把文档切块并计算 Embedding# rag_qa.py import json import os from llm_client import get_client client get_client() # 简单的向量缓存文件 VECTOR_FILE local_vectors.json def read_docs_from_dir(doc_dir: str): 读取目录下所有 txt 文档 docs [] for filename in os.listdir(doc_dir): if filename.endswith(.txt): filepath os.path.join(doc_dir, filename) with open(filepath, r, encodingutf-8) as f: content f.read() docs.append({source: filename, content: content}) return docs def chunk_text(text: str, chunk_size: int 300): 将文本切分成固定长度的片段注意保留段落边界 paragraphs [p.strip() for p in text.split(\n) if p.strip()] chunks [] current for p in paragraphs: if len(current) len(p) chunk_size: current p \n else: if current: chunks.append(current.strip()) current p \n if current: chunks.append(current.strip()) return chunks def get_embedding(text: str): 调用 Embedding 接口生成向量 resp client.embeddings.create( modeltext-embedding-3-small, inputtext, ) return resp.data[0].embedding def build_index(doc_dir: str): 入库把所有文档切块并保存向量 docs read_docs_from_dir(doc_dir) records [] for doc in docs: chunks chunk_text(doc[content]) for chunk in chunks: vector get_embedding(chunk) records.append({ source: doc[source], content: chunk, vector: vector, }) with open(VECTOR_FILE, w, encodingutf-8) as f: json.dump(records, f, ensure_asciiFalse) print(f文档入库完成共 {len(records)} 个片段) def search(query: str, top_k: int 3): 查询根据问题向量做余弦相似度检索 query_vec get_embedding(query) with open(VECTOR_FILE, r, encodingutf-8) as f: records json.load(f) scored [] for rec in records: score cosine_similarity(query_vec, rec[vector]) scored.append({score: score, content: rec[content], source: rec[source]}) scored.sort(keylambda x: x[score], reverseTrue) return scored[:top_k] def cosine_similarity(vec1, vec2): 计算两个向量的余弦相似度 dot sum(a * b for a, b in zip(vec1, vec2)) norm1 sum(a * a for a in vec1) ** 0.5 norm2 sum(a * b for a, b in zip(vec2, vec2)) ** 0.5 if norm1 0 or norm2 0: return 0 return dot / (norm1 * norm2) def rag_answer(question: str): RAG 问答主流程 results search(question) if not results: return 知识库中没有找到相关内容 context \n\n.join([ f【来源{r[source]}】\n{r[content]} for r in results ]) prompt f请基于以下资料回答问题。如果资料中没有相关内容请直接说明不知道不要编造。 资料 {context} 问题{question} resp client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: 你是一个严谨的知识库问答助手回答时尽量结合给定资料。}, {role: user, content: prompt}, ], ) return resp.choices[0].message.content运行入库python rag_qa.py # 你需要在脚本入口调用 build_index(docs)运行问答# 在脚本中调用 print(rag_answer(这篇文档里提到的最佳实践是什么))回答中会带上我们检索到的资料片段作为上下文模型会根据这些片段组织答案。你可以看到这个实现虽然简单但已经具备了 RAG 的核心闭环。6.4 RAG 落地的关键坑点在实际项目中RAG 的效果往往取决于三个细节第一切分策略。固定chunk_size的做法虽然简单但可能会切断一个完整的概念。更好的做法是结合标题结构、段落语义进行切分或者使用递归字符分割器。第二Embedding 模型选择。不同语言的 Embedding 模型差异很大中文场景建议选用对中文支持好的模型并且要评估平均相似度分数不要盲目用默认模型。第三检索后处理。不要把检索到的所有内容都塞进 Prompt一是浪费时间 Token二是无关内容会干扰模型。合理的做法是先做相关性过滤再按分数降序排列。7. 工具四Agent 函数调用编排器7.1 函数调用Function Calling基础如果说 RAG 解决的是“让模型知道更多信息”那么 Agent 解决的是“让模型能执行动作”。大模型本身只能生成文本不能真正查询数据库、调用接口、写入文件。Function Calling 机制提供了一个标准化的办法你定义好一组函数告诉模型有哪些函数可用模型在回答时如果发现需要外部能力就会返回一个“我希望调用某个函数参数是什么”的结构化结果。这样模型就从一个“聊天机器人”升级成了“能主动分发任务的调度器”。7.2 工具注册与调度我这里的实现思路是定义一组工具函数每个函数都有一个名字、描述、参数 schema。把工具列表传给模型。模型返回工具调用请求时本地执行真正的函数。把函数执行结果追加到消息历史中再让模型继续回答。7.3 完整示例我们先定义两个简单的工具一个是查询本地服务器时间一个是模拟查询天气# agent_orchestrator.py import json import datetime from llm_client import get_client client get_client() def get_current_time(): 获取当前服务器时间 return datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) def get_weather(city: str): 模拟查询城市天气 weather_map { 北京: 晴25°C, 上海: 多云28°C, 深圳: 阵雨30°C, } return weather_map.get(city, f{city} 天气数据暂未收录) TOOLS [ { type: function, function: { name: get_current_time, description: 获取当前服务器时间, parameters: { type: object, properties: {}, }, }, }, { type: function, function: { name: get_weather, description: 查询指定城市的天气, parameters: { type: object, properties: { city: { type: string, description: 城市名称例如北京, } }, required: [city], }, }, }, ] TOOL_MAP { get_current_time: get_current_time, get_weather: get_weather, } def run_agent(user_input: str): 运行一个简单的 Agent 循环 messages [{role: user, content: user_input}] # 第一次调用带工具列表 resp client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolsTOOLS, ) msg resp.choices[0].message # 模型要求调用工具 if msg.tool_calls: messages.append(msg) # 保存带 tool_calls 的 assistant 消息 for tool_call in msg.tool_calls: fn_name tool_call.function.name args json.loads(tool_call.function.arguments) fn TOOL_MAP.get(fn_name) if fn: result fn(**args) else: result f未找到工具: {fn_name} # 把工具执行结果追加为 tool 消息 messages.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps(result, ensure_asciiFalse), }) # 第二次调用让模型基于工具结果生成回答 second_resp client.chat.completions.create( modelgpt-4o-mini, messagesmessages, ) return second_resp.choices[0].message.content # 模型直接回答没有调用工具 return msg.content if __name__ __main__: print(run_agent(现在几点了)) print(run_agent(北京今天天气怎么样)) print(run_agent(帮我分别查一下上海和深圳的天气))运行后模型会自行决定调用哪些工具。比如第一个问题会触发get_current_time第二个问题会触发get_weather第三个问题甚至可能连续调用两次get_weather然后把结果汇总成一句回答。7.4 Agent 编排的复杂度控制函数调用本身不难难在处理复杂情况时的状态管理。以我自己的经验有几个问题比较常见工具调用循环深度模型可能调用工具之后还想再调用另一个工具形成多轮工具调用。代码里需要加一个最大循环次数限制否则可能死循环。并行工具调用新版模型支持一次返回多个tool_calls处理时要遍历全部调用而不是只处理第一个。错误处理工具执行时可能抛异常比如请求外部接口超时。这时应该把错误信息作为 tool 消息返回给模型让模型学会“承认失败”或换个方式尝试。如果你要做一个生产级 Agent建议引入状态机或工作流引擎来管理任务状态而不是只用简单的 while 循环。这一点在下一节的最佳实践中会再提到。8. 常见问题与排查思路在复现或使用上述代码时你很可能遇到下面这些报错。整理成表格方便快速定位。问题现象常见原因解决思路OpenAI客户端报认证失败API Key 未配置或失效检查.env文件确认OPENAI_API_KEY已正确加载使用print(get_config())调试模型返回内容经常被截断上下文长度超过模型最大限制开启对话前先截断或压缩历史消息使用max_tokens控制回复长度调用 RAG 时检索结果不相关文档切分太碎或 Embedding 模型不适合当前语言调整chunk_size尝试术语覆盖更全的向量化模型检查向量相似度分数分布Agent 调用工具时参数解析失败模型生成的 JSON 参数不规范在解析tool_call.function.arguments时增加异常捕获必要时引导模型重新生成参数流式响应时终端输出乱码编码问题Windows 终端默认编码不是 UTF-8在脚本开头执行sys.stdout.reconfigure(encodingutf-8)或修改终端代码页请求频繁超时网络连接不稳定或上游服务过载在客户端中设置合理的timeout并增加重试机制检查base_url是否可达所有请求都被限流单个 API Key 并发太高优先考虑在网关层做 Key 池轮询其次降低并发或者购买更高限额在排查这类问题时我建议遵循一个固定顺序先确认配置再确认依赖最后再看业务逻辑。不要一上来就怀疑模型出了问题。9. 最佳实践与工程建议9.1 Token 成本控制LLM 工具用多了之后最大的成本压力往往不是服务器而是 Token 费用。我总结了几个比较实用的控制技巧开启 Prompt 缓存很多云厂商支持对相同前缀做缓存如果系统 Prompt 很长合理固定前缀可以显著降本。压缩历史记录对话历史超过一定长度时可以对早期消息做摘要而不是全部保留原文。控制检索片段数量RAG 查询时top_k不是越大越好一般 3 到 5 个片段足够回答大多数问题。使用低成本模型处理简单任务比如意图识别可以用更小的模型只有复杂推理任务才用强模型。9.2 模型精度选择FP16、FP32 与 BF16很多刚接触 LLM 的开发者会纠结一个问题本地部署模型时到底应该用哪种精度这里简单展开一下。FP32单精度浮点数值精度最高但显存占用也最大。适合对精度要求极高的场景但在大模型推理中很少会直接用 FP32 推理。FP16半精度浮点显存占用比 FP32 少一半推理速度快但数值表示范围有限容易出现精度溢出。BF16BFloat16拥有和 FP32 相同的指数位因此数值范围更广在训练中更稳定。很多新硬件的推理引擎对 BF16 支持也比较好。如果要在本地部署模型通常更推荐优先尝试模型的量化版本或 BF16 版本。实际效果取决于你的 GPU 对哪种数据格式支持得更好。可以通过一个简单的压力测试脚本来对比生成质量和显存占用再决定生产环境用哪种精度。9.3 安全与权限边界这部分想认真提醒一下。LLM 工具很多时候会被赋予“调用外部 API”的能力这是很方便但也意味着风险。API Key 最小化不要把所有 Key 都放到同一个环境变量里。给不同的工具分配不同的 Key或者使用网关集中管理减少单个 Key 泄漏造成的影响。工具调用的白名单机制在 Agent 中注册工具时仔细审核函数可访问的资源范围避免模型被诱导去执行危险操作。输出校验模型生成的 SQL、HTTP 请求、文件路径等在执行前必须做合法性校验。不要直接信任模型输出的命令并执行。生产环境变更要留痕如果 Agent 会修改外部系统数据必须有权限控制、审计日志和回滚机制。测试环境和生产环境要隔离。9.4 可维护性与模块化如果你像我一样会持续往工具箱里添加新工具那么从第一天就需要注意可维护性每个工具保持单一职责不建议写一个随处可见的utils.py。配置通过环境变量管理代码里不出现明文密钥。函数调用、RAG 检索、模型调用分离成独立模块方便单元测试。为工具添加简单的命令行参数解析而不是每次都要修改代码才能切换配置。一个很值得做的改造是把四个工具的公共部分抽成一个BaseLLMClient类。比如重试机制、日志记录、Token 统计都可以在客户端类里统一完成而不需要每个工具各自实现一遍。10. 下一步可以往哪个方向深入到这里这套“离不开”的 LLM 工具集就介绍完了。我们来回顾一下每个工具的核心价值多模型 CLI 聊天终端解决 Prompt 调试和模型对比的效率问题。LLM API 统一网关收敛模型调用统一限流、日志和成本统计。本地 RAG 知识库问答让模型能回答领域私密资料。Agent 函数调用编排器让模型具备执行动作的能力。如果你之前只是用网页端聊天那你现在应该已经具备了自己搭建一套 LLM 开发工具箱的基础能力。如果你已经有项目经验也可以把我们讲的模块化设计思路融合进现有系统中。下一步可以关注的方向我觉得有这么几个学习主流 Agent 编排框架理解工具管理、状态持久化和多 Agent 协作的底层机制。深入 RAG 的进阶技巧比如混合检索、重排序、引用溯源。研究通过 MCP 协议标准化的工具接入方式让 Agent 可以更统一地连接外部数据源和应用系统。在动手实践时优先关注成本和安全这两个容易被忽略的维度。模型能力越来越强之后真正决定一个 LLM 工具能不能长期用下去的往往不是“能不能实现”而是“能不能稳定、安全、低成本地运行”。希望这份工具集对你也有帮助。你可以先复现其中一两个工具跑通后再逐步扩展成自己的版本。如果遇到问题欢迎在评论区描述现象和报错信息一起交流。