AI Agent实战指南：从工具调用到自主智能体

作者: 醉月思发布日期: 2026-02-13 分类: 技术实践

前言：Agent 不是更好的聊天机器人

很多人对 Agent 的理解停留在"会调工具的 LLM"，但真正做下来才发现：Agent 的核心难点不是调工具，而是任务规划、状态管理、错误恢复这一堆工程问题。

简单对话的边界很容易摸到——模型再聪明，没有手脚也只能给你写备忘录。Agent 给模型装上了"手脚"和"记忆"，让它能真正完成任务。但代价是工程复杂度成倍上升。

一、Agent 的核心组成

一个完整的 Agent 系统包含四个核心模块：

graph LR A[用户请求] --> B[意图理解] B --> C[任务规划] C --> D[工具执行] D --> E[结果整合] E --> F[响应生成] C --> G[记忆管理] G --> C D --> H[错误处理] H --> C

模块	职责	关键问题
意图理解	解析用户到底想要什么	歧义、隐含需求
任务规划	把目标拆成可执行步骤	线性 vs 分层、依赖关系
工具执行	调用外部能力	工具设计、错误处理
记忆管理	管理上下文和历史	短期 vs 长期、上下文窗口

二、从 API 调用到工具调用

2.1 基础 API 调用

最开始的 LLM 应用就是一行 API 调用：

from openai import OpenAI

client = OpenAI(api_key="sk-xxx")

def chat(message):
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "你是一个有用的助手。"},
            {"role": "user", "content": message}
        ]
    )
    return response.choices[0].message.content

这个版本能对话，但问题很多：

没记忆，每次对话都是独立的
没结构化输出，想提取内容还得自己解析
没错误处理，网络异常、超时、限流都是直接挂
没成本控制，用户聊多了 Token 费用不可控

2.2 加上记忆：滑动窗口

要让模型"记住"之前的对话，最简单的做法是把历史消息都塞进去：

from collections import deque

class ChatSessionWithWindow:
    def __init__(self, window_size=6):
        self.messages = deque(maxlen=window_size)
        self.messages.append(
            {"role": "system", "content": "你是一个有用的助手。"}
        )

    def chat(self, message):
        self.messages.append({"role": "user", "content": message})
        response = client.chat.completions.create(
            model="gpt-4",
            messages=list(self.messages)
        )
        answer = response.choices[0].message.content
        self.messages.append({"role": "assistant", "content": answer})
        return answer

2.3 分层记忆策略

真正生产环境用的记忆策略要分层：

class MemoryAssistant:
    def __init__(self):
        self.system_prompt = "你是一个专业的编程助手。"
        self.long_term_memory = {}  # key: user_id, value: summary
        self.conversation_history = {}  # key: session_id, value: deque

    def get_system_messages(self, user_id):
        messages = [{"role": "system", "content": self.system_prompt}]
        if user_id in self.long_term_memory:
            messages.append({
                "role": "system",
                "content": f"用户背景：{self.long_term_memory[user_id]}"
            })
        return messages

三种记忆的实践：

短期记忆：当前对话内的上下文，LangChain 自动管理
长期记忆：跨对话的持久化记忆，用向量数据库存储
上下文记忆：当前任务的临时状态，用 Python 对象管理

三、工具调用：Agent 的"手脚"

3.1 Function Calling 基础

2023 年底 OpenAI 推出 Function Calling（后来改名成 Tools），才算有了标准方案。

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "获取指定地点的当前天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "城市名称，例如：北京、上海"
                    }
                },
                "required": ["location"]
            }
        }
    }
]

response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "北京今天天气怎么样？"}],
    tools=tools
)

tool_calls = response.choices[0].message.tool_calls

3.2 工具设计的三个坑

坑一：工具粒度过细

# 错误示例：工具太细碎
tools = [
    Tool(name="ReadFile", ...),
    Tool(name="WriteFile", ...),
    Tool(name="DeleteFile", ...),
    Tool(name="MoveFile", ...),
    Tool(name="CopyFile", ...),
]

结果模型在规划任务时会过度纠结，把简单任务拆得很复杂。

坑二：工具描述太泛

# 错误：描述太宽泛
Tool(
    name="ReadFile",
    func=read_file,
    description="读取本地文件内容，输入文件路径"
)

# 正确：描述精确
Tool(
    name="ReadFile",
    func=read_file,
    description="读取指定的本地文件内容。必须提供完整、有效的文件路径。只能读取文本文件，不能浏览目录或执行文件操作。"
)

坑三：工具输出格式不统一

class ToolResult:
    def __init__(self, success, data, error=None):
        self.success = success
        self.data = data
        self.error = error

def read_file(filepath):
    try:
        with open(filepath, 'r') as f:
            content = f.read()
        return ToolResult(True, {'content': content})
    except Exception as e:
        return ToolResult(False, None, error=str(e))

3.3 工具执行器封装

实际项目里推荐用统一的执行器：

from concurrent.futures import ThreadPoolExecutor

class ToolExecutor:
    def __init__(self):
        self.tools = {}

    def register(self, name: str, description: str, parameters: dict):
        def decorator(func):
            self.tools[name] = {
                "function": func,
                "description": description,
                "parameters": parameters
            }
            return func
        return decorator

    def execute(self, tool_call) -> str:
        name = tool_call.function.name
        args = json.loads(tool_call.function.arguments)

        try:
            result = self.tools[name]["function"](**args)
            return json.dumps({"success": True, "result": result})
        except Exception as e:
            return json.dumps({"success": False, "error": str(e)})

四、任务规划：Agent 的"大脑"

4.1 线性规划的局限

最开始的 Agent 都是线性规划——模型看到任务，直接输出一系列步骤。简单场景够用，复杂任务就撑不住。

比如：“分析 CSV 文件，根据销售额排名生成报表，邮件发送给团队”

1. 读取 CSV 文件
2. 计算销售额排名
3. 生成图表
4. 发送邮件

问题：第 2 步失败怎么办？CSV 格式不对怎么办？邮件发送失败要不要重试？

4.2 分层规划

更稳的做法是把任务拆成三层：

class TaskPlanner:
    def plan(self, user_request):
        # 第一层：理解目标
        goal = self.understand_goal(user_request)

        # 第二层：拆分子任务
        subtasks = self.breakdown_goal(goal)

        # 第三层：为每个子任务规划具体步骤
        execution_plans = {}
        for subtask in subtasks:
            execution_plans[subtask.id] = self.plan_execution(subtask)

        return {
            'goal': goal,
            'subtasks': subtasks,
            'execution_plans': execution_plans,
            'dependencies': self.analyze_dependencies(subtasks)
        }

分层规划的好处：

可观察性：你知道模型在哪个层次上思考
可干预：子任务失败时，可以局部调整而不影响整体
可扩展：新类型任务只需要扩展对应的层次

4.3 AutoGPT 的陷阱

AutoGPT 的"永不放弃"精神看起来很励志，但实际操作中就是资源浪费。

典型问题：执行某个子任务时遇到依赖冲突，AutoGPT 尝试了几次失败后开始循环尝试不同方案，20 多分钟后消耗大量 API 调用，最后依然卡住。

解决方案：加合理的停止条件

MAX_TASK_ATTEMPTS = 3
MAX_TOTAL_STEPS = 20
STEP_TIMEOUT_SECONDS = 60

class TaskExecutor:
    def should_continue(self, task_id):
        # 单个任务尝试次数检查
        if self.task_attempts.get(task_id, 0) >= MAX_TASK_ATTEMPTS:
            return False, f"任务 {task_id} 已达到最大尝试次数"

        # 总步骤数检查
        if self.total_steps >= MAX_TOTAL_STEPS:
            return False, f"已达到最大步骤数"

        # 超时检查
        if time.time() - self.last_reset_time > STEP_TIMEOUT_SECONDS:
            return False, f"任务执行超时"

        return True, ""

核心教训：不要高估模型的自主能力。 给模型太多自由度反而会降低可靠性。

五、LangChain Agent 实践

5.1 LangChain Agent 基础

from langchain.tools import Tool
from langchain.agents import initialize_agent, AgentType
from langchain.chat_models import ChatOpenAI

# 定义工具
def get_github_latest_release(repo_name):
    import requests
    url = f"https://api.github.com/repos/{repo_name}/releases/latest"
    response = requests.get(url)
    return response.json().get('tag_name', 'Unknown')

tools = [
    Tool(
        name="GitHubLatestRelease",
        func=get_github_latest_release,
        description="获取 GitHub 仓库的最新版本号，输入格式为 'owner/repo'"
    )
]

# 初始化 Agent
llm = ChatOpenAI(model="gpt-4", temperature=0)
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

result = agent.run("LangChain 的最新版本是多少？")

5.2 从 LangChain 到 LangGraph

LangChain 的 Agent 在状态管理上经常出问题。LangGraph 提供了 StateGraph，更适合复杂状态流。

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
import operator

class AgentState(TypedDict):
    messages: Annotated[List, operator.add]
    retrieved_docs: List[str]
    current_step: int
    max_steps: int
    tool_calls: List[str]
    final_answer: str

# 构建工作流
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("intent_recognition", intent_recognition)
workflow.add_node("smart_retrieval", smart_retrieval)
workflow.add_node("tool_execution", tool_execution)
workflow.add_node("answer_generation", answer_generation)
workflow.add_node("satisfaction_check", satisfaction_check)

# 添加边
workflow.set_entry_point("intent_recognition")
workflow.add_edge("intent_recognition", "smart_retrieval")
workflow.add_edge("smart_retrieval", "tool_execution")
workflow.add_edge("tool_execution", "answer_generation")
workflow.add_edge("answer_generation", "satisfaction_check")

# 条件边：根据满意度决定是否重新检索
workflow.add_conditional_edges(
    "satisfaction_check",
    lambda x: "smart_retrieval" if x.get("status") != "completed" else END,
    {
        "smart_retrieval": "smart_retrieval",
        END: END
    }
)

app = workflow.compile()

LangGraph 的优势：

状态显式定义，不会丢失
节点间的数据流清晰
支持条件分支和循环
便于调试和定位问题

六、Agentic RAG：让 RAG 会思考

传统 RAG 像个死板的搜索引擎——你问什么，它就按固定套路检索什么。Agentic RAG 让 RAG 能像人一样做决策和规划。

6.1 核心能力

动态检索策略：根据问题复杂度选择不同检索方式
多步推理：复杂问题分解成子问题
工具调用：能调用外部工具（数据库、API、计算器）
自我反思：结果不满意时调整策略

6.2 意图识别节点

def intent_recognition(state: AgentState):
    last_message = state["messages"][-1]
    response = llm.invoke(intent_prompt.format(question=last_message.content))
    intent = json.loads(response.content)

    state["current_step"] = 0
    state["max_steps"] = 5 if intent["need_steps"] else 1

    return {
        **state,
        "strategy": intent["strategy"]  # vector_search / hybrid_search / multi_step
    }

6.3 满意度评估节点

def satisfaction_check(state: AgentState):
    answer = state["final_answer"]
    score = evaluate_answer(answer)

    if score < 3 and state["current_step"] < state["max_steps"]:
        # 不满意，换策略重新检索
        state["current_step"] += 1
        return {**state, "strategy": "hybrid_search"}
    else:
        return {**state, "status": "completed"}

6.4 Agentic RAG 的效果

上线两个月的实测数据：

准确率从 65% 提升到 82%
复杂问题解决率从 40% 提升到 75%
用户满意度从 3.2 分提升到 4.1 分（满分 5 分）

七、NDJSON：Agent 的流式协议

7.1 为什么 Agent 离不开 NDJSON

Agent 系统的核心特征：长时间运行 + 多阶段事件 + 可观测。

如果所有中间态塞进一个最终 JSON，会有几个现实问题：

交互体验差：用户要干等几十秒才看到回复
进度不可见：工具调用、报错来不及展示
失败难定位：任务中途挂了，手里没结构化轨迹
桥接层难写：IM、Web UI、日志系统都想边收边处理

NDJSON（Newline Delimited JSON）刚好卡住这个痛点：一行一条记录，每行本身必须是合法 JSON。

{"ts":"2026-07-20T10:00:01+08:00","type":"message","text":"帮我查一下明天日程"}
{"ts":"2026-07-20T10:00:02+08:00","type":"tool_call","name":"calendar.search"}
{"ts":"2026-07-20T10:00:03+08:00","type":"tool_result","ok":true,"count":2}
{"ts":"2026-07-20T10:00:04+08:00","type":"final","text":"明天有两场会"}

7.2 事件协议设计

用 type 做事件协议，而不是一个大而全 schema：

import json
import sys

def emit(obj: dict) -> None:
    sys.stdout.write(json.dumps(obj, ensure_ascii=False) + "\n")
    sys.stdout.flush()  # 很关键，不 flush 对端可能看不到事件

# 发送不同类型的事件
emit({"type": "ready", "session_id": "s_01"})
emit({"type": "text_delta", "delta": "正在"})
emit({"type": "text_delta", "delta": "查询"})
emit({"type": "final", "text": "正在查询日程。"})

增量文本用 delta，最终再用一条 final 兜底。

这样 UI 可以边拼边显示，存储层又不用猜"现在拼到哪了"。

7.3 坏行隔离

生产环境的事件流难免脏，要做行级容错：

import json
import logging

logger = logging.getLogger(__name__)

def iter_events(fp):
    for lineno, raw in enumerate(fp, 1):
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        try:
            yield json.loads(line)
        except json.JSONDecodeError:
            logger.warning("skip bad ndjson line=%s content=%r", lineno, line[:200])

JSON 数组中间坏一块，常常整包都解析失败；NDJSON 可以天然做行级容错。

7.4 Unix 工具友好

NDJSON 的一大优势：用普通 Unix 工具就能抽样、过滤、排障。

# 看最后 20 条事件
tail -n 20 run.ndjson

# 只看工具调用
rg '"type":"tool_call"' run.ndjson

# 统计事件类型分布
rg -o '"type":"[^"]+"' run.ndjson | sort | uniq -c | sort -nr

# 抽 100 条做本地复现
head -n 100 run.ndjson > sample.ndjson

一次失败的 run 留下一份 NDJSON 轨迹，比只留最终报错更值钱。

八、错误处理：让 Agent 知道何时求助

8.1 主动错误处理

不要等错误发生后再处理，要在规划阶段就预想可能的错误：

def safe_tool_call(tool_name, tool_func, **kwargs):
    try:
        result = tool_func(**kwargs)
        if not result.success:
            return {
                'status': 'error',
                'error_type': 'tool_execution_error',
                'message': f"工具 {tool_name} 执行失败: {result.error}"
            }
        return {'status': 'success', 'data': result.data}
    except Exception as e:
        return {
            'status': 'error',
            'error_type': 'unexpected_error',
            'message': f"工具 {tool_name} 遇到意外错误: {str(e)}"
        }

8.2 人工干预策略

有些错误模型自己解决不了，需要人工干预：

class HumanInterventionHandler:
    def __init__(self):
        self.intervention_threshold = 3  # 同类错误 3 次后请求人工干预

    def handle_error(self, error_type, error_context, error_count):
        if error_count >= self.intervention_threshold:
            return self.request_human_help(error_type, error_context)
        else:
            return self.suggest_recovery(error_type, error_context)

    def suggest_recovery(self, error_type, error_context):
        recovery_strategies = {
            'tool_execution_error': [
                '检查工具参数是否正确',
                '验证工具依赖环境是否正常',
                '尝试使用备用工具'
            ],
            'planning_error': [
                '重新评估任务目标',
                '增加子任务的颗粒度',
                '补充必要的任务步骤'
            ]
        }
        return {
            'action': 'retry_with_suggestion',
            'suggestions': recovery_strategies.get(error_type, ['重新尝试'])
        }

8.3 错误重试策略

模型调用失败是常态，重试策略要 carefully 设计：

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_model_with_retry(prompt):
    return llm.predict(prompt)

注意：不是所有错误都应该重试。参数错误不应该重试，配额不足需要特殊处理。

九、成本控制

9.1 成本爆炸的常见原因

工具调用循环：模型陷入死循环，反复调用工具
上下文膨胀：对话历史塞太多，每次调用 Token 都很多
模型选择不当：简单任务也用大模型

9.2 控制策略

分级模型策略：

def get_model_for_task(complexity: str):
    if complexity == "simple":
        return "gpt-4o-mini"  # 简单问题用小模型
    elif complexity == "complex":
        return "gpt-4o"  # 复杂问题才用大模型

答案缓存：

from functools import lru_cache

@lru_cache(maxsize=100)
def get_cached_answer(question_hash: str) -> str:
    # 从缓存获取答案
    pass

# 在意图识别前先查缓存
question_hash = hashlib.md5(question.encode()).hexdigest()
cached = get_cached_answer(question_hash)
if cached:
    return cached

Token 计数限制：

import tiktoken

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

def chat_with_limit(message, max_tokens=1000):
    input_tokens = count_tokens(message)
    if input_tokens > max_tokens:
        return "问题太长，请简化后重试。"
    # ...

十、上下文窗口管理

10.1 上下文爆炸

模型上下文窗口有限。一次复杂任务中，Agent 处理多个大文件，每个几千行代码，默认记忆管理会把所有工具调用历史都塞进上下文，结果很快就爆了。

10.2 上下文压缩策略

class ContextAwareMemory:
    def __init__(self, max_tokens=4000):
        self.max_tokens = max_tokens
        self.conversation_history = []

    def add_message(self, role, content, importance=1.0):
        message = {'role': role, 'content': content, 'importance': importance}
        self.conversation_history.append(message)
        self._prune_if_needed()

    def _prune_if_needed(self):
        current_tokens = sum(len(msg['content']) for msg in self.conversation_history)
        if current_tokens > self.max_tokens:
            # 按重要性排序，移除低重要性的消息
            sorted_messages = sorted(
                self.conversation_history,
                key=lambda x: x['importance']
            )
            # 保留最重要的 80%
            keep_count = int(len(sorted_messages) * 0.8)
            self.conversation_history = sorted_messages[-keep_count:]

10.3 对话历史压缩

把连续的工具调用压缩成摘要：

def compress_conversation(messages):
    compressed = []
    i = 0
    while i < len(messages):
        if messages[i]['role'] == 'assistant' and i + 2 < len(messages):
            if messages[i+1]['role'] == 'tool' and messages[i+2]['role'] == 'assistant':
                compressed.append({
                    'role': 'assistant',
                    'content': f"[工具调用: {messages[i]['content'][:50]}... -> 回应: {messages[i+2]['content'][:50]}...]"
                })
                i += 3
                continue
        compressed.append(messages[i])
        i += 1
    return compressed

十一、实战经验总结

11.1 几个关键教训

折腾了一圈 Agent 开发，几个比较实在的体会：

不要高估模型的自主能力：给模型太多自由度反而降低可靠性。适度约束和人工干预是必要的。
工具设计比模型能力更重要：工具描述要精确但不过度限制；工具输入要严格验证；工具输出要统一格式；工具粒度要适中。
任务规划是核心能力：Agent 的智能程度主要体现在任务规划上，而不是单个工具的执行上。
错误处理要主动：不要等错误发生后再处理，要在规划阶段就预想可能的错误。
记忆管理要精简：区分长期和短期记忆，对内容做重要性标注，定期压缩历史。

11.2 任务复杂度与成功率

实际项目中的典型数据：

任务类型	成功率	主要失败原因
简单任务（“检查服务器状态”）	95%+	边缘情况
中等任务（“重启所有后端服务”）	~80%	边缘情况、参数错误
复杂任务（“诊断性能问题”）	~60%	需要人工干预

11.3 何时该用 Agent

适合用 Agent 的场景：

任务需要多步骤、多工具配合
任务有明确的子目标划分
错误可以检测和恢复

不适合用 Agent 的场景：

任务可以用确定性算法解决
任务对延迟极其敏感
任务错误成本极高

记住：不是所有任务都需要 Agent，有时候规则系统更简单可靠。

十二、写在最后

Agent 开发不是在造一个"能代替人"的系统，而是在造一个"能辅助人"的系统。它的价值不在于完全自动化，而在于把那些重复的、机械的任务自动化，让人能专注于需要判断和决策的部分。

如果你也在折腾 Agent 开发，建议：

从小处开始：先把一个简单场景做扎实
不要追求完全自主：适度的约束和人工干预是正常的
工具设计和任务规划比模型微调更重要
把重点放在可靠性和可观测性上，而不是炫酷的功能

Agent 开发还处在早期阶段，很多问题还没有标准答案。但这恰恰是有意思的地方——我们在和一个新兴的技术一起成长。

只要别让 Agent 把你的服务器删了就行。

本文整合了 7 篇 AI Agent 应用开发文章，涵盖工具调用、任务规划、LangChain/LangGraph、Agentic RAG、NDJSON 流式协议、错误处理、成本控制等核心技术。

版权声明: 本文首发于指尖魔法屋-AI Agent实战指南：从工具调用到自主智能体（https://blog.thinkmoon.cn/post/ai-agent-comprehensive-guide/）转载或引用必须申明原指尖魔法屋来源及源地址！

主流大模型实战对比指南：从Llama到DeepSeek的选型与部署

从 Llama、Qwen、DeepSeek 到 ChatGLM、Gemma、Mistral，系统梳理主流大模型的选型、部署与优化

📁 技术实践

📅 2026-02-25

NDJSON：Agent 开发离不开的流式数据格式

JSON 要等整包到齐才能读，Agent 却得边跑边吐状态。NDJSON 把这件事拆成一行一条记录，这篇文章讲清它是什么、什么时候该用，以及几处容易被忽略的用法。

📁 学习笔记

📅 2026-07-20

Harness Engineering 的介绍与编程实践

这篇文章从 AI 编程的真实工程场景出发，梳理 Harness Engineering 如何通过上下文、工具、沙盒、验证和人工闸门，让 Agent 从“会写代码”走向“能稳定交付”。