从零到一：30天AI Agent学习计划（前23天实录，含LangChain 1.x踩坑全记录）

从零到一：30天AI Agent学习计划（前23天实录，含LangChain 1.x踩坑全记录）

本文记录了一个 Java 后端开发者用 30 天系统学习 AI Agent 开发的全过程（前 23 天），涵盖 Python 速成、LLM API 接入、Prompt 工程、Tool Calling、LangChain 1.x 框架、多轮对话 Agent 等核心内容。

最大的价值：所有代码基于 LangChain 1.3.13 实测，记录了 1.x 版本中大量 API 变更和踩坑经验，帮你避开旧教程的陷阱。

一、为什么写这篇文章

网上 LangChain 教程很多，但大多基于 0.x 或 0.3.x 版本。当我按照教程写出 from langchain.agents import AgentExecutor 时，直接 ImportError——1.x 版本已经把一大半 API 删了。

这篇文章记录了我从零开始学 AI Agent 的 23 天历程，重点是LangChain 1.x 的真实可用 API 和踩过的坑，希望能帮到同样在学的人。

学习背景

• 后端：Java Spring Boot + Vue3（JeecgBoot v3.9.1）
• Python：零基础
• 目标：做一个 MES（制造执行系统）智能助手，能查工单、查库存、上报进度

二、30天学习计划总览

技术栈：阿里百炼（qwen-turbo）+ LangChain 1.3.13 + Python 3.14

三、第一阶段：Python 速成 + LLM API（Day 1-7）

3.1 Python 基础（Day 1-3）

3 天搞定 Python 核心语法：变量、列表、字典、函数、requests、json、装饰器、生成器、上下文管理器。

作为 Java 开发者，最大的感受是：Python 太简洁了。一个 HTTP 请求：

import requests

resp = requests.get("https://api.example.com/data")
data = resp.json() # 自动解析 JSON
print(data["result"])

对比 Java 的 HttpClient + ObjectMapper，省了至少 20 行。

3.2 接入 LLM API（Day 4-7）

选用阿里百炼（免费额度，国内首选），用 openai 标准库调用 qwen-turbo：

from openai import OpenAI

client = OpenAI(
api_key="你的百炼API Key",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

response = client.chat.completions.create(
model="qwen-turbo",
messages=[
{"role": "system", "content": "你是一个助手"},
{"role": "user", "content": "你好"},
],
)
print(response.choices[0].message.content)

关键概念：

• system：设定 AI 的角色和行为规则
• user：用户的输入
• assistant：AI 的回复

Day 7 产出了一个完整的命令行聊天程序，支持多轮对话、流式输出、重试机制。

四、第二阶段：Prompt 工程 + Tool Calling（Day 8-16）

4.1 Prompt 工程核心技巧（Day 8-11）

4.2 Tool Calling：让 AI 能"做事"（Day 12-16）

这是第二阶段的核心。Tool Calling 让 AI 不只是聊天，而是能调用外部工具（API、数据库、计算器等）。

定义工具：

from openai import OpenAI
import requests

client = OpenAI(
api_key="你的API Key",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

# 定义工具 schema
tools = [{
"type": "function",
"function": {
"name": "get_weather",
"description": "查询指定城市的天气",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名"},
},
"required": ["city"],
},
},
}]

# 第一轮：AI 决定调用工具
response = client.chat.completions.create(
model="qwen-turbo",
messages=[{"role": "user", "content": "北京天气怎么样？"}],
tools=tools,
)

# AI 返回 tool_calls
tool_call = response.choices[0].message.tool_calls[0]
import json
args = json.loads(tool_call.function.arguments) # {"city": "北京"}

# 执行工具
weather = requests.get(f"https://wttr.in/{args['city']}?format=3").text

# 第二轮：把工具结果喂给 AI
response = client.chat.completions.create(
model="qwen-turbo",
messages=[
{"role": "user", "content": "北京天气怎么样？"},
{"role": "assistant", "content": None, "tool_calls": response.choices[0].message.tool_calls},
{"role": "tool", "tool_call_id": tool_call.id, "content": weather},
],
tools=tools,
)
print(response.choices[0].message.content) # "北京目前晴，25度…"

Day 14-15 做了实战项目：把 JeecgBoot 后端的仓库库存接口封装成 Tool，AI 能通过自然语言查库存。

4.3 第二阶段产出

Day 16 整合了一个 agent_tools.py 工具集：8 个工具注册表 + 执行器 + 对话管理，用纯 openai 库实现了完整的 Tool Calling 循环。

五、第三阶段：LangChain 框架 + 记忆管理（Day 17-23）

这是本文的核心部分，也是踩坑最多的地方。

5.1 LangChain 基础：LCEL 管道符（Day 17-18）

LangChain Expression Language（LCEL）用管道符 | 串联组件：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(
model="qwen-turbo",
api_key="你的API Key",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

prompt = ChatPromptTemplate.from_template("用一句话解释{topic}")

# 管道符串联：提示词模板 | 模型 | 输出解析器
chain = prompt | model | StrOutputParser()

result = chain.invoke({"topic": "递归"})
print(result) # "递归是函数调用自身的编程技巧…"

PromptTemplate 进阶（Day 18）：

• MessagesPlaceholder：插入多轮对话历史
• partial：部分填充模板变量
• FewShotChatMessagePromptTemplate：少样本提示

5.2 对话记忆管理（Day 19）

⚠️ 踩坑 #1：ConversationBufferMemory 已移除

旧教程写的：

from langchain.memory import ConversationBufferMemory # ❌ 1.x 已移除！

1.x 正确写法：用 RunnableWithMessageHistory：

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory

# 会话存储
session_store = {}

def get_history(session_id: str) –> InMemoryChatMessageHistory:
if session_id not in session_store:
session_store[session_id] = InMemoryChatMessageHistory()
return session_store[session_id]

# 给 chain 包上记忆
chain_with_history = RunnableWithMessageHistory(
chain,
get_history,
input_messages_key="input",
history_messages_key="history",
)

# 调用时指定 session_id
result = chain_with_history.invoke(
{"input": "我叫程钧"},
config={"configurable": {"session_id": "user-001"}},
)
# 下一轮自动记住上文
result2 = chain_with_history.invoke(
{"input": "我叫什么名字？"},
config={"configurable": {"session_id": "user-001"}},
)
# "你叫程钧"

5.3 create_agent：现代版 Agent 循环（Day 20）

⚠️ 踩坑 #2：AgentExecutor / create_tool_calling_agent 已移除

旧教程写的：

from langchain.agents import create_tool_calling_agent, AgentExecutor # ❌ 1.x 已移除！

1.x 正确写法：用 create_agent：

from langchain.agents import create_agent
from langchain_core.tools import tool

# 用 @tool 装饰器定义工具（自动生成 schema）
@tool
def get_weather(city: str) –> str:
"""查询指定城市的天气。"""
return f"{city}今天晴，28度。"

@tool
def calculator(expression: str) –> str:
"""计算数学表达式。"""
return str(eval(expression))

# 一行创建 Agent（内置 Tool Calling 循环）
agent = create_agent(
model,
tools=[get_weather, calculator],
system_prompt="你是智能助手，能查天气和做计算。",
)

# 调用
result = agent.invoke({"messages": [("user", "北京天气怎么样？")]})
print(result["messages"][–1].content)
# "北京今天晴，28度。"

对比 Day 12-16 手写的 50 行 Tool Calling 循环，create_agent 一行搞定。

5.4 Agent 进阶控制（Day 21）

Checkpointer：持久多轮记忆

Day 20 多轮对话还要手动累积 messages。加上 checkpointer 后，框架自动记忆：

from langgraph.checkpoint.memory import InMemorySaver

saver = InMemorySaver()

agent = create_agent(
model,
tools=[get_weather],
system_prompt="天气助手",
checkpointer=saver, # ← 自动记忆
)

# 用 thread_id 区分不同会话
config_a = {"configurable": {"thread_id": "user-A"}}

# 第1轮
agent.invoke({"messages": [("user", "查昆明天气")]}, config=config_a)

# 第2轮：不报城市名，Agent 也能记住
result = agent.invoke(
{"messages": [("user", "我刚才查的是哪个城市？")]},
config=config_a,
)
print(result["messages"][–1].content) # "昆明"

# 新会话（不记得昆明）
config_b = {"configurable": {"thread_id": "user-B"}}
result = agent.invoke(
{"messages": [("user", "我刚才查的是哪个城市？")]},
config=config_b,
)
# "你还没有查询过城市" ← 会话隔离生效

Middleware：中间件扩展

⚠️ 踩坑 #3：@before_model / @after_model 必须装饰独立函数

错误写法（装饰类方法，不触发）：

class MyMiddleware:
@before_model # ❌ 不会触发！
def before(self, state, runtime):
print("模型调用前")

正确写法（装饰独立函数）：

from langchain.agents.middleware import (
before_model, after_model, AgentState, Runtime,
ToolCallLimitMiddleware,
)

@before_model
def log_before_model(state: AgentState, runtime: Runtime):
"""模型调用前：记录消息数"""
msg_count = len(state.get("messages", []))
print(f"[日志] 模型调用前，消息数: {msg_count}")

@after_model
def log_after_model(state: AgentState, runtime: Runtime):
"""模型调用后：记录完成"""
print("[日志] 模型调用完成")

# 内置限流中间件（防死循环）
limiter = ToolCallLimitMiddleware(
thread_limit=50, # 整个会话最多调 50 次工具
run_limit=15, # 单轮最多调 15 次工具
)

agent = create_agent(
model,
tools=[get_weather],
system_prompt="助手",
checkpointer=saver,
middleware=[log_before_model, log_after_model, limiter],
)

⚠️ 踩坑 #4：ToolCallLimitMiddleware 参数名

参数是 thread_limit / run_limit，不是 max_calls。写错直接 TypeError。

结构化输出

from pydantic import BaseModel, Field

class WeatherReport(BaseModel):
city: str = Field(description="城市名")
temperature: int = Field(description="温度（摄氏度）")
suggestion: str = Field(description="穿衣建议")

agent = create_agent(
model,
tools=[get_weather],
system_prompt="天气助手",
response_format=WeatherReport, # ← 强制返回结构化对象
)

result = agent.invoke({"messages": [("user", "北京天气")]})
# result["structured_response"] 是 WeatherReport 对象

注意：response_format 适合数据提取场景。聊天助手别加，会强制每轮返回 JSON，体验很差。

流式输出

⚠️ 踩坑 #5：stream_mode=“updates” 的 chunk 中 state 可能为 None

for chunk in agent.stream(
{"messages": [("user", "北京天气")]},
config=config,
stream_mode="updates",
):
for node, state in chunk.items():
if not state: # ← 必须判空！
continue
msgs = state.get("messages", [])
if msgs:
last = msgs[–1]
content = getattr(last, "content", "")
if content and not getattr(last, "tool_calls", None):
print(f"Agent: {content}")

5.5 自定义 Tool 开发深入（Day 22）

Pydantic args_schema：给 LLM 更清晰的参数描述

from pydantic import BaseModel, Field

class WorkOrderInput(BaseModel):
order_no: str = Field(description="工单编号，格式 WO-YYYY-NNN，如 WO-2024-001")

@tool("query_work_order", args_schema=WorkOrderInput)
def query_work_order(order_no: str) –> str:
"""查询MES工单详情，包括产品名称、数量、状态、进度。"""
# … 业务逻辑
return f"工单 {order_no}：产品轴承6204，数量500件，状态进行中"

Field(description=…) 的描述会流进工具的 JSON Schema，LLM 能看到并据此决定传什么参数。

InjectedToolCallId：自动注入回调 ID

from langchain_core.tools import tool, InjectedToolCallId
from typing import Annotated

@tool
def report_progress(
order_no: str,
process_seq: int,
progress: int,
tool_call_id: Annotated[str, InjectedToolCallId], # ← 自动注入，不暴露给 LLM
) –> str:
"""上报工序进度。"""
return f"已上报，操作ID: {tool_call_id[:8]}"

InjectedToolCallId 注入的参数不会出现在工具 schema 里，LLM 看不到也传不了，纯框架内部使用。

InjectedState：注入完整图状态

⚠️ 踩坑 #6：InjectedState 的导入位置

# ❌ 错误：不在 langchain_core.tools
# from langchain_core.tools import InjectedState

# ✅ 正确：从 langgraph.prebuilt 导入
from langgraph.prebuilt import InjectedState

@tool
def save_note(
content: str,
state: Annotated[dict, InjectedState], # ← 注入图状态
) –> str:
"""保存笔记，读取当前对话的消息数。"""
msg_count = len(state.get("messages", []))
return f"已保存（当前对话已有{msg_count}条消息）"

注意：InjectedState 注入的是图状态（含 messages 等），不是 config。thread_id 在 config 里，工具通过 InjectedState 读不到。

return_direct：跳过模型总结

@tool(return_direct=True)
def query_inventory(item_code: str) –> str:
"""查询物料库存。"""
return f"物料{item_code}：库存150件，A区-03货架"

# Agent 调用这个工具后，直接返回工具结果，不再让模型"总结"一遍

⚠️ 踩坑 #7（最重要）：ToolException 在 1.x 会崩 Agent

旧教程教你在工具里抛 ToolException 让模型自动重试：

from langchain_core.tools import ToolException

@tool
def divide(a: int, b: int) –> str:
"""除法运算。"""
if b == 0:
raise ToolException("除数不能为零") # ❌ 1.x 会崩掉整个 Agent！
return str(a / b)

原因：langgraph.prebuilt.tool_node._default_handle_tool_errors 默认是重抛异常，且 create_agent 没有暴露 handle_tool_errors 参数。

正确写法——工具内部 try/except 返回错误字符串（版本无关，永远有效）：

@tool
def divide(a: int, b: int) –> str:
"""除法运算。"""
try:
return str(a / b)
except Exception as e:
return f"计算失败：{e}。请换一个非零的除数。"

实测问"10除以0"，模型优雅回复：“除数不能为零，请更换一个非零的除数重新计算。” ✅

⚠️ 踩坑 #8：return_direct + 流式的交互

return_direct=True 的工具在流式模式下，结果作为 ToolMessage 出现在 tools 节点，不在 model 节点的 AI 消息里。如果直接过滤 ToolMessage，会丢失 return_direct 的结果。

解决方案：缓存工具结果，流结束后判断有无 AI 回复，没有就显示工具结果。

5.6 整合实战：MES 工单查询 Agent（Day 23）

把第三阶段学的全部整合成一个完整项目：

整合架构：
┌─────────────────────────────────────────────┐
│ 交互式 REPL 主循环 │
├─────────────────────────────────────────────┤
│ create_agent (Day 20) ← 自动 Tool Calling │
│ ├─ checkpointer (Day 21) ← 多轮记忆 │
│ ├─ middleware (Day 21) ← 日志 + 限流 │
│ ├─ stream (Day 21) ← 流式输出 │
│ └─ 6 个 @tool (Day 22) ← 业务工具 │
│ ├─ query_work_order (args_schema) │
│ ├─ query_process (Literal 枚举) │
│ ├─ report_progress (InjectedToolCallId)│
│ ├─ query_inventory (return_direct) │
│ ├─ check_quality (Pydantic + 筛选) │
│ └─ get_equipment (防御式 try/except) │
└─────────────────────────────────────────────┘

完整代码（核心部分）：

import os
import time
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool, InjectedToolCallId
from langchain.agents import create_agent
from langchain.agents.middleware import (
before_model, after_model, AgentState, Runtime,
ToolCallLimitMiddleware,
)
from langgraph.checkpoint.memory import InMemorySaver
from pydantic import BaseModel, Field
from typing import Annotated, Literal

# — 模型 —
model = ChatOpenAI(
model="qwen-turbo",
api_key=os.environ["DASHSCOPE_API_KEY"],
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

# — 工具1: 查询工单（Pydantic args_schema）—
class WorkOrderInput(BaseModel):
order_no: str = Field(description="工单编号，格式 WO-YYYY-NNN")

@tool("query_work_order", args_schema=WorkOrderInput)
def query_work_order(order_no: str) –> str:
"""查询MES工单详情。"""
try:
# 实际项目：requests.get("http://192.168.2.251:9091/jeecg-boot/mitt/…")
wo = WORK_ORDERS.get(order_no)
if not wo:
return f"未找到工单 {order_no}"
return f"工单{order_no}：产品{wo['product']}，数量{wo['qty']}件，进度{wo['progress']}%"
except Exception as e:
return f"查询出错：{e}"

# — 工具2: 查询工序（Literal 枚举）—
@tool
def query_process(
order_no: str,
detail_level: Literal["summary", "detail"] = "summary",
) –> str:
"""查询工单工序信息。"""
try:
wo = WORK_ORDERS.get(order_no)
if not wo:
return f"未找到工单 {order_no}"
procs = wo["processes"]
if detail_level == "summary":
done = sum(1 for p in procs if p["status"] == "已完成")
return f"共{len(procs)}道工序，已完成{done}道"
else:
lines = [f" {p['seq']}. {p['name']} — {p['status']}" for p in procs]
return "\\n".join(lines)
except Exception as e:
return f"查询出错：{e}"

# — 工具3: 上报进度（InjectedToolCallId）—
@tool
def report_progress(
order_no: str,
process_seq: int,
progress: int,
tool_call_id: Annotated[str, InjectedToolCallId],
) –> str:
"""上报工序完成进度。"""
try:
# 模拟更新
return f"已上报：工单{order_no}第{process_seq}道工序进度{progress}% (ID:{tool_call_id[:8]})"
except Exception as e:
return f"上报出错：{e}"

# — 工具4: 查库存（return_direct）—
@tool(return_direct=True)
def query_inventory(item_code: str) –> str:
"""查询物料库存。"""
try:
item = INVENTORY.get(item_code)
if not item:
return f"未找到物料 {item_code}"
return f"物料{item_code}({item['name']})：库存{item['qty']}件，库位{item['location']}"
except Exception as e:
return f"查询出错：{e}"

# — 中间件 —
@before_model
def log_before(state: AgentState, runtime: Runtime):
msg_count = len(state.get("messages", []))
print(f" [日志] 模型调用前，消息数: {msg_count}")

@after_model
def log_after(state: AgentState, runtime: Runtime):
print(" [日志] 模型调用完成")

limiter = ToolCallLimitMiddleware(
thread_limit=50,
run_limit=15,
)

# — 构建 Agent —
saver = InMemorySaver()
agent = create_agent(
model,
tools=[query_work_order, query_process, report_progress,
query_inventory, check_quality, get_equipment],
system_prompt="你是MES工单查询助手…",
checkpointer=saver,
middleware=[log_before, log_after, limiter],
)

# — 交互式 REPL —
config = {"configurable": {"thread_id": "default"}}

while True:
user_input = input("你 >>> ").strip()
if user_input in ("quit", "exit"):
break

# 流式输出
for chunk in agent.stream(
{"messages": [("user", user_input)]},
config=config,
stream_mode="updates",
):
for node, state in chunk.items():
if not state:
continue
msgs = state.get("messages", [])
if not msgs:
continue
last = msgs[–1]
content = getattr(last, "content", "")
if content and not getattr(last, "tool_calls", None):
msg_type = last.__class__.__name__
if msg_type == "ToolMessage":
continue # return_direct 的工具结果单独处理
print(f" Agent: {content}")

运行效果：

你 >>> 查工单 WO-2024-001
Agent: 工单WO-2024-001：产品轴承6204，数量500件，状态进行中，进度60%

你 >>> 它的工序详情 ← 不报工单号，记忆生效！
Agent: 工单WO-2024-001工序详情：
1. 下料 — 已完成
2. 车削 — 进行中
3. 热处理 — 待开始
4. 质检 — 待开始

你 >>> 查库存 A001 ← return_direct 直接返回
Agent: 物料A001(轴承6204)：库存150件，A区-03货架，安全库存50件

你 >>> 设备 EQ-002 什么状态
Agent: 设备EQ-002(热处理炉RT-90)：状态停机，利用率0%，下次维护2024-08-01

六、LangChain 1.x 踩坑总结（核心价值）

这是全文最有价值的部分。如果你正在学 LangChain 1.x，以下 8 个坑大概率会遇到：

最关键的教训

ToolException 那个坑坑了我一下午。旧教程都说 raise ToolException("错误信息") 会让模型自动重试，但在 1.x 里它直接把整个 Agent 搞崩。翻源码发现 langgraph.prebuilt.tool_node._default_handle_tool_errors 默认是重抛异常，而且 create_agent 没有暴露 handle_tool_errors 参数给你改。

最终结论：工具内部 try/except 返回错误字符串是最稳的方案，版本无关、永远有效。

七、学习方法论

先验证再学

每次遇到新 API，先写个小脚本验证能不能 import、能不能调用，确认没问题再写笔记和练习。这个习惯帮我避开了无数坑——与其照着旧教程写一堆跑不通的代码，不如先花 5 分钟验证 API。

每天产出

每天固定产出三个文件：

• dayXX_notes.md：学习笔记
• dayXX_practice.py：5 道练习题 + 测验
• 练习必须实跑通过才算完成

Token 统计

每次练习都加 token 统计，养成成本意识：

def print_token_usage(response):
usage = getattr(response, "usage_metadata", None)
if usage:
print(f"[token] 输入={usage.get('input_tokens')} "
f"输出={usage.get('output_tokens')} "
f"总计={usage.get('total_tokens')}")

结合实际项目

学 Tool Calling 时，直接把公司的 JeecgBoot 后端接口封装成 Tool；学 Agent 时，做 MES 工单查询 Agent。不要学抽象概念，要学能用的东西。

八、下一步计划（Day 24-30）

最终目标：一个完整的 MES 智能助手，有 Web 界面，能查工单、查库存、问答产品手册。

九、总结

23 天，从 Python 零基础到独立写出一个多轮对话 MES Agent，关键在于：

如果你也在学 AI Agent，希望这篇踩坑记录能帮你少走弯路。第四阶段 RAG + 部署完成后我会继续更新，敬请关注。

作者：程钧
技术栈：Python 3.14 + LangChain 1.3.13 + 阿里百炼 qwen-turbo
学习周期：2026.06.22 ~ 至今（前23天已完成）
项目背景：JeecgBoot v3.9.1 MES 制造执行系统二次开发