AI Agent 性能优化教程｜附完整命令

随着大模型能力不断增强，越来越多团队开始构建 AI Agent：它不仅能对话，还能调用工具、检索知识库、执行代码、访问数据库、自动规划任务并完成复杂流程。

但在真实业务中，很多 Agent 上线后会遇到类似问题：

• 响应速度慢，用户等待时间长；
• Token 消耗过高，调用成本失控；
• 多轮任务中频繁跑偏，稳定性差；
• 工具调用过多，执行链路冗长；
• 检索结果不准，导致回答质量不稳定；
• 并发一高就超时、排队甚至服务崩溃。

本文将从 模型选择、Prompt 优化、上下文压缩、RAG 检索优化、工具调用优化、并发与缓存、日志监控、部署调优 等角度，系统讲解 AI Agent 的性能优化方法，并附上可直接使用的完整命令。

一、AI Agent 性能瓶颈从哪里来？

在优化之前，我们需要先理解 AI Agent 的典型执行链路。

一个常见 Agent 流程如下：

用户输入
  ↓
意图识别
  ↓
任务规划
  ↓
上下文构建
  ↓
知识库检索 / 工具调用 / API 调用
  ↓
模型推理
  ↓
结果校验
  ↓
返回用户

相比普通 ChatBot，Agent 最大的性能开销通常来自以下几个方面：

二、第一步：建立性能基线

优化不能靠感觉，必须先建立可量化指标。

常用指标包括：

• TTFT：Time To First Token，首 token 延迟；
• Total Latency：完整响应耗时；
• Input Tokens：输入 token 数；
• Output Tokens：输出 token 数；
• Tool Calls：工具调用次数；
• RAG Latency：检索耗时；
• Agent Iterations：Agent 推理轮数；
• Cost Per Request：单次请求成本；
• Success Rate：任务成功率；
• Timeout Rate：超时率。

如果你使用 Python，可以先写一个简单压测脚本。

1. 安装依赖

pip install openai tiktoken aiohttp pandas tqdm

2. 创建基础测试脚本

新建文件：

mkdir -p ai-agent-benchmark
cd ai-agent-benchmark
touch benchmark.py

写入以下内容：

import asyncio
import time
import statistics
from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key="YOUR_API_KEY",
    base_url="https://api.openai.com/v1"
)

TEST_PROMPTS = [
    "请总结一下企业知识库中合同审批流程的关键步骤。",
    "帮我生成一个Python函数，用于校验手机号格式。",
    "如果客户投诉物流延迟，客服Agent应该如何处理？",
    "请分析这个SQL查询为什么慢，并给出优化建议。",
    "请根据用户需求制定一个三步执行计划。"
]

async def run_one(prompt):
    start = time.time()
    first_token_time = None
    output_text = ""

    stream = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "你是一个高效、准确的AI Agent助手。"},
            {"role": "user", "content": prompt}
        ],
        temperature=0.2,
        stream=True
    )

    async for chunk in stream:
        now = time.time()
        if first_token_time is None:
            first_token_time = now - start
        delta = chunk.choices[0].delta.content
        if delta:
            output_text += delta

    total_time = time.time() - start
    return {
        "prompt": prompt,
        "ttft": first_token_time,
        "latency": total_time,
        "output_len": len(output_text)
    }

async def main():
    results = []
    for prompt in TEST_PROMPTS:
        result = await run_one(prompt)
        results.append(result)
        print(result)

    print("\n=== Summary ===")
    print("Avg TTFT:", statistics.mean([r["ttft"] for r in results]))
    print("Avg Latency:", statistics.mean([r["latency"] for r in results]))

if __name__ == "__main__":
    asyncio.run(main())

运行命令：

python benchmark.py

如果你希望测试并发性能，可以改成并发执行：

async def main():
    tasks = [run_one(prompt) for prompt in TEST_PROMPTS * 5]
    results = await asyncio.gather(*tasks)

    print("\n=== Summary ===")
    print("Requests:", len(results))
    print("Avg TTFT:", statistics.mean([r["ttft"] for r in results]))
    print("Avg Latency:", statistics.mean([r["latency"] for r in results]))

三、模型选择：不要所有任务都用最强模型

很多 Agent 性能差、成本高，本质原因是 所有步骤都调用大模型。

事实上，一个完整 Agent 通常包含不同子任务：

1. 意图识别；
2. 参数抽取；
3. 任务规划；
4. 工具选择；
5. 知识总结；
6. 最终回答；
7. 结果校验。

这些任务不一定都需要最强模型。

推荐策略是：

例如：

MODEL_CONFIG = {
    "router": "gpt-4o-mini",
    "extractor": "gpt-4o-mini",
    "planner": "gpt-4o",
    "executor": "gpt-4o-mini",
    "final_answer": "gpt-4o"
}

实际调用时：

async def call_llm(task_type, messages):
    model = MODEL_CONFIG.get(task_type, "gpt-4o-mini")

    response = await client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=0.2
    )

    return response.choices[0].message.content

这样做的好处是：

• 简单任务使用便宜快速的小模型；
• 复杂任务只在必要时使用强模型；
• 降低平均延迟；
• 降低整体调用成本。

四、Prompt 优化：越短越稳定

Agent 的 Prompt 不是越详细越好。很多团队为了让 Agent “更聪明”，把大量业务规则、示例、历史对话、工具说明全部塞进系统提示词，最后导致：

• 输入 token 爆炸；
• 模型注意力分散；
• 工具调用不稳定；
• 推理速度变慢；
• 成本变高。

1. 精简系统提示词

不推荐：

你是一个非常聪明、非常严谨、非常耐心、非常专业的AI助手……
你需要在所有场景中尽可能帮助用户……
以下是公司背景……
以下是所有业务流程……
以下是所有部门说明……
以下是所有工具说明……

推荐：

你是企业内部任务执行Agent。
目标：准确理解用户任务，必要时调用工具完成任务。
规则：
1. 不确定时先澄清；
2. 能直接回答则不调用工具；
3. 调用工具前必须确认所需参数；
4. 最终回答要简洁，并说明关键依据。

2. 将长规则模块化

不要一次性塞入所有规则，而是按需加载。

示例目录：

mkdir -p prompts
touch prompts/base.md
touch prompts/sql_agent.md
touch prompts/customer_service.md
touch prompts/code_agent.md

基础 Prompt：

你是一个企业AI Agent，负责理解用户意图并完成任务。

通用规则：
- 优先给出准确、可执行的答案；
- 信息不足时先提问；
- 不编造工具结果；
- 工具调用失败时说明原因并给出替代方案。

按业务加载：

def load_prompt(agent_type):
    with open("prompts/base.md", "r", encoding="utf-8") as f:
        base = f.read()

    path = f"prompts/{agent_type}.md"
    try:
        with open(path, "r", encoding="utf-8") as f:
            extra = f.read()
    except FileNotFoundError:
        extra = ""

    return base + "\n\n" + extra

这样可以避免所有任务都携带无关规则。

五、减少上下文：Token 越少，速度越快

模型推理延迟与输入输出 token 数高度相关。优化 Agent 性能，最直接的方法就是减少 token。

1. 统计 token

安装：

pip install tiktoken

统计脚本：

import tiktoken

def count_tokens(text, model="gpt-4o-mini"):
    enc = tiktoken.encoding_for_model(model)
    return len(enc.encode(text))

text = "这是一个AI Agent性能优化教程。"
print(count_tokens(text))

2. 多轮对话摘要

不要把完整历史对话无限追加到上下文中，应使用摘要记忆。

示例：

async def summarize_history(history):
    messages = [
        {
            "role": "system",
            "content": "请将以下对话压缩为简洁摘要，保留用户目标、关键约束、已完成事项和待办事项。"
        },
        {
            "role": "user",
            "content": str(history)
        }
    ]

    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        temperature=0
    )

    return response.choices[0].message.content

构造上下文时：

def build_context(summary, recent_messages):
    messages = [
        {"role": "system", "content": "你是一个高效的AI Agent。"}
    ]

    if summary:
        messages.append({
            "role": "system",
            "content": f"历史摘要：{summary}"
        })

    messages.extend(recent_messages[-6:])
    return messages

建议：

• 最近 3～6 轮对话保留原文；
• 更早内容转为摘要；
• 摘要只保留任务相关信息；
• 用户偏好、长期记忆单独存储。

六、RAG 检索优化：减少无效上下文

很多 Agent 会连接知识库。如果 RAG 做不好，模型会收到大量无关内容，既慢又容易答错。

1. 文档切块优化

切块太大：召回内容冗余，token 过多。
切块太小：语义不完整，答案缺上下文。

推荐参数：

chunk_size: 500 ~ 1000 中文字符
chunk_overlap: 50 ~ 150 中文字符

如果使用 LangChain：

pip install langchain langchain-community langchain-openai faiss-cpu

示例切块：

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=100,
    separators=["\n\n", "\n", "。", "！", "？", "；", "，", " "]
)

docs = splitter.split_text(raw_text)

2. 使用向量库 FAISS

pip install faiss-cpu

构建索引：

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    api_key="YOUR_API_KEY"
)

vectorstore = FAISS.from_texts(
    texts=docs,
    embedding=embeddings
)

vectorstore.save_local("faiss_index")

加载索引：

vectorstore = FAISS.load_local(
    "faiss_index",
    embeddings,
    allow_dangerous_deserialization=True
)

检索：

retriever = vectorstore.as_retriever(
    search_kwargs={"k": 5}
)

results = retriever.invoke("合同审批需要哪些步骤？")
for doc in results:
    print(doc.page_content)

3. 增加重排 Rerank

初次召回可以取更多，例如 k=20，然后用重排模型筛选前 3～5 条。

伪代码：

candidates = retriever.invoke(query, k=20)
reranked = rerank(query, candidates)
top_docs = reranked[:5]

如果没有专门 rerank 模型，也可以用 LLM 简单筛选：

async def filter_docs(query, docs):
    content = "\n\n".join([
        f"[{i}] {doc.page_content}" for i, doc in enumerate(docs)
    ])

    messages = [
        {
            "role": "system",
            "content": "你是知识库检索结果筛选器。请选出与用户问题最相关的文档编号，最多5个。只返回编号数组。"
        },
        {
            "role": "user",
            "content": f"问题：{query}\n\n候选文档：\n{content}"
        }
    ]

    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        temperature=0
    )

    return response.choices[0].message.content

七、工具调用优化：减少 Agent 无效循环

Agent 的强大之处在于能调用工具，但工具调用也是常见性能杀手。

1. 工具描述要清晰

错误示例：

{
  "name": "search",
  "description": "搜索信息"
}

模型不知道何时调用、查什么、返回什么，很容易乱用。

推荐：

{
  "name": "search_policy",
  "description": "当用户询问公司制度、流程、审批规则、报销政策等内部知识时调用。不要用于闲聊、代码问题或通用常识问题。",
  "parameters": {
    "type": "object",
    "properties": {
      "query": {
        "type": "string",
        "description": "需要检索的具体问题，必须包含业务对象和用户关注点"
      }
    },
    "required": ["query"]
  }
}

2. 限制最大调用轮数

不要让 Agent 无限循环。

MAX_ITERATIONS = 5

async def run_agent(user_input):
    iterations = 0

    while iterations < MAX_ITERATIONS:
        iterations += 1

        result = await agent_step(user_input)

        if result["type"] == "final":
            return result["content"]

        if result["type"] == "tool_call":
            tool_result = await call_tool(result["tool"], result["args"])
            user_input += f"\n工具结果：{tool_result}"

    return "任务执行轮数已达到上限，请补充更明确的信息或缩小任务范围。"

3. 能规则判断就不要调用模型

例如用户只是问订单状态，完全可以通过正则抽取订单号，然后直接查接口。

import re

def extract_order_id(text):
    match = re.search(r"订单号[:：]?\s*([A-Za-z0-9\-]+)", text)
    return match.group(1) if match else None

user_input = "帮我查一下订单号 A20250101001 的物流状态"
order_id = extract_order_id(user_input)

if order_id:
    result = query_order_api(order_id)
else:
    result = call_llm_for_missing_info(user_input)

八、缓存优化：相同问题不要重复算

缓存是 Agent 性能优化中收益非常高的一项。

适合缓存的内容：

• Embedding 结果；
• RAG 检索结果；
• 工具 API 返回；
• Prompt 模板；
• 高频问答；
• 用户权限信息；
• 数据库元信息；
• Agent 中间规划结果。

1. 使用 Redis 缓存

安装 Redis：

docker run -d \
  --name redis-agent-cache \
  -p 6379:6379 \
  redis:7

安装 Python 客户端：

pip install redis

缓存示例：

import redis
import json
import hashlib

r = redis.Redis(host="localhost", port=6379, db=0)

def make_cache_key(prefix, text):
    h = hashlib.sha256(text.encode("utf-8")).hexdigest()
    return f"{prefix}:{h}"

def get_cache(prefix, text):
    key = make_cache_key(prefix, text)
    value = r.get(key)
    if value:
        return json.loads(value)
    return None

def set_cache(prefix, text, value, ttl=3600):
    key = make_cache_key(prefix, text)
    r.setex(key, ttl, json.dumps(value, ensure_ascii=False))

使用：

query = "合同审批流程是什么？"

cached = get_cache("rag", query)
if cached:
    print("命中缓存")
    docs = cached
else:
    docs = search_knowledge_base(query)
    set_cache("rag", query, docs, ttl=3600)

2. 语义缓存

普通缓存要求文本完全一致，但用户可能换一种说法：

合同审批流程是什么？
合同怎么审批？
公司合同审批要经过哪些步骤？

这些问题语义相近，可以使用语义缓存：先对问题做 embedding，再找相似历史问题。如果相似度超过阈值，直接复用答案。

阈值建议：

0.88 ~ 0.95

阈值太低容易误命中，阈值太高命中率低。

九、并发与异步：不要串行等待

很多 Agent 会同时做多件事：

• 检索知识库；
• 查询用户权限；
• 获取业务数据；
• 调用外部 API；
• 加载用户记忆。

如果这些操作串行执行，延迟会叠加。应尽量异步并发。

串行写法

profile = await get_user_profile(user_id)
memory = await get_user_memory(user_id)
docs = await search_docs(query)
permission = await check_permission(user_id)

总耗时约等于所有任务耗时之和。

并发写法

profile, memory, docs, permission = await asyncio.gather(
    get_user_profile(user_id),
    get_user_memory(user_id),
    search_docs(query),
    check_permission(user_id)
)

总耗时约等于最慢的那个任务。

设置超时

async def with_timeout(coro, timeout=3):
    try:
        return await asyncio.wait_for(coro, timeout=timeout)
    except asyncio.TimeoutError:
        return None

使用：

docs = await with_timeout(search_docs(query), timeout=2)

十、流式输出：降低用户感知延迟

即使整体耗时不变，流式输出也能显著改善体验。

async def stream_answer(messages):
    stream = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        temperature=0.2,
        stream=True
    )

    async for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            print(delta, end="", flush=True)

如果你使用 FastAPI，可以这样返回 SSE。

安装：

pip install fastapi uvicorn openai

创建 app.py：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI(api_key="YOUR_API_KEY")

async def generate_stream(prompt):
    stream = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "你是一个高效的AI Agent。"},
            {"role": "user", "content": prompt}
        ],
        stream=True
    )

    async for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            yield delta

@app.get("/chat")
async def chat(prompt: str):
    return StreamingResponse(
        generate_stream(prompt),
        media_type="text/plain"
    )

启动：

uvicorn app:app --host 0.0.0.0 --port 8000

测试：

curl "http://localhost:8000/chat?prompt=请介绍AI%20Agent性能优化方法"

十一、结构化输出：减少反复修正

Agent 经常需要输出 JSON，用于后续工具调用。如果 JSON 格式不稳定，会导致解析失败，再次调用模型修复，浪费性能。

推荐明确格式：

请只输出 JSON，不要输出 Markdown，不要解释。
格式如下：
{
  "intent": "search_policy | query_order | chat | code_help",
  "confidence": 0.0,
  "args": {}
}

解析示例：

import json

def parse_json_output(text):
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return None

更好的方式是使用结构化输出或函数调用，让模型直接按 schema 返回。

示例工具定义：

tools = [
    {
        "type": "function",
        "function": {
            "name": "query_order",
            "description": "查询订单物流状态",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "订单号"
                    }
                },
                "required": ["order_id"]
            }
        }
    }
]

十二、日志与链路追踪：找到真正慢点

没有日志，就无法知道 Agent 到底慢在哪里。

建议每次请求记录：

{
  "request_id": "xxx",
  "user_id": "u001",
  "model": "gpt-4o-mini",
  "input_tokens": 1200,
  "output_tokens": 300,
  "tool_calls": 2,
  "rag_latency_ms": 180,
  "llm_latency_ms": 1200,
  "total_latency_ms": 1600,
  "success": true
}

简单 Python 记录方式：

import time
import uuid
import logging

logging.basicConfig(
    filename="agent.log",
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s"
)

async def handle_request(user_id, query):
    request_id = str(uuid.uuid4())
    start = time.time()

    try:
        result = await run_agent(query)
        success = True
    except Exception as e:
        result = str(e)
        success = False

    latency_ms = int((time.time() - start) * 1000)

    logging.info({
        "request_id": request_id,
        "user_id": user_id,
        "latency_ms": latency_ms,
        "success": success
    })

    return result

查看日志：

tail -f agent.log

统计慢请求：

grep "latency_ms" agent.log | tail -n 100

十三、部署优化：用 Docker 和 Gunicorn 提升稳定性

如果使用 FastAPI 部署 Agent 服务，推荐使用 Gunicorn + Uvicorn Worker。

安装：

pip install fastapi uvicorn gunicorn

启动命令：

gunicorn app:app \
  -k uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000 \
  --workers 4 \
  --timeout 120

如果服务器 CPU 核心数为 4，可以设置 4 个 worker；如果 Agent 主要是 I/O 密集型，也可以适当增加 worker。

查看 CPU 核心：

nproc

使用 Dockerfile：

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["gunicorn", "app:app", "-k", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000", "--workers", "4", "--timeout", "120"]

构建镜像：

docker build -t ai-agent-service:latest .

运行容器：

docker run -d \
  --name ai-agent-service \
  -p 8000:8000 \
  -e OPENAI_API_KEY="YOUR_API_KEY" \
  ai-agent-service:latest

查看日志：

docker logs -f ai-agent-service

十四、完整优化清单

上线前可以按以下清单检查：

• [ ] 是否统计 TTFT、总延迟、Token、工具调用次数？
• [ ] 是否区分简单任务和复杂任务，使用不同模型？
• [ ] Prompt 是否精简且模块化？
• [ ] 是否限制最大 Agent 循环轮数？
• [ ] 是否对长对话做摘要？
• [ ] RAG 是否做合理切块？
• [ ] 检索结果是否限制数量并做重排？
• [ ] 高频问题是否使用缓存？
• [ ] 外部 API 是否设置超时？
• [ ] 多个 I/O 操作是否并发执行？
• [ ] 是否开启流式输出？
• [ ] JSON 输出是否结构化？
• [ ] 是否记录完整链路日志？
• [ ] 是否设置服务 worker、超时和重试策略？

十五、推荐优化优先级

如果你刚开始优化，不建议一次性改所有东西。可以按收益从高到低执行：

1. 开启流式输出：最快改善用户体验；
2. 减少 Prompt 和上下文 token：直接降低延迟和成本；
3. 限制 Agent 迭代次数：防止无效循环；
4. 优化 RAG 切块和 top-k：减少无关内容；
5. 增加缓存：对高频问题收益极高；
6. 异步并发工具调用：降低复杂任务耗时；
7. 模型分层调用：降低整体成本；
8. 日志追踪：持续发现瓶颈；
9. 结构化输出：减少解析失败和二次修复；
10. 部署层扩容与限流：保障高并发稳定性。

结语

AI Agent 性能优化不是单纯“换更快的模型”，而是一个系统工程。真正高性能的 Agent，通常具备以下特征：

• 输入上下文短而准；
• 工具调用少而必要；
• 检索结果相关性高；
• 简单任务用小模型；
• 复杂任务才用强模型；
• 多个外部调用并发执行；
• 高频结果可缓存；
• 全链路有监控和日志；
• 出错后能降级处理。

只要按照本文的方法逐步优化，你的 AI Agent 通常可以在 响应速度、调用成本、稳定性和用户体验 上获得明显提升。对于生产环境来说，建议先建立性能基线，再逐项优化，每次只改一个变量，持续观察指标变化，这样才能真正把 Agent 调到稳定、快速、低成本的状态。