本文从 大模型应用开发技术路线清单 中拆分，聚焦“三、Agent 智能体”部分。让模型能“用工具、做决策、执行多步任务”。
什么是 Agent？
如果说 Prompt Engineering 是“教模型说话”，RAG 是“教模型查资料”，那 Agent 就是“教模型做事”。
Agent = LLM + 记忆 + 工具 + 规划
┌─────────────────────────────────────────┐
│ Agent │
│ ┌───────┐ ┌───────┐ ┌───────┐ │
│ │ LLM │←→│ 记忆 │←→│ 规划 │ │
│ └───┬───┘ └───────┘ └───────┘ │
│ │ │
│ ↓ │
│ ┌───────┐ ┌───────┐ ┌───────┐ │
│ │ 工具1 │ │ 工具2 │ │ 工具3 │ │
│ └───────┘ └───────┘ └───────┘ │
└─────────────────────────────────────────┘核心范式转变：从“模型直接回答”到“模型决定下一步做什么”。
一、Agent 核心技术
1.1 Function Calling：Agent 的基石
Function Calling 让模型能够选择调用哪个函数、提取参数。这是 Agent 与外部世界交互的基础。
from openai import OpenAI
import json
client = OpenAI()
# 定义工具
tools = [
{
"type": "function",
"function": {
"name": "search_knowledge_base",
"description": "搜索公司知识库，查找相关文档",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词",
},
"category": {
"type": "string",
"enum": ["hr", "tech", "product", "finance"],
"description": "知识库分类",
},
},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "获取指定城市的当前天气",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
},
"required": ["city"],
},
},
},
]
# 调用模型
messages = [{"role": "user", "content": "北京今天天气怎么样？"}]
response = client.chat.completions.create(
model="gpt-4",
messages=messages,
tools=tools,
tool_choice="auto", # 让模型自己决定是否调用工具
)
# 检查模型是否决定调用工具
if response.choices[0].message.tool_calls:
tool_call = response.choices[0].message.tool_calls[0]
function_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
print(f"模型决定调用: {function_name}")
print(f"参数: {arguments}")
# 输出：模型决定调用: get_current_weather
# 输出：参数: {"city": "北京", "unit": "celsius"}
# 执行函数
weather_result = get_current_weather(**arguments)
# 把结果返回给模型
messages.append(response.choices[0].message)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(weather_result, ensure_ascii=False),
})
final_response = client.chat.completions.create(
model="gpt-4",
messages=messages,
)
print(final_response.choices[0].message.content)1.2 ReAct 范式：思考-行动-观察循环
ReAct（Reasoning + Acting）是 Agent 的核心工作模式：
Thought: 用户问的是天气问题，我需要查询天气 API
Action: get_current_weather(city="北京")
Observation: {"temperature": 22, "weather": "晴", "wind": "北风3级"}
Thought: 我已经获得了天气信息，可以回答用户了
Final Answer: 北京今天天气晴朗，气温22°C，北风3级。完整实现：
def react_agent(user_input, tools, max_iterations=5):
"""最简化的 ReAct Agent"""
messages = [
{"role": "system", "content": """你是一个 AI 助手，可以使用工具来完成任务。
可用工具：
- search_web(query): 搜索网页
- calculate(expression): 计算数学表达式
- get_weather(city): 查询天气
请按照以下步骤思考：
1. 分析用户需要什么
2. 决定是否需要使用工具
3. 如果需要，调用合适的工具
4. 根据工具结果回答用户
"""},
{"role": "user", "content": user_input},
]
for i in range(max_iterations):
response = client.chat.completions.create(
model="gpt-4",
messages=messages,
tools=tools,
)
msg = response.choices[0].message
# 模型不再调用工具 → 得到最终答案
if not msg.tool_calls:
return msg.content
# 模型调用了工具 → 执行工具并继续循环
messages.append(msg)
for tool_call in msg.tool_calls:
result = execute_tool(tool_call.function.name,
json.loads(tool_call.function.arguments))
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(result),
})
return "达到最大迭代次数"1.3 Tool Use：工具设计
工具设计原则：
| 原则 | 说明 | 示例 |
|---|---|---|
| 命名清晰 | 函数名和参数名要自解释 | search_web 而非 tool1 |
| 描述充分 | description 要告诉模型什么时候用 | “当需要搜索实时信息时使用” |
| 参数简洁 | 只传必要参数 | 不要传10个可选参数 |
| 返回结构化 | JSON 格式，方便模型理解 | {"result": "...", "source": "..."} |
常用工具类型：
# 搜索类工具
def search_web(query: str) -> dict:
"""搜索网页获取实时信息"""
results = web_search_api.search(query, num=5)
return {"results": [{"title": r.title, "url": r.url, "snippet": r.snippet} for r in results]}
# 数据库查询工具
def query_database(sql: str) -> dict:
"""执行只读 SQL 查询"""
# 安全校验：只允许 SELECT
if not sql.strip().upper().startswith("SELECT"):
return {"error": "只允许 SELECT 查询"}
result = db.execute(sql)
return {"rows": result.fetchall(), "columns": result.keys()}
# 文件操作工具
def read_file(path: str) -> dict:
"""读取文件内容"""
with open(path, "r") as f:
return {"content": f.read()}
def write_file(path: str, content: str) -> dict:
"""写入文件"""
with open(path, "w") as f:
f.write(content)
return {"status": "success", "path": path}
# API 调用工具
def send_email(to: str, subject: str, body: str) -> dict:
"""发送邮件"""
email_client.send(to=to, subject=subject, body=body)
return {"status": "sent", "to": to}1.4 记忆管理
| 记忆类型 | 存储方式 | 用途 |
|---|---|---|
| 短期记忆 | 对话历史（Messages 数组） | 当前对话上下文 |
| 长期记忆 | 向量数据库 | 跨会话的用户偏好、历史信息 |
| 工作记忆 | Agent 内部状态 | 任务进度、中间结果 |
class AgentMemory:
def __init__(self):
self.short_term = [] # 对话历史
self.working = {} # 工作状态
self.long_term = VectorStore() # 长期记忆
def add_message(self, role, content):
self.short_term.append({"role": role, "content": content})
# 上下文太长时，摘要压缩
if self.count_tokens() > MAX_TOKENS:
self.summarize_history()
def search_long_term(self, query, k=3):
"""检索长期记忆"""
return self.long_term.similarity_search(query, k=k)
def save_to_long_term(self, content, metadata=None):
"""保存到长期记忆"""
self.long_term.add_texts([content], metadatas=[metadata])1.5 规划能力
任务拆解示例：
用户需求："帮我分析上个月的销售数据，找出 Top10 产品，生成报告"
Agent 规划：
1. [查询] 获取上个月的销售数据（SQL 查询）
2. [分析] 按产品汇总销售额，排序取 Top10
3. [可视化] 生成柱状图
4. [撰写] 写分析报告（趋势、异常、建议）
5. [输出] 保存为 PDF 报告二、Agent 进阶技术
2.1 Multi-Agent 多智能体
多个 Agent 协作完成复杂任务：
┌─────────────────────────────────────────────┐
│ Manager Agent │
│ (任务拆解、进度监控) │
└──────┬──────────┬──────────┬────────────────┘
↓ ↓ ↓
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Coder │ │ Reviewer │ │ Tester │
│ Agent │ │ Agent │ │ Agent │
│(写代码) │ │(代码审查) │ │(执行测试) │
└──────────┘ └──────────┘ └──────────┘2.2 Agent 编排框架对比
| 框架 | 特点 | 适用场景 |
|---|---|---|
| LangChain | 生态最全，组件丰富 | 通用 Agent 开发，快速原型 |
| LlamaIndex | 专注 RAG 和数据索引 | 知识密集型应用 |
| AutoGen | 多智能体协作，对话驱动 | 复杂任务分解 |
| CrewAI | 角色扮演，任务委派 | 需要“角色”协作的场景 |
| LangGraph | 状态图编排，支持循环/条件 | 复杂工作流，精细控制 |
| OpenManus | 模块化，支持本地运行 | 企业级定制 |
2.3 LangGraph 深入
LangGraph 是构建复杂 Agent 工作流的首选框架：
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
# 定义状态
class AgentState(TypedDict):
messages: list
plan: str
current_step: int
results: dict
# 定义节点
def planner(state: AgentState) -> AgentState:
"""规划节点：拆解任务"""
plan = llm.invoke(f"将任务拆解为步骤：{state['messages'][-1]}")
return {"plan": plan, "current_step": 0}
def executor(state: AgentState) -> AgentState:
"""执行节点：执行当前步骤"""
step = state["plan"][state["current_step"]]
result = execute_step(step)
return {"results": {**state["results"], step: result}}
def reviewer(state: AgentState) -> AgentState:
"""审查节点：检查结果"""
review = llm.invoke(f"审查以下结果：{state['results']}")
return {"messages": state["messages"] + [review]}
# 构建图
graph = StateGraph(AgentState)
graph.add_node("planner", planner)
graph.add_node("executor", executor)
graph.add_node("reviewer", reviewer)
# 定义边
graph.set_entry_point("planner")
graph.add_edge("planner", "executor")
graph.add_conditional_edges(
"executor",
lambda s: "reviewer" if s["current_step"] >= len(s["plan"]) - 1 else "executor",
)
graph.add_edge("reviewer", END)
# 编译并运行
app = graph.compile()
result = app.invoke({"messages": [user_input]})2.4 工具链标准化（MCP）
MCP（Model Context Protocol）是一种标准化的工具接入协议：
传统方式：每个 Agent 自定义工具接口
MCP 方式：统一的工具发现和调用协议
优势：
- 工具复用：一个 MCP 工具可在多个 Agent 中使用
- 标准化：统一的参数描述和返回格式
- 可发现性：Agent 自动发现可用工具2.5 Agent 可观测性
追踪 Agent 的每一步推理和工具调用：
# 使用 LangSmith 追踪
import langsmith
from langsmith import traceable
@traceable
def agent_step(step_name, input_data):
"""每个步骤都被追踪"""
# ... 执行逻辑
return result
# 使用 Langfuse（自部署）
from langfuse.callback import CallbackHandler
handler = CallbackHandler()
response = agent.invoke(input, config={"callbacks": [handler]})三、Agent 实战案例
3.1 案例1：故障诊断 Agent
工具集：
├── 搜索工具：查询知识库、文档
├── 监控工具：查询系统指标、日志
├── 执行工具：重启服务、修改配置
└── 通知工具：发送告警、生成报告
流程：
1. 接收告警 → 2. 分析症状 → 3. 搜索知识库
4. 查询监控 → 5. 定位根因 → 6. 执行修复
7. 验证结果 → 8. 生成报告tools = [
{
"name": "query_logs",
"description": "查询指定服务的日志",
"parameters": {
"service": "服务名",
"time_range": "时间范围",
"level": "日志级别"
}
},
{
"name": "query_metrics",
"description": "查询系统监控指标",
"parameters": {
"metric": "指标名（CPU/内存/请求量/延迟）",
"service": "服务名"
}
},
{
"name": "restart_service",
"description": "重启指定服务",
"parameters": {
"service": "服务名"
}
},
{
"name": "send_alert",
"description": "发送告警通知",
"parameters": {
"channel": "通知渠道",
"message": "告警内容"
}
}
]
# Agent Prompt
system_prompt = """
你是一个运维故障诊断 Agent。收到告警后：
1. 分析告警信息，确定受影响的服务
2. 查询相关日志和监控指标
3. 根据症状在知识库中搜索类似案例
4. 定位根因
5. 如果是已知问题，执行修复方案
6. 生成故障报告
注意：重启服务前必须确认影响范围，重大操作需要人工确认。
"""3.2 案例2：AI 写作助手（Multi-Agent）
模块设计：
├── Orchestrator：任务编排，分配写作任务
├── Research Agent：收集素材，搜索资料
├── Writer Agent：生成内容，风格控制
├── Editor Agent：润色修改，质量检查
└── Memory：保存写作偏好、历史稿件
流程：
1. 用户输入主题 → 2. 研究 Agent 收集素材
3. 写作 Agent 生成初稿 → 4. 编辑 Agent 润色
5. 用户确认 → 6. 保存到知识库from langgraph.graph import StateGraph
class WritingState(TypedDict):
topic: str
research_results: list
draft: str
final_version: str
feedback: str
def research_agent(state: WritingState) -> WritingState:
"""研究 Agent：搜索素材"""
results = search_web(state["topic"])
return {"research_results": results}
def writer_agent(state: WritingState) -> WritingState:
"""写作 Agent：生成初稿"""
prompt = f"""
主题：{state['topic']}
素材：{state['research_results']}
请根据以上素材撰写一篇 2000 字的文章。
"""
draft = llm.invoke(prompt)
return {"draft": draft}
def editor_agent(state: WritingState) -> WritingState:
"""编辑 Agent：润色修改"""
prompt = f"""
请对以下文章进行编辑润色：
{state['draft']}
编辑要求：
1. 修正语法错误
2. 优化表达方式
3. 检查逻辑连贯性
4. 确保准确性
"""
final = llm.invoke(prompt)
return {"final_version": final}3.3 案例3：Multi-Agent 协作（代码开发）
角色设计：
├── Manager Agent：任务拆解，进度监控
├── Coder Agent：代码编写，单元测试
├── Reviewer Agent：代码审查，安全扫描
├── Tester Agent：集成测试，性能测试
└── Deployer Agent：部署上线，监控告警
协作模式：
Manager → 分配任务 → Coder → 提交代码
↓
Reviewer ← Code Review
↓
Tester → 测试通过
↓
Deployer → 部署上线四、Spring AI Agent 实践（Java）
// Spring AI Function Calling 示例
@Configuration
public class ToolConfig {
@Bean
@Description("查询公司知识库")
public Function<KnowledgeQuery, KnowledgeResult> searchKnowledge() {
return query -> {
List<Document> docs = vectorStore.similaritySearch(query.getQuery());
return new KnowledgeResult(docs.stream()
.map(Document::getContent)
.collect(Collectors.joining("\n---\n")));
};
}
@Bean
@Description("查询数据库")
public Function<DatabaseQuery, DatabaseResult> queryDatabase() {
return query -> {
// SQL 安全校验
if (!query.getSql().trim().toUpperCase().startsWith("SELECT")) {
return new DatabaseResult("只允许 SELECT 查询");
}
List<Map<String, Object>> rows = jdbcTemplate.queryForList(query.getSql());
return new DatabaseResult(rows);
};
}
}
// Agent 调用
@RestController
public class AgentController {
private final ChatModel chatModel;
private final List<FunctionCallback> tools;
@PostMapping("/agent/chat")
public String chat(@RequestBody String userMessage) {
var prompt = Prompt.builder()
.messages(new SystemMessage("你是一个 AI 助手，可以使用工具完成任务。"),
new UserMessage(userMessage))
.toolCallbacks(tools)
.build();
return chatModel.call(prompt).getResult().getOutput().getContent();
}
}五、学习建议
动手路径
Week 1: Function Calling
├── 定义 3 个工具（搜索、计算、查天气）
├── 实现基础 Function Calling
└── 理解 tool_calls 和 tool 响应
Week 2: ReAct Agent
├── 实现 ReAct 循环
├── 加入错误处理
└── 做一个能搜索+计算的 Agent
Week 3: 进阶
├── 学习 LangGraph
├── 实现 Multi-Agent 协作
└── 加入可观测性（LangSmith/Langfuse）上一篇：二、RAG 检索增强生成 下一篇：四、模型微调 Fine-tuning 返回导航：大模型应用开发技术路线清单