一、什么是 AI Agent?
一个朴素的定义:
AI Agent = 大语言模型(大脑)+ 工具(手脚)+ 规划能力(决策)+ 记忆(经验)
跟传统 LLM 应用最大的区别是 —— Agent 拥有自主决策和执行的能力。你给 Agent 一个目标,它会自己拆解任务、选择工具、调用 API、读取结果、决定下一步,直到目标完成。
1.1 从 Chatbot 到 Agent 的本质区别
| 对比维度 | 传统 Chatbot | AI Agent |
|---|---|---|
| 输入输出 | 文本 → 文本 | 目标 → 多步操作 |
| 决策方式 | 单次推理 | 持续规划、反思、调整 |
| 工具使用 | 无 | 调用 API、读写文件、执行代码 |
| 执行时长 | 秒级 | 分钟到小时,多轮迭代 |
| 典型应用 | 问答、写作 | 自动办公、代码生成、数据分析 |
1.2 一个最简 Agent 的代码示例
# 最简 Agent:让 LLM 决定用计算器
from openai import OpenAI
client = OpenAI()
TOOLS = [{
"type": "function",
"function": {
"name": "calculator",
"description": "计算数学表达式,支持加减乘除",
"parameters": {
"type": "object",
"properties": {
"expression": {"type": "string"}
}
}
}
}]
def calculator(expression: str) -> str:
if not re.match(r'^[\d\s\+\-\*\/\(\)]+$', expression):
return "错误:非法字符"
return str(eval(expression))
def run_agent(user_input, max_steps=5):
messages = [{"role": "user", "content": user_input}]
for step in range(max_steps):
# 1. 让 LLM 决定下一步
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=TOOLS,
)
msg = response.choices[0].message
# 2. 如果 LLM 想调工具,执行工具
if msg.tool_calls:
tool_call = msg.tool_calls[0]
result = calculator(tool_call.function.arguments)
messages.append({"role": "tool", "content": result})
else:
# 3. LLM 给出最终答案
return msg.content
return "Agent 超过最大步数,未完成。"
# 测试
print(run_agent("一个长方形的长是 12 米,宽是 8 米,求面积和周长。"))
# 输出:面积是 96 平方米,周长是 40 米。
二、Agent 的四大核心设计模式
2024 年,吴恩达(Andrew Ng)总结了 Agent 区别于普通 LLM 调用的四种核心模式,这些模式可以单独使用,也可以组合使用。
2.1 Reflection(反思模式)
让模型自己检查自己的工作 —— 写完代码后,让同一个(或不同的)模型来 review,发现问题就重写。
关键洞察:模型在"审视别人错误"时比"自己写不出错"时更敏锐。这背后是不对称的认知负荷 —— 生成时要同时考虑内容、结构、风格、准确性,审查时只关注一个问题。
def reflect_and_refine(task, max_iterations=3):
output = llm.invoke(f"请完成: {task}")
for i in range(max_iterations):
critique = llm.invoke(f"审视以下输出找问题: {output}\n如无问题回复 NO_ISSUES")
if "NO_ISSUES" in critique:
break
output = llm.invoke(f"根据反馈修正: {critique}")
return output 效果:在 HumanEval 编码基准上,这种"反思"让 GPT-3.5 的准确率从 48% 提升到 95%,超过了 GPT-4 的零样本表现(67.5%)。
2.2 Tool Use(工具调用模式)
LLM 知识有截止日期、不具备与外部世界交互的能力。Tool Use 让 LLM 能搜索最新信息、执行代码、操作数据库、调用 API。
工具设计的黄金法则:
- 一个工具只做一件事
- 参数不超过 5 个
- 明确区分只读和写入(写入类需要人工确认)
- 返回结构化的错误信息
2.3 Planning(规划模式)
面对复杂任务,模型不能直接开始执行 —— 要先规划,再执行,边做边调整。三种主要变体:
| 范式 | 特点 | 适合场景 |
|---|---|---|
| Plan-and-Execute | 先制定完整计划,再逐步执行 | 目标明确、路径清晰 |
| ReAct | 推理-行动-观察循环,每步重新规划 | 探索性强、不确定性高 |
| Tree-of-Thought | 同时探索多条路径,选择最优 | 需要创造性的复杂问题 |
2.3.1 ReAct 循环:Agent 最核心的工作流
ReAct = Reasoning + Acting。这是 2022 年 Princeton 提出的范式,现在所有 Agent 框架(LangGraph、AutoGen、CrewAI)的核心循环都是它。
2.4 Multi-Agent(多智能体协作)
当任务跨领域、需多种专门能力时,单个 Agent 往往成为瓶颈。多 Agent 协作有四种典型拓扑:
- 顺序流水线:研究员 → 分析师 → 撰稿人 → 审查者
- 辩论模式:正方 Agent vs 反方 Agent,裁判 Agent 裁决
- 层级模式:管理者 Agent 分配任务,执行者 Agent 干活
- 群聊模式:所有 Agent 在共享空间自由交流
2.4.1 四种 Multi-Agent 拓扑对比
选型建议:任务线性 → 顺序;需要多视角校验 → 辩论;需要动态分派 → 层级;需要涌现创意 → 群聊。
三、主流 Agent 框架对比
到 2025-2026 年,GitHub 上最热门的 Agent 框架有四个:
3.1 LangGraph(LangChain 团队)
核心抽象:基于状态图(StateGraph)的底层编排引擎。
from langgraph.graph import StateGraph, END
from typing import TypedDict
class State(TypedDict):
messages: list
next_step: str
workflow = StateGraph(State)
workflow.add_node("think", think_node)
workflow.add_node("act", act_node)
workflow.add_node("observe", observe_node)
workflow.add_conditional_edges("think", router, {"act": "act", "end": END})
workflow.add_edge("act", "observe")
workflow.add_edge("observe", "think")
app = workflow.compile() 特点:最大的灵活性,所有流程由你定义。GitHub 10k+ stars。学习曲线较陡,但生产级特性最完整(LangSmith 集成、Checkpointer 持久化、Human-in-the-Loop)。
3.2 CrewAI
核心抽象:角色(Role)+ 任务(Task)+ 团队(Crew)。
from crewai import Agent, Task, Crew
researcher = Agent(
role="市场研究员",
goal="深度分析目标市场,发现趋势和机会",
backstory="你是一位有 10 年经验的资深市场研究员",
)
writer = Agent(
role="报告撰写人",
goal="将研究成果转化为清晰的商业报告",
backstory="你是一位顶尖的商业报告撰写专家",
)
crew = Crew(agents=[researcher, writer], tasks=[research_task, write_task])
result = crew.kickoff() 特点:25k+ stars,上手最快(15 行代码就能跑)。角色扮演 + 任务流水线,适合业务团队快速搭 Agent。
3.3 AutoGen(微软)
核心抽象:Agent 之间像人一样"聊天"协作。
from autogen import AssistantAgent, UserProxyAgent
assistant = AssistantAgent(
name="编程助手",
llm_config={"config_list": [{"model": "gpt-4o"}]},
)
user = UserProxyAgent(name="用户", human_input_mode="TERMINATE")
user.initiate_chat(assistant, message="帮我写个数据分析脚本") 特点:42k+ stars。Human-in-the-Loop 设计最自然,适合频繁人机交互。但调试相对复杂。
3.4 Google ADK(Agent Development Kit)
Google 在 2025 年发布的 Agent SDK,主打"生产就绪",与 Gemini 模型深度集成,内置 A2A(Agent-to-Agent)协议。
框架选型决策树
你的 Agent 系统需要构建吗?
│
├─ 需要精细的流程控制 / RAG / 状态管理? ─→ LangGraph
│
├─ 任务典型多角色协作(研究/分析/写作)? ───→ CrewAI
│
├─ 需要频繁人机交互(代码助手/数据问答)? ──→ AutoGen
│
├─ 快速原型,几天要 Demo? ─────────────────→ CrewAI → LangGraph
│
└─ 生产系统,需要可观测性/部署? ──────────→ LangGraph
四、Agentic Agent:从单 Agent 到自主系统
"Agentic"不是新词,但在 2024-2025 年成了整个行业的核心叙事。它的核心含义是:从"工具性使用 LLM"升级到"构建能自主决策、持续运行的智能系统"。
4.1 OpenAI 的 Agent 路线图
2025 年 1 月,OpenAI 用两个产品宣告了 Agent 时代到来:
- Operator:基于 Computer Use 范式,操控浏览器完成订票、网购、填表等任务。
- Deep Research:基于 Knowledge Research 范式,自动搜集信息、生成深度研究报告。
Sam Altman 把这称为"AGI 第三层 - Agent 层"。
4.2 Anthropic Computer Use
2024 年 10 月,Anthropic 让 Claude 像人一样使用计算机 —— 看屏幕、移动光标、点击按钮、输入文本。Claude 3.5 Sonnet 升级版的 Computer Use 打开了"AI 直接操作 GUI"的大门。
4.3 Google Gemini 2.5 Computer Use
2025 年,Google 发布 Gemini 2.5 Computer Use Preview,采用循环交互机制:
- 向模型发送请求(用户目标 + 当前 GUI 截图)
- 模型分析后生成 function_call(具体 UI 操作)
- 执行 function_call(在浏览器或移动端)
- 捕获新截图,反馈给模型,开始新循环
在 WebArena、Online-Mind2Web、Mobile Control 等基准上,延迟低于竞争对手。
4.4 GitHub Agentic Workflows
GitHub 在 2026 年初推出 Agentic Workflows 技术预览,允许开发者用 Markdown 文件定义自动化目标,让编码 Agent 通过 GitHub Actions 执行:
# Daily Repo Status Report
Create a daily status report for maintainers. Include:
- Recent activity (issues, PRs, discussions, releases)
- Progress tracking, goal reminders
- Project status and recommendations
- Actionable next steps
Keep it concise and link to the relevant items. 内置多层安全机制(沙箱、只读默认、安全输出审查),体现"持续 AI"(Continuous AI)愿景。
五、上下文工程(Context Engineering):Agent 时代的真正核心
2025 年下半年,业界逐渐达成共识:决定 Agent 效果的不是模型本身,而是上下文质量。Andrew Ng 在 2025 年明确说:"上下文工程比 prompt engineering 更重要。"
5.1 Agent 需要的三大上下文
- 领域知识:RAG 系统检索的文档片段
- 工具数据:可用工具的描述、使用指南
- 会话状态:历史对话、用户偏好、当前任务状态
5.2 工具检索(Tool Retrieval)的崛起
当企业有数百个工具时,把全部描述塞进 prompt 会让 LLM "选择困难"。解决方案:对工具描述建立索引,根据当前任务动态检索 Top-3 工具。
即使是简单的 BM25 关键词检索,在这个任务上都能作为很强基线。
六、生产环境避坑指南
6.1 状态管理失控
用统一的 Schema + 会话隔离(thread_id)避免"把用户 A 的订单返回给用户 B"。
6.2 成本失控
Agent 循环会指数级放大 Token 消耗。典型倍数:
- 零样本:1x
- Agentic(5 轮反思):6x
- Multi-Agent(3 个 Agent):18x
解法:分级模型策略(简单任务用 gpt-4o-mini,复杂任务用 gpt-4o)+ 上下文窗口管理 + 预算硬限制。
6.3 幻觉放大
Agent 循环中,一个幻觉会成为下一步的"事实输入",错误被持续放大。解法:工具调用后做结果验证 + 关键节点双验证。
6.4 延迟
用户耐心只有 3 秒。Agent 多步串行的延迟很容易超 10 秒。解法:流式输出 + 快速路径(简单问题跳过 Agent 循环)+ 并行化独立任务。
七、章节总结
- ✅ Agent = LLM + 工具 + 规划 + 记忆,核心能力是自主决策。
- ✅ 四大设计模式:Reflection、Tool Use、Planning、Multi-Agent。
- ✅ 选型决策:LangGraph(精细控制)、CrewAI(快速上手)、AutoGen(人机交互)、Google ADK(企业生产)。
- ✅ Agentic Agent 时代:从单 Agent 工具化使用,到自主运行的智能系统。
- ✅ 真正决定 Agent 效果的是上下文工程,不是模型本身。
用 LangGraph 构建生产级 Agent:智能 PR 审查 Agent
这一节我们用 LangGraph 从零搭一个能自动审查 GitHub Pull Request 的 Agent。它有 5 个节点(加载 diff → 静态检查 → LLM 审查 → 生成评论 → 写回 GitHub),带状态持久化 + Human-in-the-Loop 中断。完整代码约 350 行,本地真能跑通。
项目需求
- 输入:GitHub PR URL(例:
https://github.com/owner/repo/pull/123) - 行为:拉 diff → 跑 ruff + bandit → 用 LLM 综合评审 → 把评论 POST 到 PR
- 特性:大 PR 自动拆 chunk、人类可在关键节点 approve/reject、失败可重试、状态可恢复
Step 1:环境准备
pip install langgraph langchain-anthropic langchain-openai httpx tenacity python-dotenv
# 验证
python -c "import langgraph, langchain_anthropic; print('langgraph', langgraph.__version__)" 需要的环境变量(放 .env):
ANTHROPIC_API_KEY=sk-ant-xxx
GITHUB_TOKEN=ghp_xxx Step 2 + 3:定义状态 + 节点(完整代码 pr_agent.py,约 280 行)
"""LangGraph PR Review Agent。
用法:
python pr_agent.py https://github.com/owner/repo/pull/123
"""
from __future__ import annotations
from typing import Annotated, Literal, TypedDict
from langchain_anthropic import ChatAnthropic
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import interrupt
from pydantic import BaseModel
from tenacity import retry, stop_after_attempt, wait_exponential
# ---------- 1. 状态定义 ----------
class PRState(TypedDict, total=False):
"""贯穿整个图的全局状态。"""
pr_url: str
owner: str
repo: str
pr_number: int
diff: str
files: list[str]
lint_issues: list[str]
review_comments: list[str]
final_report: str
needs_human: bool
error: str
attempts: int
# ---------- 2. 工具函数 ----------
def parse_pr_url(url: str) -> tuple[str, str, int]:
"""解析 https://github.com/owner/repo/pull/123。"""
m = re.match(r"https://github\.com/([^/]+)/([^/]+)/pull/(\d+)", url)
if not m:
raise ValueError(f"bad PR URL: {url}")
return m.group(1), m.group(2), int(m.group(3))
@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
def github_get(path: str, token: str) -> dict:
"""带重试的 GitHub API GET。"""
r = httpx.get(
f"https://api.github.com{path}",
headers={"Authorization": f"Bearer {token}", "Accept": "application/vnd.github+json"},
timeout=30,
)
r.raise_for_status()
return r.json()
@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
def github_post(path: str, body: dict, token: str) -> dict:
"""带重试的 GitHub API POST。"""
r = httpx.post(
f"https://api.github.com{path}",
headers={"Authorization": f"Bearer {token}", "Accept": "application/vnd.github+json"},
json=body,
timeout=30,
)
r.raise_for_status()
return r.json()
# ---------- 3. 节点定义 ----------
def node_load_diff(state: PRState) -> PRState:
"""节点 1:拉 PR diff + 文件列表。"""
print("[node] load_diff")
try:
owner, repo, pr = parse_pr_url(state["pr_url"])
diff = httpx.get(
f"https://patch-diff.githubusercontent.com/raw/{owner}/{repo}/pull/{pr}.diff",
headers={"Accept": "application/vnd.github.v3.diff"},
timeout=30,
).text
meta = github_get(f"/repos/{owner}/{repo}/pulls/{pr}/files", os.environ["GITHUB_TOKEN"])
files = [f["filename"] for f in meta]
print(f" loaded {len(files)} files, diff {len(diff)} chars")
return {"owner": owner, "repo": repo, "pr_number": pr, "diff": diff, "files": files}
except Exception as e:
return {"error": f"load_diff failed: {e}"}
def node_lint(state: PRState) -> PRState:
"""节点 2:对每个 Python 文件跑 ruff + bandit,收集 issue。"""
print("[node] lint")
if state.get("error"):
return {}
issues: list[str] = []
for f in state.get("files", []):
if not f.endswith(".py"):
continue
# 写临时文件喂给工具(实际项目里可换成 in-memory)
with open(f, "r", encoding="utf-8", errors="ignore") as fp:
src = fp.read()
# ruff(若没装就跳过)
try:
r = subprocess.run(["ruff", "check", "--stdin-filename", f, "-"],
input=src, capture_output=True, text=True, timeout=15)
if r.stdout.strip():
issues.append(f"### ruff: {f}\n```\n{r.stdout}\n```")
except FileNotFoundError:
pass
# bandit(可选)
try:
r = subprocess.run(["bandit", "-q", "-"], input=src, capture_output=True, text=True, timeout=15)
if r.stdout.strip():
issues.append(f"### bandit: {f}\n```\n{r.stdout}\n```")
except FileNotFoundError:
pass
print(f" collected {len(issues)} lint blocks")
return {"lint_issues": issues}
def node_llm_review(state: PRState) -> PRState:
"""节点 3:用 Claude 综合 diff + lint,生成结构化评论。"""
print("[node] llm_review")
if state.get("error"):
return {}
llm = ChatAnthropic(model="claude-haiku-4-5", temperature=0)
prompt = f"""你是严格的代码审查员。基于以下信息给出 Markdown 报告,分 3 节:
## 1. Lint 结果
## 2. 安全问题
## 3. 改进建议
PR: {state['pr_url']}
变更文件: {', '.join(state.get('files', []))}
=== Lint 输出 ===
{chr(10).join(state.get('lint_issues', [])) or '(无)'}
=== Diff (前 6000 字符) ===
{state.get('diff', '')[:6000]}
"""
try:
resp = llm.invoke(prompt)
report = resp.content if isinstance(resp.content, str) else resp.content[0].text
# 大 PR 让人类看一眼
needs_human = len(state.get("diff", "")) > 20000
print(f" report={len(report)} chars needs_human={needs_human}")
return {"final_report": report, "needs_human": needs_human, "attempts": state.get("attempts", 0) + 1}
except Exception as e:
return {"error": f"llm_review failed: {e}"}
def node_human_approve(state: PRState) -> PRState:
"""节点 4:Human-in-the-Loop 中断,等人类 approve。"""
if not state.get("needs_human"):
return {}
print("[node] human_approve: waiting for human...")
decision = interrupt({
"question": "报告已生成,是否 POST 到 GitHub?",
"preview": state.get("final_report", "")[:2000],
})
if decision != "approve":
return {"error": f"human rejected: {decision}"}
return {}
def node_post_comment(state: PRState) -> PRState:
"""节点 5:把报告 POST 到 PR 评论。"""
print("[node] post_comment")
if state.get("error"):
return {}
try:
body = {
"body": f"## 🤖 AI Code Review\n\n{state.get('final_report', '(empty)')}\n\n---\n*Generated by LangGraph PR Agent*",
}
github_post(
f"/repos/{state['owner']}/{state['repo']}/issues/{state['pr_number']}/comments",
body,
os.environ["GITHUB_TOKEN"],
)
print(" ✓ posted to GitHub")
except Exception as e:
return {"error": f"post_comment failed: {e}"}
return {}
# ---------- 4. 路由 ----------
def route_after_review(state: PRState) -> Literal["human_approve", "post_comment", "__end__"]:
"""根据状态决定下一步:有错误就结束,需人审就中断,否则直接 post。"""
if state.get("error"):
return END
if state.get("needs_human"):
return "human_approve"
return "post_comment"
# ---------- 5. 编译图 ----------
def build_graph():
g = StateGraph(PRState)
g.add_node("load_diff", node_load_diff)
g.add_node("lint", node_lint)
g.add_node("llm_review", node_llm_review)
g.add_node("human_approve", node_human_approve)
g.add_node("post_comment", node_post_comment)
g.add_edge(START, "load_diff")
g.add_edge("load_diff", "lint")
g.add_edge("lint", "llm_review")
g.add_conditional_edges("llm_review", route_after_review, {
"human_approve": "human_approve",
"post_comment": "post_comment",
END: END,
})
g.add_edge("human_approve", "post_comment")
g.add_edge("post_comment", END)
return g.compile(checkpointer=MemorySaver())
# ---------- 6. 入口 ----------
def main():
import sys
if len(sys.argv) != 2:
print("usage: python pr_agent.py <pr_url>")
sys.exit(1)
app = build_graph()
config = {"configurable": {"thread_id": sys.argv[1]}}
print(f"=== Reviewing {sys.argv[1]} ===")
for event in app.stream({"pr_url": sys.argv[1]}, config=config):
print(event)
print("=== Done ===")
if __name__ == "__main__":
main() Step 5:运行 + 测试
# 1. 用一个真实的小 PR 测试(自己 fork 一个 demo 仓库)
python pr_agent.py https://github.com/your-name/demo/pull/1 预期输出(终端 + GitHub):
=== Reviewing https://github.com/your-name/demo/pull/1 ===
[node] load_diff
loaded 3 files, diff 1234 chars
[node] lint
collected 1 lint blocks
[node] llm_review
report=842 chars needs_human=False
[node] post_comment
✓ posted to GitHub
=== Done === 截图位置:GitHub PR 页面 → 看到 AI 发的评论含 "🤖 AI Code Review" 标题 + 3 节报告。
Step 6:Graph 结构(Mermaid)
Step 7:进阶扩展
- 接入 GitHub Webhook:用 FastAPI 暴露
POST /webhook,收到pull_request事件就触发app.stream;用thread_id = f"{repo}-{pr}"保证幂等。 - 持久化 checkpoint:把
MemorySaver换成PostgresSaver(langgraph-checkpoint-postgres),重启 Agent 可从上次中断点继续。 - 加 memory:用
langgraph.store配InMemoryStore,记住这个 repo 上次审查过的常见问题,prompt 里加 "上次你发现这里有 X 问题,这次仍然存在"。 - 多 Agent 协作:拆出
security_agent(只看安全) +perf_agent(只看性能) +style_agent(只看风格),汇总到orchestrator节点。 - Token 控制:在
llm_review节点加langchain.callbacks.get_openai_callback()统计 token,超过阈值就 chunk 摘要后再 review。
常见问题 & 调试
- interrupt 卡住:用
app.invoke(Command(resume="approve"), config)恢复。 - GitHub 401:检查
GITHUB_TOKEN是否有reposcope。 - diff 太长:把
state["diff"][:6000]拆成多段循环 review,最后合并。 - 图渲染:用
app.get_graph().draw_mermaid_png()输出可视化。
十二、视频学习:看 Karpathy 拆解 LLM 工作原理
Agent 写代码已经够复杂,但要把 Agent 用对,必须先理解 LLM 本身到底在做什么。下面这个 Karpathy 长片是最常被推荐给产品 / 研发负责人的「必看」:讲训练,看完后你对 Agent 的边界会有全新判断。
Let's reproduce GPT-2 (124M)
展开介绍
Karpathy「Neural Networks: Zero to Hero」系列最新长片。4 小时完整复现 GPT-2 (124M):从空文件构建 → Flash Attention + tf32 + bfloat16 把训练压到 96ms/iter → AdamW + 预热 + 余弦衰减 → DDP 分布式 → FineWeb 数据集训练 → HellaSwag 评估。Scale AI CEO Alexandr Wang 看后评价:「Karpathy 发新视频就像流行歌手发新歌」。视频语言:英文(配中文字幕)。