企业官网建设流程全解析

这次不聊概念，直接动手搭一套能跑、能接业务、能批量处理的智能体工具链。

很多人学 Agent 开发，第一步就装框架。装完 LangChain 这类重量级依赖，demo 能跑通，但真正要接到业务里，面对模型版本切换、工具参数校验、上下文管理、并发请求，反而不知道从哪里改。这篇指南换一个思路：不用任何重量级 Agent 框架，从模型接入、函数调用、记忆存储到 API 服务，用最朴素的 Python 代码把一条完整的工具链搭出来。

成品核心代码量不大，但每一层都可以替换、可以测试、可以接进现有业务系统。读完之后，你应该能独立完成一个最小可用 Agent 服务的开发，并且知道后续要加什么模块、踩什么坑。

1. 核心能力速览

这里重点强调一件事：工具链不是某一个框架，而是多个模块的组合。模型接入负责和 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 chromadb

3.3 模型服务选择

如果你使用云厂商模型，只需要一段 API Key 和接口地址。如果你在本地部署模型，需要确认：

• 显卡显存是否满足模型尺寸需求，建议从 7B 到 14B 的量化模型开始测试。
• 本地推理服务是否支持函数调用（tools），多数较新版本模型支持。
• 启动后接口地址是什么，例如http://127.0.0.1:8000/v1。

这里给一个通用的模型接入示例，兼容 OpenAI 协议的服务都可以用同一套代码：

from openai import OpenAI client = OpenAI( base_url="http://127.0.0.1:8000/v1", # 本地或云端 OpenAI 兼容地址 api_key="sk-your-key" ) def chat_completion(messages, tools=None, temperature=0.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 response

4. Agent 工具链的整体架构设计

动手前先把模块边界画清楚。一套完整的 Agent 工具链，可以拆成五层：

模块之间尽量解耦。模型层只负责发请求收响应，不关心业务逻辑。工具层只负责执行，不关心模型怎么调用。编排层负责把模型输出和工具调用串联起来。

核心 loop 通常是这个流程：

1. 接收用户输入。
2. 将系统提示词、历史消息、用户输入拼接成 messages。
3. 调用模型。
4. 如果模型返回 tool_calls，执行对应工具，把工具结果回传给模型。
5. 如果模型返回普通文本，把这轮结果返回给用户。

这个循环是 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": f"unknown 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, tools=tools) 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, tools=tools) 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, tools=tools) 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_turns=10): 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( name="knowledge_base", embedding_function=embed_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_results=top_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_results=top_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(title="Agent 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_model=ChatResponse) async def chat(request: ChatRequest): if not request.content.strip(): raise HTTPException(status_code=400, detail="content cannot be empty") session = get_session(request.session_id) reply = session.run(request.content) return ChatResponse(session_id=request.session_id, reply=reply) @app.get("/v1/agent/health") async def health(): return {"status": "ok"}

启动命令：

uvicorn main:app --host 127.0.0.1 --port 8000 --reload

7.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, json=payload, timeout=120) 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 f"data: {chunk}\n\n" return StreamingResponse(event_generator(), media_type="text/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_ok=True) for file_path in Path(input_dir).glob("*.txt"): text = file_path.read_text(encoding="utf-8") result = run_agent(f"请对以下内容进行摘要：\n{text}") output_path = Path(output_dir) / f"{file_path.stem}_summary.md" output_path.write_text(result, encoding="utf-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 8000

10. 常见问题与排查方法

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(level=logging.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、引入更完整的评测体系。这些话题，后面的文章可以逐个展开。