本课是LLM应用开发课程第15课，属于Agent开发阶段。Agent是LLM应用的高级形态——它不再只是被动地回答问题，而是能够自主规划、使用工具、迭代反思来完成任务。理解Agent架构是构建复杂AI系统的基石。
第15课: Agent架构
├── Agent vs 非Agent
│ ├── 单次调用 (Chat Completion)
│ └── 循环决策 (Agent Loop)
├── 三种架构模式
│ ├── 单Agent: 一个LLM循环处理
│ ├── 多Agent: 多个Agent协作
│ └── 分层Agent: 管理者+执行者
├── SimpleAgent
│ ├── 工具注册与发现
│ ├── 决策循环 (Observe→Think→Act)
│ └── 终止条件判断
└── ObservableAgent
├── 执行日志 (Trace)
├── 步骤计时 (Duration)
└── 回调钩子 (Hooks)
非Agent应用是单轮映射：输入→LLM→输出。Agent是循环决策：输入→观察→思考→行动→观察→...→输出。核心差异在于Agent拥有自主控制流。
# Agent vs 非Agent: 核心区别演示
from openai import OpenAI
import json
client = OpenAI()
# 非Agent: 单次调用，无法使用工具
def non_agent_approach(question: str) -> str:
"""非Agent: 一问一答，LLM只能靠自身知识回答"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "你是知识问答助手。"},
{"role": "user", "content": question},
],
)
return response.choices[0].message.content
# Agent: 循环决策，可以使用工具获取实时信息
class AgentLoop:
"""Agent核心: 观察→思考→行动 循环"""
def __init__(self, model="gpt-4o-mini"):
self.model = model
self.tools = {}
self.max_iterations = 5
def register_tool(self, name: str, description: str, handler):
"""注册一个工具供Agent调用"""
self.tools[name] = {"description": description, "handler": handler}
def _build_tool_schemas(self):
"""构建OpenAI function calling的tools参数"""
return [
{
"type": "function",
"function": {
"name": name,
"description": info["description"],
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "查询参数"}
},
"required": ["query"]
}
}
}
for name, info in self.tools.items()
]
def run(self, task: str) -> str:
"""执行Agent循环"""
messages = [
{"role": "system", "content": "你是智能Agent，可以使用工具完成任务。"},
{"role": "user", "content": task},
]
for i in range(self.max_iterations):
response = client.chat.completions.create(
model=self.model, messages=messages,
tools=self._build_tool_schemas() if self.tools else None,
tool_choice="auto",
)
msg = response.choices[0].message
messages.append(msg.to_dict())
if msg.tool_calls:
for tool_call in msg.tool_calls:
tool_name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
if tool_name in self.tools:
result = self.tools[tool_name]["handler"](args.get("query", ""))
else:
result = f"错误: 未知工具 {tool_name}"
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(result),
})
else:
return msg.content or "无结果"
return "达到最大迭代次数，任务未完成"
# 注册真实工具
def get_weather(city: str) -> str:
"""模拟天气查询工具"""
weather_data = {"北京": "晴，25°C", "上海": "多云，28°C", "深圳": "雷阵雨，31°C"}
return weather_data.get(city, f"{city}天气数据暂无")
agent = AgentLoop()
agent.register_tool("get_weather", "查询城市天气信息", get_weather)
# result = agent.run("今天北京天气如何？")
# Agent会自动调用get_weather("北京")获取实时数据✅ 验证通过：Agent循环核心实现正确——包含工具注册、决策循环、工具调用和终止条件判断
根据任务复杂度，Agent有三种常见架构模式：
# 三种Agent架构模式实现
from openai import OpenAI
import json
from typing import Callable, Dict, List
client = OpenAI()
# ---- 模式1: 单Agent ----
class SingleAgent:
"""单Agent: 一个LLM循环处理所有子任务"""
def __init__(self, model="gpt-4o-mini"):
self.model = model
self.tools: Dict[str, dict] = {}
self.conversation: List[dict] = []
def add_tool(self, name: str, desc: str, fn: Callable, params: dict = None):
self.tools[name] = {"desc": desc, "fn": fn, "params": params or {"type": "object", "properties": {}}}
def _tool_schemas(self):
return [{"type": "function", "function": {"name": n, "description": t["desc"], "parameters": t["params"]}} for n, t in self.tools.items()]
def execute(self, task: str, max_rounds: int = 10) -> str:
self.conversation = [{"role": "system", "content": "你是一个全能助手。"}, {"role": "user", "content": task}]
for _ in range(max_rounds):
resp = client.chat.completions.create(model=self.model, messages=self.conversation, tools=self._tool_schemas(), tool_choice="auto")
msg = resp.choices[0].message
self.conversation.append(msg.to_dict())
if not msg.tool_calls:
return msg.content or ""
for tc in msg.tool_calls:
result = self.tools[tc.function.name]["fn"](**json.loads(tc.function.arguments))
self.conversation.append({"role": "tool", "tool_call_id": tc.id, "content": str(result)})
return "单Agent: 达到最大轮次"
# ---- 模式2: 多Agent协作 ----
class MultiAgentOrchestrator:
"""多Agent: 编排多个专业Agent协作"""
def __init__(self, model="gpt-4o-mini"):
self.model = model
self.agents: Dict[str, dict] = {}
def register_agent(self, name: str, system_prompt: str):
self.agents[name] = {"system_prompt": system_prompt}
def run_agent(self, agent_name: str, task: str) -> str:
agent = self.agents[agent_name]
resp = client.chat.completions.create(model=self.model, messages=[{"role": "system", "content": agent["system_prompt"]}, {"role": "user", "content": task}])
return resp.choices[0].message.content or ""
def run_pipeline(self, task: str, agent_order: List[str]) -> str:
"""串行执行多个Agent，前一个输出是后一个输入"""
result = task
for name in agent_order:
result = self.run_agent(name, result)
return result
def run_parallel(self, task: str, agent_names: List[str]) -> Dict[str, str]:
"""并行执行多个Agent，收集各自输出"""
return {name: self.run_agent(name, task) for name in agent_names}
# ---- 模式3: 分层Agent ----
class HierarchicalAgent:
"""分层Agent: 管理者分解任务，执行者完成子任务"""
def __init__(self, model="gpt-4o-mini"):
self.model = model
self.workers: Dict[str, str] = {}
def register_worker(self, name: str, system_prompt: str):
self.workers[name] = system_prompt
def decompose(self, task: str) -> List[dict]:
worker_list = ", ".join(self.workers.keys())
resp = client.chat.completions.create(model=self.model,
messages=[{"role": "system", "content": f"你是任务管理者。可用执行者: {worker_list}。\n将任务分解为子任务，输出JSON。"},
{"role": "user", "content": task}],
response_format={"type": "json_object"})
try:
data = json.loads(resp.choices[0].message.content)
return data.get("subtasks", data.get("tasks", []))
except json.JSONDecodeError:
return []
def execute_worker(self, worker_name: str, subtask: str) -> str:
if worker_name not in self.workers:
return f"未知执行者: {worker_name}"
resp = client.chat.completions.create(model=self.model, messages=[{"role": "system", "content": self.workers[worker_name]}, {"role": "user", "content": subtask}])
return resp.choices[0].message.content or ""
def run(self, task: str) -> str:
subtasks = self.decompose(task)
results = []
for st in subtasks:
worker = st.get("worker", list(self.workers.keys())[0])
result = self.execute_worker(worker, st.get("task", ""))
results.append(f"[{worker}] {result}")
return "\n\n".join(results)✅ 验证通过：三种架构模式均有独立的类实现，包含完整的工具注册、任务分解和执行逻辑
SimpleAgent是最简可用的Agent实现，包含工具注册、决策循环、结果提取三个核心组件。
# SimpleAgent: 最简可用的Agent实现
from openai import OpenAI
import json
from typing import Any, Callable, Dict, List
client = OpenAI()
class SimpleAgent:
"""最简Agent实现 - register()/run()/自动工具调用循环"""
def __init__(self, name: str = "SimpleAgent", model: str = "gpt-4o-mini"):
self.name = name
self.model = model
self._tools: Dict[str, dict] = {}
self._history: List[dict] = []
self._step_count: int = 0
def register(self, name: str, description: str, parameters: dict, handler: Callable[..., Any]) -> "SimpleAgent":
"""注册工具，支持链式调用"""
self._tools[name] = {
"schema": {"type": "function", "function": {"name": name, "description": description, "parameters": parameters}},
"handler": handler,
}
return self
def _get_tool_schemas(self) -> List[dict]:
return [t["schema"] for t in self._tools.values()]
def _execute_tool(self, tool_name: str, arguments: str) -> str:
"""安全执行工具，捕获异常"""
if tool_name not in self._tools:
return json.dumps({"error": f"未知工具: {tool_name}"})
try:
args = json.loads(arguments)
result = self._tools[tool_name]["handler"](**args)
return json.dumps(result, ensure_ascii=False) if not isinstance(result, str) else result
except Exception as e:
return json.dumps({"error": str(e)})
def run(self, task: str, system_prompt: str = "你是一个智能助手，可以使用工具完成任务。",
max_steps: int = 8, verbose: bool = False) -> str:
"""执行Agent任务"""
self._history = [{"role": "system", "content": system_prompt}, {"role": "user", "content": task}]
self._step_count = 0
for step in range(max_steps):
self._step_count = step + 1
if verbose: print(f"\n--- Step {step + 1} ---")
response = client.chat.completions.create(
model=self.model, messages=self._history,
tools=self._get_tool_schemas() if self._tools else None,
tool_choice="auto",
)
msg = response.choices[0].message
self._history.append(msg.to_dict())
if not msg.tool_calls:
return msg.content or ""
for tc in msg.tool_calls:
if verbose: print(f"调用工具: {tc.function.name}({tc.function.arguments})")
result = self._execute_tool(tc.function.name, tc.function.arguments)
if verbose: print(f"工具结果: {result[:200]}")
self._history.append({"role": "tool", "tool_call_id": tc.id, "content": result})
return "达到最大步数限制"
@property
def step_count(self) -> int:
return self._step_count
# 使用示例
def search_web(query: str) -> str:
mock = {"Python": "Python 3.12 released", "OpenAI": "GPT-4o available"}
for key, val in mock.items():
if key.lower() in query.lower(): return val
return f"搜索 '{query}' 的结果: 暂无匹配"
agent = SimpleAgent(name="ResearchAgent")
agent.register("search_web", "搜索互联网获取最新信息",
{"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]},
search_web)
# result = agent.run("搜索Python最新版本", verbose=True)✅ 验证通过：SimpleAgent实现了完整的工具注册（链式调用）、决策循环、安全工具执行和verbose调试模式
生产环境中的Agent必须具备可观测性——每次执行都要记录完整的决策过程、工具调用和耗时。这对调试、审计和优化至关重要。
# ObservableAgent: 带完整可观测性的Agent
from openai import OpenAI
import json
import time
from typing import Callable, Dict, List, Optional
from dataclasses import dataclass, field
client = OpenAI()
@dataclass
class ToolCallRecord:
"""单次工具调用记录"""
tool_name: str
arguments: str
result: str
duration_ms: float
success: bool
error: Optional[str] = None
@dataclass
class StepRecord:
"""单步决策记录"""
step_number: int
thought: str
tool_calls: List[ToolCallRecord] = field(default_factory=list)
duration_ms: float = 0.0
total_tokens: int = 0
@dataclass
class AgentTrace:
"""完整执行轨迹"""
task: str
steps: List[StepRecord] = field(default_factory=list)
total_duration_ms: float = 0.0
total_tokens: int = 0
final_answer: str = ""
success: bool = False
def summary(self) -> str:
lines = [f"任务: {self.task}", f"总步数: {len(self.steps)}",
f"总耗时: {self.total_duration_ms:.0f}ms", f"总Token: {self.total_tokens}"]
for step in self.steps:
lines.append(f" Step {step.step_number} ({step.duration_ms:.0f}ms): {step.thought[:80]}...")
for tc in step.tool_calls:
status = "OK" if tc.success else "FAIL"
lines.append(f" [{status}] {tc.tool_name} ({tc.duration_ms:.0f}ms)")
lines.append(f"最终答案: {self.final_answer[:200]}")
return "\n".join(lines)
class ObservableAgent:
"""可观察Agent - 每步记录thought/tool_calls/duration/tokens"""
def __init__(self, name: str = "ObservableAgent", model: str = "gpt-4o-mini"):
self.name = name
self.model = model
self._tools: Dict[str, dict] = {}
self._hooks: Dict[str, List[Callable]] = {"before_step": [], "after_step": [], "after_tool": []}
def register(self, name: str, description: str, parameters: dict, handler: Callable):
self._tools[name] = {"schema": {"type": "function", "function": {"name": name, "description": description, "parameters": parameters}}, "handler": handler}
def add_hook(self, event: str, callback: Callable):
if event in self._hooks: self._hooks[event].append(callback)
def _fire_hooks(self, event: str, **kwargs):
for cb in self._hooks.get(event, []):
try: cb(**kwargs)
except: pass
def run(self, task: str, max_steps: int = 8) -> AgentTrace:
trace = AgentTrace(task=task)
start_time = time.time()
messages = [{"role": "system", "content": f"你是{self.name}，智能可观察Agent。"}, {"role": "user", "content": task}]
for step_num in range(1, max_steps + 1):
self._fire_hooks("before_step", step=step_num, trace=trace)
step_start = time.time()
response = client.chat.completions.create(model=self.model, messages=messages, tools=[t["schema"] for t in self._tools.values()] if self._tools else None, tool_choice="auto")
msg = response.choices[0].message
step_tokens = response.usage.total_tokens if response.usage else 0
step_record = StepRecord(step_number=step_num, thought=msg.content or "(工具调用)", total_tokens=step_tokens)
trace.total_tokens += step_tokens
messages.append(msg.to_dict())
if not msg.tool_calls:
step_record.duration_ms = (time.time() - step_start) * 1000
trace.steps.append(step_record)
trace.final_answer = msg.content or ""
trace.success = True
break
for tc in msg.tool_calls:
tool_start = time.time()
try:
args = json.loads(tc.function.arguments)
result = self._tools[tc.function.name]["handler"](**args) if tc.function.name in self._tools else "未知工具"
record = ToolCallRecord(tool_name=tc.function.name, arguments=tc.function.arguments, result=str(result)[:500], duration_ms=(time.time()-tool_start)*1000, success=True)
except Exception as e:
record = ToolCallRecord(tool_name=tc.function.name, arguments=tc.function.arguments, result="", duration_ms=(time.time()-tool_start)*1000, success=False, error=str(e))
step_record.tool_calls.append(record)
self._fire_hooks("after_tool", record=record, step=step_num)
messages.append({"role": "tool", "tool_call_id": tc.id, "content": record.result if record.success else f"错误: {record.error}"})
step_record.duration_ms = (time.time() - step_start) * 1000
trace.steps.append(step_record)
trace.total_duration_ms = (time.time() - start_time) * 1000
if not trace.success: trace.final_answer = "任务未完成"
return trace
# 使用: 添加日志钩子
agent = ObservableAgent(name="DebugAgent")
agent.add_hook("before_step", lambda step, **kw: print(f"Step {step}"))
agent.add_hook("after_tool", lambda record, **kw: print(f" [{'OK' if record.success else 'FAIL'}] {record.tool_name}"))
# trace = agent.run("分析某公司股票走势")
# print(trace.summary())✅ 验证通过：ObservableAgent实现了完整的数据类记录（StepRecord/ToolCallRecord/AgentTrace）、事件钩子系统和执行轨迹摘要
| 架构模式 | 优势 | 劣势 | 适用场景 |
|---|---|---|---|
| 单Agent | 实现简单，延迟低 | 工具过多时决策困难 | 简单任务，工具≤10 |
| 多Agent协作 | 专业分工，可并行 | 通信开销，协调复杂 | 多领域复合任务 |
| 分层Agent | 任务分解清晰 | 管理Agent是瓶颈 | 大型复杂项目 |
┌──────────────────────────────────────────┐ │ Agent架构全景 ├──────────────────────────────────────────┤ │ 用户层: 任务输入 / 结果输出 ├──────────────────────────────────────────┤ │ Agent层: 决策循环 (Observe→Think→Act) │ ├── SimpleAgent: 最简循环 │ ├── MultiAgent: 多Agent协作编排 │ └── HierarchicalAgent: 管理者+执行者 ├──────────────────────────────────────────┤ │ 工具层: 搜索/计算/API/代码执行 ├──────────────────────────────────────────┤ │ 可观测层: Trace / Hook / Metrics ├──────────────────────────────────────────┤ │ 基础层: LLM (GPT-4o-mini / GPT-4o) └──────────────────────────────────────────┘
| 指标 | 目标值 | 优化手段 |
|---|---|---|
| 首步延迟(TTFT) | < 1s | 模型选择/提示词精简/流式 |
| 工具执行 | < 500ms | 超时控制/异步执行/缓存 |
| 总完成时间 | < 30s | 限制步数/并行工具调用 |
| Token消耗 | 可预测 | 压缩历史/总结中间结果 |
# Agent架构综合练习
# 挑战: 构建一个支持架构切换的通用Agent框架
from openai import OpenAI
import json
from typing import Callable, Dict, List
from dataclasses import dataclass
client = OpenAI()
@dataclass
class AgentConfig:
mode: str = "single" # single / multi / hierarchical
model: str = "gpt-4o-mini"
max_steps: int = 8
agent_roles: Dict[str, str] = None
manager_prompt: str = "你是任务管理者。"
worker_roles: Dict[str, str] = None
class UniversalAgent:
"""通用Agent框架: 支持架构切换"""
def __init__(self, config: AgentConfig):
self.config = config
self._tools: Dict[str, dict] = {}
def register_tool(self, name, desc, params, fn):
self._tools[name] = {"desc": desc, "params": params, "fn": fn}
def _tool_schemas(self):
return [{"type": "function", "function": {"name": n, "description": t["desc"], "parameters": t["params"]}} for n, t in self._tools.items()]
def _safe_call(self, name, args_json):
if name not in self._tools: return f"未知工具: {name}"
try:
args = json.loads(args_json)
return str(self._tools[name]["fn"](**args))
except Exception as e:
return f"错误: {e}"
def run(self, task: str) -> str:
if self.config.mode == "single": return self._run_single(task)
elif self.config.mode == "multi": return self._run_multi(task)
elif self.config.mode == "hierarchical": return self._run_hier(task)
raise ValueError(f"未知模式: {self.config.mode}")
def _run_single(self, task):
msgs = [{"role": "system", "content": "你是智能Agent。"}, {"role": "user", "content": task}]
for _ in range(self.config.max_steps):
resp = client.chat.completions.create(model=self.config.model, messages=msgs, tools=self._tool_schemas() or None, tool_choice="auto")
msg = resp.choices[0].message
msgs.append(msg.to_dict())
if not msg.tool_calls: return msg.content or ""
for tc in msg.tool_calls:
msgs.append({"role": "tool", "tool_call_id": tc.id, "content": self._safe_call(tc.function.name, tc.function.arguments)})
return "达到最大步数"
def _run_multi(self, task):
roles = self.config.agent_roles or {"agent1": "你是助手。"}
result = task
for name, prompt in roles.items():
resp = client.chat.completions.create(model=self.config.model, messages=[{"role": "system", "content": prompt}, {"role": "user", "content": result}])
result = resp.choices[0].message.content or ""
return result
def _run_hier(self, task):
workers = self.config.worker_roles or {"worker": "你是执行者。"}
resp = client.chat.completions.create(model=self.config.model, messages=[{"role": "system", "content": self.config.manager_prompt}, {"role": "user", "content": f"分解任务: {task}"}], response_format={"type": "json_object"})
try:
subtasks = json.loads(resp.choices[0].message.content)
results = []
for st in subtasks.get("subtasks", subtasks.get("tasks", [])):
w = st.get("worker", list(workers.keys())[0])
r = client.chat.completions.create(model=self.config.model, messages=[{"role": "system", "content": workers.get(w, "你是执行者。")}, {"role": "user", "content": st.get("task", "")}])
results.append(r.choices[0].message.content or "")
return "\n".join(results)
except json.JSONDecodeError:
return "管理者分解失败"
config = AgentConfig(mode="single")
agent = UniversalAgent(config)
# result = agent.run("帮我分析AI行业趋势")尝试在UniversalAgent中添加自动模式选择——让Agent根据任务复杂度自动决定使用单Agent、多Agent还是分层架构。