从零搭建Agent智能体工具链：模型接入、函数调用与API服务实战

这次不聊概念直接动手搭一套能跑、能接业务、能批量处理的智能体工具链。很多人学 Agent 开发第一步就装框架。装完 LangChain 这类重量级依赖demo 能跑通但真正要接到业务里面对模型版本切换、工具参数校验、上下文管理、并发请求反而不知道从哪里改。这篇指南换一个思路不用任何重量级 Agent 框架从模型接入、函数调用、记忆存储到 API 服务用最朴素的 Python 代码把一条完整的工具链搭出来。成品核心代码量不大但每一层都可以替换、可以测试、可以接进现有业务系统。读完之后你应该能独立完成一个最小可用 Agent 服务的开发并且知道后续要加什么模块、踩什么坑。1. 核心能力速览能力项说明项目定位Agent 智能体工具链开发实战教程技术栈Python、OpenAI 兼容接口、FastAPI、向量数据库核心模块模型接入层、工具函数调度、记忆系统、API 服务、批量任务部署方式命令行脚本或 HTTP API 服务批量任务支持目录批量处理、队列化任务分发API 能力FastAPI 提供标准 HTTP 接口扩展方向记忆持久化、多智能体协作、评估测试适合读者有 Python 基础、想深入 Agent 工程化的开发者这里重点强调一件事工具链不是某一个框架而是多个模块的组合。模型接入负责和 LLM 通信函数调用让模型能操作外部系统记忆模块解决上下文丢失问题API 服务把能力暴露给上层业务批量任务解决规模化执行。每个模块职责单一组合起来才叫工具链。2. 适用场景与使用边界这套工具链适合以下场景需要把大模型接入内部知识库做问答和检索增强。需要让模型调用数据库、搜索接口、内部 API完成具体业务操作。需要对一批文档、工单、日志做自动分类、摘要、内容提取。需要给前端或第三方系统提供一个稳定的 Agent 交互接口。需要记录每一次对话做后续数据分析和效果优化。不适合的场景也需要说清楚需要强实时、低延迟的简单对话直接用模型 API 就好不要套一层又一层模块Agent 工具链的优势是复杂任务编排不是极速响应。需要绝对可靠的结果输出当前 LLM 天然存在幻觉必须加人工复核环节。设备资源有限的纯 CPU 环境如果用大模型做 Base 模型推理速度会很难接受建议优先考虑远程模型服务或量化小模型。合规边界方面如果 Agent 要处理用户隐私数据、人脸信息、声音、版权材料必须确认数据来源有合法授权。工具函数的执行权限也要做最小化设计不能在 Agent 里放一个execute_command万能接口直接跑任意系统命令更不要把数据库写权限无条件暴露给模型。本地部署的数据也要做访问控制API 服务尽量不要直接绑定 0.0.0.0 暴露到公网。3. 环境准备与前置条件开始写代码之前先把环境准备好。下面是一份通用检查清单具体版本以你本机的兼容性为准。3.1 基础环境Python 3.10 或更高版本。pip 包管理工具。一个可以访问的 LLM 服务可以是云厂商的模型接口也可以是本地的 vLLM、Ollama、llama.cpp 等服务。它们大多提供 OpenAI 兼容接口。FastAPI 和 Uvicorn 用于提供 HTTP 服务。向量数据库可选用于长期记忆和知识检索。3.2 安装依赖python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate pip install openai fastapi uvicorn pydantic requests # 如果做向量检索可以安装 chromadb 或 faiss-cpu pip install chromadb3.3 模型服务选择如果你使用云厂商模型只需要一段 API Key 和接口地址。如果你在本地部署模型需要确认显卡显存是否满足模型尺寸需求建议从 7B 到 14B 的量化模型开始测试。本地推理服务是否支持函数调用tools多数较新版本模型支持。启动后接口地址是什么例如http://127.0.0.1:8000/v1。这里给一个通用的模型接入示例兼容 OpenAI 协议的服务都可以用同一套代码from openai import OpenAI client OpenAI( base_urlhttp://127.0.0.1:8000/v1, # 本地或云端 OpenAI 兼容地址 api_keysk-your-key ) def chat_completion(messages, toolsNone, temperature0.7): params { model: your-model-name, messages: messages, temperature: temperature, } if tools: params[tools] tools params[tool_choice] auto response client.chat.completions.create(**params) return response4. Agent 工具链的整体架构设计动手前先把模块边界画清楚。一套完整的 Agent 工具链可以拆成五层层级职责关键组件接入层接收用户请求管理会话FastAPI、WebSocket编排层解析用户意图决定下一步Agent Loop、Planning模型层与大模型通信OpenAI 兼容客户端工具层调用外部系统搜索、数据库、HTTP API记忆层短期与长期上下文对话历史、向量存储模块之间尽量解耦。模型层只负责发请求收响应不关心业务逻辑。工具层只负责执行不关心模型怎么调用。编排层负责把模型输出和工具调用串联起来。核心 loop 通常是这个流程接收用户输入。将系统提示词、历史消息、用户输入拼接成 messages。调用模型。如果模型返回 tool_calls执行对应工具把工具结果回传给模型。如果模型返回普通文本把这轮结果返回给用户。这个循环是 Agent 工具链的最小内核后面所有模块都是围绕它扩展。5. 从零搭建 Agent 核心模块5.1 定义工具函数先从工具层开始。工具函数是模型操作外部系统的桥梁必须定义成结构化 JSON Schema模型才能理解什么时候用什么工具。下面定义两个工具一个查询知识库一个执行只读 SQL。tools [ { type: function, function: { name: search_knowledge, description: 检索内部知识库获取与问题相关的文档片段, parameters: { type: object, properties: { query: { type: string, description: 检索关键词或问题描述 } }, required: [query] } } }, { type: function, function: { name: query_readonly_sql, description: 对业务数据库执行只读 SQL 查询仅允许 SELECT, parameters: { type: object, properties: { sql: { type: string, description: 只读 SQL 查询语句 } }, required: [sql] } } } ]工具描述要写清楚用途和参数。描述越模糊模型越容易传错参数。5.2 实现工具调度器工具调度器负责根据模型返回的工具名称和参数路由到对应函数。import json def dispatch_tool(name: str, arguments: str): args json.loads(arguments) if name search_knowledge: return knowledge_search(args.get(query)) elif name query_readonly_sql: return sql_query(args.get(sql)) else: return json.dumps({error: funknown tool: {name}})工具函数的具体实现你需要替换成自己的逻辑。比如knowledge_search调用向量数据库检索sql_query连接数据库执行只读查询。注意这里强调只读是为了安全写操作不建议直接暴露给模型。5.3 Agent 主循环主循环是整个工具链的核心。它负责维护消息列表判断模型是否要调用工具以及决定何时终止。def run_agent(user_input: str, max_rounds: int 5): messages [ { role: system, content: 你是一个智能助手。你可以调用工具来获取知识或查询数据。 如果工具返回结果请基于结果回答用户的问题。 }, {role: user, content: user_input} ] for _ in range(max_rounds): response chat_completion(messages, toolstools) msg response.choices[0].message messages.append(msg) if msg.tool_calls: for call in msg.tool_calls: tool_result dispatch_tool(call.function.name, call.function.arguments) messages.append({ role: tool, tool_call_id: call.id, content: tool_result }) else: return msg.content return 已达到最大执行轮数任务终止每一步的细节都很重要tool_call_id必须和模型返回的 id 一致否则模型无法对应工具结果。工具结果要转成字符串模型接口要求的 content 是字符串。max_rounds必须限制否则模型可能陷入循环调用工具的陷阱。系统提示词要说明工具的用途但不要写死业务逻辑让模型自己判断。5.4 测试一个简单场景if __name__ __main__: result run_agent(帮我查一下知识库里关于 Agent 工具链的资料) print(result)如果一切正常模型会先返回tool_calls调度器执行检索把结果回传模型再次生成答案。最终输出应该是一段基于检索结果的回答而不是模型自己编的内容。5.5 多轮对话扩展上面的代码只处理单轮输入真实业务需要多轮对话。最简单的做法是维护一个会话对象每次把历史消息带上。class AgentSession: def __init__(self, session_id: str): self.session_id session_id self.messages [] self.max_history 20 def add_message(self, message: dict): self.messages.append(message) if len(self.messages) self.max_history: self.messages self.messages[-self.max_history:] def run(self, user_input: str): self.add_message({role: user, content: user_input}) response chat_completion(self.messages, toolstools) msg response.choices[0].message self.add_message(msg) if msg.tool_calls: for call in msg.tool_calls: tool_result dispatch_tool(call.function.name, call.function.arguments) self.add_message({ role: tool, tool_call_id: call.id, content: tool_result }) # 工具结果回传后需要再次调用模型生成最终答案 final_response chat_completion(self.messages, toolstools) final_msg final_response.choices[0].message self.add_message(final_msg) return final_msg.content return msg.content这个思路能处理绝大多数业务场景。真正生产环境还需要考虑历史消息压缩、对话隔离、并发安全这些会在后面的 API 服务章节展开。6. 记忆系统从短期记忆到向量检索Agent 的上下文窗口再大也有限。短期记忆靠拼接历史消息长期记忆必须靠外部存储。6.1 短期记忆管理短期记忆就是最近几轮对话。要注意的问题不能无限累积消息否则会超出上下文窗口。超长历史要做摘要压缩而不是简单截断。工具调用过程中的中间消息要不要保留看业务需求。有些中间结果很长模型只需要最终结论。一个简单的策略是保留最近 10 轮用户与助手消息工具调用中间结果只保留最近 2 轮。这个策略可以先用配置控制跑起来再调。class ShortMemory: def __init__(self, max_turns10): self.max_turns max_turns self.messages [] def append(self, message: dict): self.messages.append(message) if len(self.messages) self.max_turns * 2: self.messages self.messages[-(self.max_turns * 2):]6.2 向量记忆与知识检索长期记忆适合用向量数据库实现。核心流程先把文档切块用 Embedding 模型编码成向量存入库中查询时把用户问题编码做相似度检索把 TopK 结果放进上下文。import chromadb from chromadb.utils import embedding_functions client chromadb.PersistentClient(path./memory_store) embed_fn embedding_functions.DefaultEmbeddingFunction() collection client.get_or_create_collection( nameknowledge_base, embedding_functionembed_fn ) def add_document(doc_id: str, text: str, metadata: dict None): collection.add( ids[doc_id], documents[text], metadatas[metadata or {}] ) def knowledge_search(query: str, top_k: int 3) - str: results collection.query(query_texts[query], n_resultstop_k) docs results[documents][0] return \n\n.join(docs)然后把这个knowledge_search替换掉前面工具调度器里的占位函数。向量检索的坑文档切块大小要合适太短语义不全太长检索精度下降。Embedding 模型要固定不要来回换否则向量空间不一致。检索结果要回传完整来源信息方便审计。6.3 会话隔离多用户场景下每个 session 的记忆不能串。最简单的方式向量数据里增加 session_id 字段查询时带上过滤条件。def session_search(session_id: str, query: str, top_k: int 3) - str: results collection.query( query_texts[query], n_resultstop_k, where{session_id: session_id} ) return results[documents][0]生产环境还要考虑权限控制用户 A 不能检索到用户 B 的数据。这个where过滤只是基础更严格的做法是在应用层做权限校验。7. 把 Agent 封装成 API 服务核心跑通之后下一步就是暴露成 HTTP 服务。用 FastAPI 做这一层非常合适。7.1 FastAPI 服务代码from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI(titleAgent Toolchain Service) class ChatRequest(BaseModel): session_id: str content: str class ChatResponse(BaseModel): session_id: str reply: str # 简单会话池生产环境建议用 Redis session_pool {} def get_session(session_id: str): if session_id not in session_pool: session_pool[session_id] AgentSession(session_id) return session_pool[session_id] app.post(/v1/agent/chat, response_modelChatResponse) async def chat(request: ChatRequest): if not request.content.strip(): raise HTTPException(status_code400, detailcontent cannot be empty) session get_session(request.session_id) reply session.run(request.content) return ChatResponse(session_idrequest.session_id, replyreply) app.get(/v1/agent/health) async def health(): return {status: ok}启动命令uvicorn main:app --host 127.0.0.1 --port 8000 --reload7.2 curl 调用测试curl -X POST http://127.0.0.1:8000/v1/agent/chat \ -H Content-Type: application/json \ -d {session_id: user-001, content: 帮我查询昨天订单数据}返回结果{ session_id: user-001, reply: 根据数据库查询结果昨天共有 125 个订单…… }7.3 Python 客户端调用import requests url http://127.0.0.1:8000/v1/agent/chat payload { session_id: user-002, content: 总结一下知识库里关于模型微调的内容 } resp requests.post(url, jsonpayload, timeout120) print(resp.json()[reply])7.4 流式输出实时对话场景建议用流式接口。FastAPI 可以用StreamingResponse实现from fastapi.responses import StreamingResponse app.post(/v1/agent/chat/stream) async def chat_stream(request: ChatRequest): session get_session(request.session_id) def event_generator(): for chunk in session.run_stream(request.content): yield fdata: {chunk}\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)流式输出的核心问题是工具调用阶段怎么处理。通常做法工具调用阶段不流式输出等工具结果拿到后最后生成阶段再流式输出。否则用户会看到中间过程体验反而不好。7.5 生产环境的服务改进会话池改成 Redis解决多实例部署的会话一致性问题。接口加 API Key 校验不能裸奔。加请求日志记录每次请求的 session_id、输入长度、耗时、模型输出。限流防止单个用户的超大请求打爆后端。错误返回要统一格式前端好处理。8. 批量任务设计与实现Agent 不只服务在线对话很多场景是批量跑数据批量总结文档、批量打标工单、批量抽取合同关键信息。这些任务不适合走实时 API应该走队列。8.1 文件目录批量处理最简单的批量方案输入目录放文件Agent 逐个处理结果写到输出目录。import os from pathlib import Path def batch_process(input_dir: str, output_dir: str): os.makedirs(output_dir, exist_okTrue) for file_path in Path(input_dir).glob(*.txt): text file_path.read_text(encodingutf-8) result run_agent(f请对以下内容进行摘要\n{text}) output_path Path(output_dir) / f{file_path.stem}_summary.md output_path.write_text(result, encodingutf-8) print(f处理完成: {file_path.name})8.2 队列化任务设计文件方案只适合小规模。正式一点的做法是引入任务队列# 这里用一个简单的列表模拟队列生产环境建议 Redis Celery 或消息队列 task_queue [] class TaskItem: def __init__(self, task_id: str, payload: dict): self.task_id task_id self.payload payload self.status pending def enqueue_task(task_id: str, payload: dict): task_queue.append(TaskItem(task_id, payload)) def worker_loop(): while True: if not task_queue: time.sleep(1) continue task task_queue.pop(0) task.status processing try: result run_agent(task.payload[content]) task.status done # 保存结果到数据库或输出目录 except Exception as e: task.status failed # 记录错误日志批量任务的几个建议任务必须有唯一 ID方便追踪。任务状态要落库不能只存在内存里。失败任务要有重试机制但重试次数要限制。每批任务都要有进度日志方便监控。并发数量要控制避免同时打爆模型服务。{ task_id: 20250201_001, input_file: ./inputs/contract_001.txt, output_file: ./outputs/contract_001_summary.md, status: pending, retry_count: 0 }9. 资源占用与性能观察Agent 工具链的资源消耗模型和普通 Web 服务不太一样。主要观察这几个指标。9.1 模型服务侧在线对话场景看请求 QPS、延迟、并发数。模型推理延迟取决于模型大小、输入长度、输出长度和硬件。如果使用本地模型显存占用是硬指标。模型加载后显存占用基本固定但推理时的 KV Cache 会随上下文长度波动。批量任务场景关注吞吐量也就是每小时能处理多少个文件而不是单次延迟。9.2 应用层会话池对象会占用内存多用户场景要定期清理不活跃会话。向量检索的延迟一般在几十毫秒到几百毫秒取决于数据量和索引质量。长上下文对话会显著增加模型推理延迟可能从 1 秒涨到 10 秒以上。工具调用会增加多轮模型请求一个包含两次工具调用的任务实际模型调用次数可能是 3 到 4 次。9.3 性能优化方向对工具的中间结果做摘要压缩而不是原样回传。工具返回 5000 字文档摘要可能模型只需要其中 200 字结论。并行检索多个独立工具可以并行调用而不是串行等待。批量任务用异步 IO 提高并发度但要注意模型服务限流。缓存相同问题的检索结果可以做短时间缓存。历史消息压缩对超过 N 轮的旧消息先做摘要再放入上下文。9.4 观察命令# 查看 GPU 显存占用 nvidia-smi -l 1 # 查看应用进程内存 top -p $(pgrep -f uvicorn main:app) # 查看端口监听状态 netstat -an | grep 800010. 常见问题与排查方法问题现象可能原因排查方式解决方案模型返回空内容模型服务超时或参数错误查看模型服务日志直接调模型接口测试检查接口地址、API Key、输入格式工具调用一直失败tool_call_id 对不上打印 messages 列表检查工具消息格式确保 tool_call_id 与模型返回完全一致工具结果没生效工具结果没有追加进 messages检查主循环代码确认工具结果追加位置工具结果必须在调用工具后的下一轮请求前追加上下文越来越长延迟暴涨历史消息没有压缩观察 requests 输入 token 数增加历史消息压缩或摘要策略批量任务卡住单个任务抛异常worker 循环中断查看 worker 日志任务处理加 try-except单任务失败不阻塞队列向量检索结果不相关文档切块不合理或 Embedding 模型不匹配单独测试检索质量调整切块大小固定 Embedding 模型API 请求超时模型推理耗时长客户端超时时间短查看模型服务响应耗时调大客户端超时时间或改用异步任务多用户会话串线session_id 管理错误或会话池未隔离检查会话池代码确保 session_id 唯一向量存储增加 session_id 过滤10.1 工具函数执行报错的排查工具函数内部出错时不要把异常直接抛出。建议捕获异常返回一个可读的错误信息给模型让它换个参数重试。def safe_dispatch_tool(name: str, arguments: str) - str: try: return dispatch_tool(name, arguments) except Exception as e: return json.dumps({ error: str(e), suggestion: 请检查工具参数是否合法或换一种查询方式 })这样模型可以根据错误信息调整参数而不是让整个 Agent 任务崩溃。10.2 模型不支持工具调用的兜底如果本地部署的模型不支持 function calling可以用 prompt 模拟把工具列表写进系统提示词要求模型输出特定格式的 JSON。但这种方式不稳定优先建议换一个支持工具调用的模型版本。11. 最佳实践与使用建议11.1 从最小闭环开始第一次跑通不要追求功能全面。先实现一个模型接口、一个工具函数、一个主循环。确认这三者能跑通再逐步增加记忆、批量任务、API 服务。11.2 模块边界要清晰模型层、工具层、编排层、记忆层分开写。不要在一个文件里塞全部逻辑。这样做的直接好处是换模型服务时只改模型层加新工具时只加工具层不影响其他代码。11.3 日志和监控生产环境必须记录用户输入和 Agent 最终输出。每次工具调用的名称、参数、执行结果、耗时。模型调用次数和各阶段耗时。错误堆栈。这些日志是排查问题的第一手材料。没有日志的 Agent 服务出问题只能靠猜。建议的最小日志配置import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(agent) logger.info(session%s user_input%s, session_id, user_input) logger.info(tool_call%s args%s result%s, tool_name, tool_args, tool_result) logger.warning(agent loop exceeded max_rounds, session%s, session_id)11.4 提示词工程与工具描述的迭代Agent 效果不好问题往往不在模型而在工具描述和系统提示词。工具描述要写清楚“什么时候该用”“参数含义”系统提示词要写清楚“你的角色边界”。这两部分值得反复迭代系统提示词里写清楚输出格式要求避免 Agent 返回过长废话。工具描述里举例说明参数格式。如果 Agent 频繁调用错工具检查描述是否产生歧义。11.5 评估机制给 Agent 每次任务打分是否完成任务中途有没有幻觉工具调用参数是否正确先建立小规模评测集每次改动后跑一遍效果不退化再上线。简单实现{ session_id: eval_001, task: 查询订单数量并生成摘要, expected_tool_call: query_readonly_sql, expected_output_contains: [125], actual_output: ..., pass: true }11.6 合规与安全工具函数只提供最小权限。涉及用户隐私数据时接口层必须先鉴权。对外提供服务时所有输出都应经过安全过滤和审计。如果 Agent 涉及人脸、音色、版权材料处理必须确认授权文件完整。12. 总结与下一步这套 Agent 工具链的核心价值在于它用极少的代码覆盖了从模型接入到批量处理的完整链路。最小闭环只需要几百行 Python后续所有模块都可以独立替换和升级。第一次上手建议先跑通第 5 节的最小 Agent 循环确认你选的模型支持工具调用然后加一个真实业务工具函数比如搜索你自己的知识库。跑通之后再考虑 API 服务、批量任务和记忆扩展。最容易踩的坑是三个工具调用和消息回传的顺序问题、上下文无限增长导致的延迟抬升、批量任务的异常中断。下一步可以扩展的方向包括接入 Agent 框架做多智能体协作、增加更复杂的任务规划能力、把会话记忆迁移到 Redis 和 Postgres、引入更完整的评测体系。这些话题后面的文章可以逐个展开。