AI Agent 架构与工程化落地
#Agent · #ReAct · #FunctionCalling · #工具调用 · #状态机 · #多轮对话 · #长期记忆 · #Memory · #多模态Agent · #ComputerUse · #CodeAgent · #Go · #Python
用户在 HTML 页面上点击一个按钮，后端 Agent 如何运转？本专题覆盖 ReAct 智能体循环、严谨的 Function Calling 设计、以及多轮对话状态管理。
什么是 AI Agent？
AI Agent（智能体）是一个能够自主感知环境、制定计划、调用工具、执行行动的 AI 系统。与传统的"一问一答"不同，Agent 会经历"思考→行动→观察→思考"的循环。
mermaid
graph TD
USER["用户任务"] --> AGENT["🤖 AI Agent"]
AGENT --> THINK["💭 思考<br/>我需要什么工具？"]
THINK --> ACT["🔧 行动<br/>调用 API / 执行代码"]
ACT --> OBSERVE["👁️ 观察<br/>分析执行结果"]
OBSERVE --> THINK
OBSERVE --> DONE["✅ 完成任务<br/>返回最终答案"]
style AGENT fill:#9b59b6,color:#fff
style THINK fill:#f39c12,color:#fff
style ACT fill:#2ecc71,color:#fffAgent 的核心能力三角
| 能力 | 说明 | 技术实现 |
|---|---|---|
| 推理 (Reasoning) | 理解任务、拆解步骤、制定计划 | ReAct / CoT / Tree-of-Thought |
| 行动 (Acting) | 调用工具执行具体操作 | Function Calling / Tool Use |
| 记忆 (Memory) | 多轮对话上下文维护 | Session 管理 / 向量数据库 / 摘要 |
基于 ReAct 的智能体循环
ReAct 原理
ReAct（Reasoning + Acting）是 Agent 最经典的范式。模型交替进行思考（Thought）→ 行动（Action）→ 观察（Observation），直到完成任务。
mermaid
sequenceDiagram
participant User as 👤 用户
participant Agent as 🤖 Agent 核心
participant LLM as 🧠 LLM 模型
participant Tool as 🔧 工具/API
User->>Agent: "帮我查下今天深圳天气，并给我穿衣建议"
loop ReAct 循环
Agent->>LLM: 发送 Prompt + 工具列表 + 历史
LLM-->Agent: Thought: 需要先获取天气
LLM-->Agent: Action: get_weather("深圳")
Agent->>Tool: 调用 get_weather API
Tool-->>Agent: Observation: 多云 28°C
Agent->>LLM: 追加 Observation 到上下文
LLM-->Agent: Thought: 天气已获取，可以给出建议
LLM-->Agent: Action: FINISH("多云28°C，建议穿短袖...")
end
Agent-->>User: 返回最终答案使用 Go 实现 ReAct 状态机
go
package agent
import (
"context"
"encoding/json"
"fmt"
"strings"
)
// ReActState 定义 Agent 在 ReAct 循环中的状态
type ReActState int
const (
StateIdle ReActState = iota // 空闲，等待用户输入
StateThinking // 正在推理
StateActing // 正在执行工具调用
StateObserving // 正在观察工具结果
StateFinished // 任务完成
StateError // 出错
)
func (s ReActState) String() string {
return [...]string{"Idle", "Thinking", "Acting", "Observing", "Finished", "Error"}[s]
}
// ReActStep 表示 ReAct 循环中的一步
// 对应 Prompt 驱动模型输出的结构：
// Thought: <推理文本>
// Action: <工具名>(<参数JSON>)
// 或 Action: FINISH(<最终答案>)
type ReActStep struct {
Thought string `json:"thought"` // 模型的推理过程
Action string `json:"action"` // 工具名 或 "FINISH"
Input string `json:"input"` // 工具参数 (JSON) 或最终答案
}
// Tool 定义 Agent 可调用的工具接口
type Tool interface {
Name() string
Description() string
// JSONSchema 返回工具参数的 JSON Schema，用于 LLM 生成正确参数
JSONSchema() json.RawMessage
// Execute 执行工具，返回观测结果
Execute(ctx context.Context, input json.RawMessage) (string, error)
}
// ReActAgent 实现了 ReAct 智能体循环
type ReActAgent struct {
llm LLMClient // LLM 客户端（OpenAI 兼容接口）
tools map[string]Tool // 已注册的工具
config AgentConfig // 配置
state ReActState // 当前状态
steps []ReActStep // 历史步骤记录
}
// AgentConfig Agent 的行为配置
type AgentConfig struct {
MaxSteps int // 最大循环步数，防止无限循环（默认 10）
Temperature float32 // LLM 温度（0 = 确定，1 = 创造）
SystemPrompt string // 系统提示词
Verbose bool // 是否打印详细日志
}
// NewReActAgent 创建一个新的 ReAct Agent
func NewReActAgent(llm LLMClient, config AgentConfig) *ReActAgent {
if config.MaxSteps == 0 {
config.MaxSteps = 10
}
return &ReActAgent{
llm: llm,
tools: make(map[string]Tool),
config: config,
}
}
// RegisterTool 注册一个工具到 Agent
func (a *ReActAgent) RegisterTool(t Tool) {
a.tools[t.Name()] = t
}
// Run 执行完整的 ReAct 循环，处理用户请求
func (a *ReActAgent) Run(ctx context.Context, userInput string) (string, error) {
// 初始化上下文：系统提示 + 用户输入
messages := a.buildInitialMessages(userInput)
for step := 0; step < a.config.MaxSteps; step++ {
a.state = StateThinking
// 1. 调用 LLM 生成 Thought + Action
response, err := a.llm.Chat(ctx, messages)
if err != nil {
a.state = StateError
return "", fmt.Errorf("LLM 调用失败(step %d): %w", step, err)
}
// 2. 解析 LLM 输出为 ReAct 格式
reactStep, isFinal, err := a.parseResponse(response)
if err != nil {
a.state = StateError
return "", fmt.Errorf("解析 LLM 输出失败(step %d): %w", step, err)
}
a.steps = append(a.steps, reactStep)
if isFinal {
a.state = StateFinished
return reactStep.Input, nil // FINISH 的内容就是最终答案
}
// 3. 执行工具调用
a.state = StateActing
tool, ok := a.tools[reactStep.Action]
if !ok {
observation := fmt.Sprintf("错误: 未找到工具 '%s'。可用工具: %v",
reactStep.Action, a.toolNames())
messages = append(messages, assistantMsg(reactStep), userMsg(observation))
continue
}
observation, err := tool.Execute(ctx, json.RawMessage(reactStep.Input))
if err != nil {
observation = fmt.Sprintf("工具执行错误: %v", err)
}
// 4. 追加 Observation 到对话历史
a.state = StateObserving
messages = append(messages,
assistantMsg(reactStep),
userMsg(fmt.Sprintf("Observation: %s", observation)),
)
if a.config.Verbose {
fmt.Printf("[Step %d] Thought: %s\n Action: %s(%s)\n Observation: %s\n",
step, reactStep.Thought, reactStep.Action, reactStep.Input, observation)
}
}
return "", fmt.Errorf("超过最大步数限制 (%d)", a.config.MaxSteps)
}
// buildInitialMessages 构建初始 Prompt
//
// ReAct Prompt 模板设计是关键——这里通过 Markdown 层级结构
// 精准约束 LLM 的输出格式
func (a *ReActAgent) buildInitialMessages(userInput string) []Message {
systemPrompt := fmt.Sprintf(`你是一个 AI 智能体，通过交替进行思考和行动来解决用户任务。
## 规则
1. 每轮输出必须严格遵循以下格式：
Thought: <你的推理过程>
Action: <工具名>(<JSON参数>) 或 Action: FINISH(<最终答案>)
2. 一次只能调用一个工具
3. 当任务完成时，使用 FINISH 给出最终答案
4. 如果工具执行失败，分析错误并尝试其他方法
## 可用工具
%s`, a.formatToolDescriptions())
return []Message{
{Role: "system", Content: systemPrompt},
{Role: "user", Content: userInput},
}
}
// parseResponse 解析 LLM 输出
func (a *ReActAgent) parseResponse(response string) (step ReActStep, isFinal bool, err error) {
// 提取 Thought
thoughtStart := strings.Index(response, "Thought:")
actionStart := strings.Index(response, "Action:")
if thoughtStart == -1 || actionStart == -1 {
return step, false, fmt.Errorf("缺少 Thought: 或 Action: 字段:\n%s", response)
}
step.Thought = strings.TrimSpace(response[thoughtStart+9 : actionStart])
actionPart := strings.TrimSpace(response[actionStart+8:])
// 检查是否是 FINISH
if strings.HasPrefix(actionPart, "FINISH(") && strings.HasSuffix(actionPart, ")") {
finalAnswer := actionPart[7 : len(actionPart)-1]
return ReActStep{
Thought: step.Thought,
Action: "FINISH",
Input: finalAnswer,
}, true, nil
}
// 解析工具调用: ToolName({"key": "value"})
parenIdx := strings.Index(actionPart, "(")
if parenIdx == -1 || !strings.HasSuffix(actionPart, ")") {
return step, false, fmt.Errorf("工具调用格式错误（应为 ToolName({...})）:\n%s", actionPart)
}
step.Action = actionPart[:parenIdx]
step.Input = actionPart[parenIdx+1 : len(actionPart)-1]
return step, false, nil
}
// 辅助方法
func (a *ReActAgent) formatToolDescriptions() string {
var sb strings.Builder
for _, t := range a.tools {
sb.WriteString(fmt.Sprintf("- **%s**: %s\n", t.Name(), t.Description()))
// 如果工具提供了 JSON Schema，一并展示
if schema := t.JSONSchema(); len(schema) > 0 {
sb.WriteString(fmt.Sprintf(" 参数格式: ```json\n %s\n ```\n", string(schema)))
}
}
return sb.String()
}
func (a *ReActAgent) toolNames() []string {
names := make([]string, 0, len(a.tools))
for name := range a.tools {
names = append(names, name)
}
return names
}
// 消息辅助函数
func assistantMsg(s ReActStep) Message {
return Message{Role: "assistant", Content: fmt.Sprintf(
"Thought: %s\nAction: %s(%s)", s.Thought, s.Action, s.Input)}
}
func userMsg(content string) Message {
return Message{Role: "user", Content: content}
}
// 类型定义
type Message struct {
Role string `json:"role"`
Content string `json:"content"`
}
// LLMClient 抽象 LLM 接口（实现可使用 OpenAI / Ollama / vLLM 等后端）
type LLMClient interface {
Chat(ctx context.Context, messages []Message) (string, error)
}使用 Python 实现 ReAct Agent
python
"""
ReAct Agent 的 Python 实现（使用 OpenAI 兼容接口）
"""
import json
import re
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional
from abc import ABC, abstractmethod
# --- 工具抽象 ---
class Tool(ABC):
"""工具基类，所有 Agent 工具需继承此类"""
@property
@abstractmethod
def name(self) -> str:
"""工具名称"""
...
@property
@abstractmethod
def description(self) -> str:
"""工具描述（会写入 Prompt 中）"""
...
@property
@abstractmethod
def parameters(self) -> Dict[str, Any]:
"""JSON Schema 格式的参数定义"""
...
@abstractmethod
async def execute(self, **kwargs) -> str:
"""执行工具，返回字符串结果"""
...
# --- ReAct Agent 实现 ---
@dataclass
class ReActStep:
"""ReAct 循环中的一步"""
thought: str # 推理过程
action: str # 工具名或 "FINISH"
input: str # 工具参数（JSON 字符串）或最终答案
@dataclass
class AgentConfig:
"""Agent 配置"""
max_steps: int = 10 # 最大步数
temperature: float = 0.7 # 模型温度
system_prompt: str = "" # 系统提示词
verbose: bool = True # 是否打印日志
class ReActAgent:
"""
基于 ReAct 范式的智能体
工作原理：
1. 接收用户任务
2. 构建 Prompt（系统提示 + 工具列表 + 用户输入）
3. 循环调用 LLM：
- LLM 输出 Thought + Action
- 解析后调用对应工具
- 将 Observation 追加到上下文
4. LLM 输出 FINISH 时结束循环
"""
def __init__(
self,
llm_client: Any, # OpenAI 兼容客户端
config: AgentConfig = None,
):
self.llm = llm_client
self.config = config or AgentConfig()
self.tools: Dict[str, Tool] = {}
self.steps: List[ReActStep] = []
def register_tool(self, tool: Tool):
"""注册工具"""
self.tools[tool.name] = tool
async def run(self, user_input: str) -> str:
"""执行 ReAct 循环"""
# 构建初始消息
messages = self._build_initial_messages(user_input)
for step_num in range(self.config.max_steps):
if self.config.verbose:
print(f"\n{'='*50}")
print(f"🔄 Step {step_num + 1}/{self.config.max_steps}")
# 1. 调用 LLM
response = await self._call_llm(messages)
# 2. 解析输出
react_step, is_final = self._parse_response(response)
self.steps.append(react_step)
if is_final:
return react_step.input # FINISH 的内容 = 最终答案
# 3. 执行工具
tool = self.tools.get(react_step.action)
if tool is None:
observation = f"错误: 未找到工具 '{react_step.action}'。可用工具: {list(self.tools.keys())}"
else:
try:
params = json.loads(react_step.input) if react_step.input else {}
observation = await tool.execute(**params)
except json.JSONDecodeError:
observation = f"错误: 工具参数 JSON 解析失败: {react_step.input}"
except Exception as e:
observation = f"工具执行异常: {type(e).__name__}: {e}"
# 4. 追加到对话历史
messages.append({
"role": "assistant",
"content": f"Thought: {react_step.thought}\nAction: {react_step.action}({react_step.input})"
})
messages.append({"role": "user", "content": f"Observation: {observation}"})
if self.config.verbose:
print(f" 💭 Thought: {react_step.thought[:100]}...")
print(f" 🔧 Action: {react_step.action}({react_step.input[:100]}...)")
print(f" 👁️ Observation: {observation[:100]}...")
raise RuntimeError(f"Agent 超过最大步数限制 ({self.config.max_steps})")
def _build_initial_messages(self, user_input: str) -> List[Dict]:
"""构建 ReAct Prompt —— 层次化 Markdown 模板"""
tools_desc = ""
for name, tool in self.tools.items():
params_schema = json.dumps(tool.parameters, ensure_ascii=False, indent=2)
tools_desc += f"- **{name}**: {tool.description}\n"
tools_desc += f" ```json\n {params_schema}\n ```\n"
system_prompt = f"""你是一个 AI 智能体，通过交替进行思考和行动来解决用户任务。
## 输出格式（严格遵守）
每轮输出必须包含：Thought: <你对当前状况的分析和下一步计划> Action: <工具名>(<JSON格式参数>) 或 Action: FINISH(<最终答案>)
## 规则
1. 一次只能调用一个工具
2. 必须在 Action 中给出合法的 JSON 参数
3. 使用 FINISH 报告最终结果
4. 如果出错，分析原因并调整策略
## 可用工具
{tools_desc}
"""
return [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_input},
]
def _parse_response(self, response: str) -> tuple:
"""解析 LLM 输出"""
# 提取 Thought
thought_match = re.search(r'Thought:\s*(.+?)(?=\nAction:|$)', response, re.DOTALL)
action_match = re.search(r'Action:\s*(.+)$', response, re.MULTILINE)
if not thought_match or not action_match:
raise ValueError(f"无法解析 LLM 输出:\n{response}")
thought = thought_match.group(1).strip()
action_text = action_match.group(1).strip()
# 检查 FINISH
finish_match = re.match(r'FINISH\((.+)\)', action_text, re.DOTALL)
if finish_match:
return ReActStep(thought=thought, action="FINISH", input=finish_match.group(1)), True
# 解析工具调用: ToolName({"key": "value"})
tool_match = re.match(r'(\w+)\((.+)\)', action_text, re.DOTALL)
if not tool_match:
raise ValueError(f"无法解析工具调用格式: {action_text}")
return ReActStep(
thought=thought,
action=tool_match.group(1),
input=tool_match.group(2),
), False
async def _call_llm(self, messages: List[Dict]) -> str:
"""调用 LLM（OpenAI 兼容接口）"""
# 注意: 这里需要配合实际的 LLM 客户端实现
# 例如 openai.AsyncOpenAI().chat.completions.create(...)
raise NotImplementedError("需要实现具体的 LLM 调用逻辑")严谨的 Function Calling 设计
JSON Schema 工具定义
将后端 API 抽象为标准的 JSON Schema，让 LLM 能精确生成调用参数。
python
"""
严谨的 Function Calling 工具定义示例
核心原则：
1. Schema 要精确（字段、类型、约束都明确）
2. Description 要清晰（告诉模型何时用、怎么用）
3. 必填项要明确标记
"""
GET_WEATHER_TOOL = {
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的实时天气信息。当用户询问天气、气温、降雨等情况时使用此工具。",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，支持中英文（如：'深圳'、'Shenzhen'）",
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位，默认为 celsius（摄氏度）",
},
},
"required": ["city"], # unit 是可选的，有默认值
},
},
}
EXECUTE_SQL_TOOL = {
"type": "function",
"function": {
"name": "execute_sql",
"description": (
"在数据库上执行只读 SQL 查询。只支持 SELECT 语句。"
"数据库包含以下表：users(id, name, email), orders(id, user_id, amount, created_at)"
),
"parameters": {
"type": "object",
"properties": {
"sql": {
"type": "string",
"description": "待执行的 SELECT SQL 语句。必须是完整有效的 SQL。禁止 DDL/DML。",
},
"limit": {
"type": "integer",
"minimum": 1,
"maximum": 100,
"description": "最大返回行数，默认 20",
},
},
"required": ["sql"],
},
},
}JSON 容错与自动修复
LLM 输出的 JSON 经常有瑕疵——多余的逗号、缺失引号、注释等。后端的自动修复机制至关重要：
go
package functioncalling
import (
"encoding/json"
"fmt"
"regexp"
"strings"
)
// RepairJSON 自动修复 LLM 输出的损坏 JSON
//
// 常见 LLM JSON 错误：
// 1. 尾部多余逗号: {"name": "test",}
// 2. 缺少引号的 key: {name: "test"}
// 3. 单引号: {'name': 'test'}
// 4. 注释混入: {"name": "test" /* comment */}
// 5. 多余文本包围: 结果是 {"name": "test"}，请检查
// 6. 数组尾部多余逗号: [1, 2, 3,]
func RepairJSON(raw string) (string, error) {
// Step 1: 提取 JSON 部分（去除 Markdown 代码块包裹）
raw = extractJSONBlock(raw)
// Step 2: 去除注释
raw = removeComments(raw)
// Step 3: 替换单引号为双引号（注意字符串内的单引号）
raw = normalizeQuotes(raw)
// Step 4: 修复无引号的 key
raw = quoteUnquotedKeys(raw)
// Step 5: 去除尾部多余逗号
raw = removeTrailingCommas(raw)
// Step 6: 验证修复结果
var result interface{}
if err := json.Unmarshal([]byte(raw), &result); err != nil {
return raw, fmt.Errorf("JSON 修复后仍无效: %w\n修复结果: %s", err, raw)
}
return raw, nil
}
// extractJSONBlock 从 LLM 输出中提取 JSON
func extractJSONBlock(raw string) string {
// 尝试匹配 ```json ... ``` 或 ``` ... ```
re := regexp.MustCompile("(?s)```(?:json)?\\s*\\n?(.+?)\\n?```")
if matches := re.FindStringSubmatch(raw); len(matches) > 1 {
return strings.TrimSpace(matches[1])
}
// 尝试找到第一个 { 和最后一个 }
start := strings.Index(raw, "{")
end := strings.LastIndex(raw, "}")
if start != -1 && end > start {
return raw[start : end+1]
}
// 尝试找到第一个 [ 和最后一个 ]
start = strings.Index(raw, "[")
end = strings.LastIndex(raw, "]")
if start != -1 && end > start {
return raw[start : end+1]
}
return raw
}
// removeComments 去除 JSON 中的注释
func removeComments(raw string) string {
// 去除行注释 //
reLine := regexp.MustCompile(`//[^\n]*`)
raw = reLine.ReplaceAllString(raw, "")
// 去除块注释 /* */
reBlock := regexp.MustCompile(`/\*[\s\S]*?\*/`)
raw = reBlock.ReplaceAllString(raw, "")
return raw
}
// normalizeQuotes 将单引号标准化为双引号
func normalizeQuotes(raw string) string {
// 简化处理：先将所有单引号替换为双引号
// 注意：这可能会破坏字符串内容中的单引号（如 "it's"）
// 生产环境需要更复杂的基于状态的解析
result := strings.ReplaceAll(raw, `'`, `"`)
return result
}
// quoteUnquotedKeys 为无引号的 JSON key 添加引号
func quoteUnquotedKeys(raw string) string {
// 匹配 { 或 , 后的无引号标识符:
// 如 {name: "test"} → {"name": "test"}
re := regexp.MustCompile(`([{,])\s*([a-zA-Z_]\w*)\s*:`)
return re.ReplaceAllString(raw, `$1"$2":`)
}
// removeTrailingCommas 去除尾部多余逗号
func removeTrailingCommas(raw string) string {
// 对象尾部: {"a": 1,} → {"a": 1}
reObj := regexp.MustCompile(`,(\s*[}\]])`)
raw = reObj.ReplaceAllString(raw, `$1`)
// 数组尾部: [1, 2,] → [1, 2]
return raw
}
// SafeCallTool 带容错的工具调用封装
//
// 当 LLM 输出损坏的 JSON 时，自动尝试最多 3 次修复与重试
func SafeCallTool(
toolName string,
rawArgs string,
execute func(args json.RawMessage) (string, error),
) (result string, err error) {
// 第一次尝试：直接解析
args, err := RepairJSON(rawArgs)
if err == nil {
result, callErr := execute(json.RawMessage(args))
if callErr == nil {
return result, nil
}
// 工具执行错误直接返回（不是 JSON 问题）
return "", fmt.Errorf("工具 %s 执行失败: %w", toolName, callErr)
}
// JSON 解析失败，返回详细的错误信息给 LLM，让它重新生成
return "", fmt.Errorf(
"工具 %s 的 JSON 参数无效。原始输入: %s\n错误: %w\n"+
"请修正 JSON 格式后重试。"+
"注意: key 必须用双引号包裹，字符串值必须用双引号，不能使用尾随逗号。",
toolName, rawArgs, err,
)
}容错重试的完整流程
mermaid
graph TD
LLM["LLM 输出 Tool Call"] --> PARSE{"JSON 解析成功?"}
PARSE -->|是| EXEC["执行工具"]
PARSE -->|否| FIX["自动修复 JSON<br/>(去注释、补引号、去尾部逗号)"]
FIX --> RETRY1{"第2次解析成功?"}
RETRY1 -->|是| EXEC
RETRY1 -->|否| ERR["返回详细错误给 LLM<br/>提示 JSON 格式要求"]
ERR --> LLM2["LLM 重新生成"]
LLM2 --> PARSE2{"第3次解析成功?"}
PARSE2 -->|是| EXEC
PARSE2 -->|否| FALLBACK["最终降级:<br/>返回错误并提示用户"]
style PARSE fill:#f39c12,color:#fff
style FIX fill:#2ecc71,color:#fff
style FALLBACK fill:#e74c3c,color:#fff状态管理与多轮对话持久化
会话生命周期
mermaid
stateDiagram-v2
[*] --> Created: 用户打开界面
Created --> Active: 发送第一条消息
Active --> Active: 多轮对话
Active --> Paused: 用户离开(超时)
Paused --> Active: 用户返回
Active --> Expired: 超时未活动(>30min)
Active --> Closed: 用户关闭
Expired --> [*]
Closed --> [*]使用 Redis 管理 Session
go
package session
import (
"context"
"encoding/json"
"fmt"
"time"
"github.com/redis/go-redis/v9"
)
// Session 表示一个用户的对话会话
type Session struct {
ID string `json:"id"` // Session ID (UUID)
UserID string `json:"user_id"` // 用户标识
Messages []Message `json:"messages"` // 对话历史（最近 N 轮）
Summary string `json:"summary"` // 早期对话的摘要（压缩长历史）
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
TTL time.Duration `json:"-"` // 过期时间
}
// SessionManager 使用 Redis 管理对话会话
type SessionManager struct {
rdb *redis.Client
maxHistory int // 最大保留的历史消息数（超过则压缩）
defaultTTL time.Duration // 默认 Session 过期时间
}
func NewSessionManager(rdb *redis.Client, maxHistory int) *SessionManager {
return &SessionManager{
rdb: rdb,
maxHistory: maxHistory,
defaultTTL: 30 * time.Minute, // 30 分钟无活动则过期
}
}
// CreateSession 创建新的对话会话
func (sm *SessionManager) CreateSession(ctx context.Context, userID string) (*Session, error) {
sessionID := generateUUID() // 使用 UUID 生成 session ID
session := &Session{
ID: sessionID,
UserID: userID,
Messages: []Message{},
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
TTL: sm.defaultTTL,
}
return session, sm.saveSession(ctx, session)
}
// AddMessage 向会话中追加一条消息
func (sm *SessionManager) AddMessage(ctx context.Context, sessionID string, msg Message) error {
session, err := sm.GetSession(ctx, sessionID)
if err != nil {
return err
}
session.Messages = append(session.Messages, msg)
// 如果消息过多，压缩早期对话为摘要
if len(session.Messages) > sm.maxHistory {
session.Messages, session.Summary = sm.compress(session)
}
session.UpdatedAt = time.Now()
return sm.saveSession(ctx, session)
}
// GetSession 获取会话
func (sm *SessionManager) GetSession(ctx context.Context, sessionID string) (*Session, error) {
data, err := sm.rdb.Get(ctx, sessionKey(sessionID)).Bytes()
if err == redis.Nil {
return nil, fmt.Errorf("session %s 不存在或已过期", sessionID)
}
if err != nil {
return nil, fmt.Errorf("读取 session 失败: %w", err)
}
var session Session
if err := json.Unmarshal(data, &session); err != nil {
return nil, fmt.Errorf("解析 session 失败: %w", err)
}
return &session, nil
}
// saveSession 持久化会话到 Redis
func (sm *SessionManager) saveSession(ctx context.Context, s *Session) error {
data, err := json.Marshal(s)
if err != nil {
return err
}
return sm.rdb.Set(ctx, sessionKey(s.ID), data, s.TTL).Err()
}
// compress 压缩历史对话：保留最近 N 条，将更早的压缩为摘要
func (sm *SessionManager) compress(s *Session) (recent []Message, summary string) {
keepCount := sm.maxHistory / 2
oldMessages := s.Messages[:len(s.Messages)-keepCount]
recent = s.Messages[len(s.Messages)-keepCount:]
// 将旧消息拼接为文本
var oldContent string
for _, m := range oldMessages {
oldContent += fmt.Sprintf("[%s]: %s\n", m.Role, m.Content)
}
// 合并已有摘要和新摘要
if s.Summary != "" {
summary = fmt.Sprintf("之前对话摘要:\n%s\n\n最近对话:\n%s", s.Summary, oldContent)
} else {
summary = fmt.Sprintf("对话历史摘要:\n%s", oldContent)
}
return recent, summary
}
// Lua 脚本：原子性地更新 Session（防止并发写冲突）
const luaUpdateSession = `
local key = KEYS[1]
local msg_json = ARGV[1]
local max_history = tonumber(ARGV[2])
local data = redis.call('GET', key)
if not data then return nil end
local session = cjson.decode(data)
local msg = cjson.decode(msg_json)
-- 追加消息
table.insert(session.messages, msg)
-- 如果消息过多，截断（Lua 中简化处理）
if #session.messages > max_history then
local keep = math.floor(max_history / 2)
local old = {}
for i = 1, #session.messages - keep do
table.insert(old, session.messages[i])
end
local recent = {}
for i = #session.messages - keep + 1, #session.messages do
table.insert(recent, session.messages[i])
end
session.messages = recent
session.summary = '历史对话已压缩'
end
session.updated_at = ARGV[3]
redis.call('SET', key, cjson.encode(session), 'EX', ARGV[4])
return 1
`
func (sm *SessionManager) AddMessageAtomic(ctx context.Context, sessionID string, msg Message) error {
return sm.rdb.Eval(ctx, luaUpdateSession,
[]string{sessionKey(sessionID)},
mustMarshal(msg), sm.maxHistory, time.Now().Format(time.RFC3339), int(sm.defaultTTL.Seconds()),
).Err()
}
func sessionKey(id string) string {
return fmt.Sprintf("agent:session:%s", id)
}
// 简化：实际项目应使用 UUID 库
func generateUUID() string {
return fmt.Sprintf("%d-%d", time.Now().UnixNano(), time.Now().Nanosecond())
}
func mustMarshal(v interface{}) string {
data, _ := json.Marshal(v)
return string(data)
}使用 MySQL 存储 Session（备选方案）
sql
-- Agent 会话持久化表
CREATE TABLE agent_sessions (
id VARCHAR(36) PRIMARY KEY, -- UUID
user_id VARCHAR(64) NOT NULL, -- 用户标识
messages JSON NOT NULL DEFAULT ('[]'), -- 对话历史 [{role, content, time}]
summary TEXT DEFAULT '', -- 历史对话摘要
metadata JSON DEFAULT ('{}'), -- 扩展元数据（模型、参数等）
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
INDEX idx_user_id (user_id),
INDEX idx_updated_at (updated_at) -- 用于清理过期会话
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- 定期清理过期会话的定时任务
-- DELETE FROM agent_sessions WHERE updated_at < NOW() - INTERVAL 1 DAY;完整 Agent 架构图
mermaid
graph TD
subgraph "前端"
UI["HTML+JS Web 界面<br/>对话窗口 + 工具展示"]
end
subgraph "API 网关"
GW["WebSocket / HTTP<br/>流式传输 SSE"]
end
subgraph "Agent 服务"
ROUTER["请求路由<br/>按 User/Session 分发"]
AGT["ReAct Agent 引擎<br/>思考→行动→观察循环"]
FCP["Function Calling<br/>JSON Schema → 工具执行"]
SM["Session Manager<br/>Redis/MySQL 多轮对话"]
end
subgraph "LLM 后端"
LLM1["OpenAI API"]
LLM2["本地 llama.cpp<br/>Metal / CUDA"]
LLM3["vLLM / TGI"]
end
subgraph "工具集"
T1["天气 API"]
T2["SQL 查询"]
T3["代码执行"]
T4["文件操作"]
end
UI <--> GW
GW <--> ROUTER
ROUTER --> AGT
AGT --> FCP
AGT --> SM
AGT --> LLM1
AGT --> LLM2
AGT --> LLM3
FCP --> T1
FCP --> T2
FCP --> T3
FCP --> T4
style AGT fill:#9b59b6,color:#fff
style FCP fill:#2ecc71,color:#fff
style SM fill:#3498db,color:#fff核心设计原则：Agent 引擎（ReAct 循环）负责调度推理和工具调用，Session Manager 负责持久化上下文，Function Calling 负责将自然语言指令转化为精确的 API 调用。三者解耦，各自演进。
LangGraph — 图结构的 Agent 编排
为什么需要 LangGraph？
ReAct 是线性循环，但真实的 Agent 任务往往需要分支、并行、条件跳转。LangGraph 将 Agent 的执行流程建模为有向图：
mermaid
graph TD
START["开始"] --> CLASSIFY["分类器<br/>判断任务类型"]
CLASSIFY -->|"代码问题"| CODE["代码 Agent<br/>搜索代码库+生成"]
CLASSIFY -->|"文档问题"| DOC["文档 Agent<br/>RAG 检索+总结"]
CLASSIFY -->|"数据问题"| DATA["数据 Agent<br/>SQL 查询+可视化"]
CODE --> REVIEW["审查节点<br/>检查代码质量"]
DOC --> FORMAT["格式化节点<br/>Markdown 输出"]
DATA --> FORMAT
REVIEW -->|"通过"| END["返回结果"]
REVIEW -->|"不通过"| CODE
FORMAT --> END
style CLASSIFY fill:#f39c12,color:#fff
style REVIEW fill:#e74c3c,color:#fffLangGraph 核心概念
| 概念 | 说明 | 类比 |
|---|---|---|
| State | 图的全局状态（贯穿所有节点） | 全局变量 |
| Node | 图中的一个处理节点（函数） | 函数 |
| Edge | 节点之间的连接 | 函数调用 |
| Conditional Edge | 根据状态决定下一个节点 | if-else |
| Checkpoint | 状态快照（支持回溯和恢复） | 存档点 |
Python 实现示例
python
"""
LangGraph 风格的 Agent 图实现
核心思想：将 Agent 的执行流程建模为状态机图
"""
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional, TypedDict
from enum import Enum
import asyncio
class NodeStatus(Enum):
"""节点执行状态"""
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
class AgentState(TypedDict, total=False):
"""Agent 全局状态（贯穿所有节点）"""
messages: List[dict] # 对话历史
task_type: str # 任务分类
intermediate_results: List[str] # 中间结果
final_answer: str # 最终答案
error: Optional[str] # 错误信息
retry_count: int # 重试次数
tools_used: List[str] # 已使用的工具
class StateGraph:
"""
状态图 — LangGraph 的核心抽象
用法：
1. 定义节点（处理函数）
2. 定义边（节点间的连接）
3. 编译并运行
"""
def __init__(self, state_class: type):
self.state_class = state_class
self.nodes: Dict[str, Callable] = {}
self.edges: Dict[str, List[str]] = {}
self.conditional_edges: Dict[str, Callable] = {}
self.entry_point: str = ""
def add_node(self, name: str, func: Callable):
"""添加节点"""
self.nodes[name] = func
if not self.entry_point:
self.entry_point = name
def add_edge(self, from_node: str, to_node: str):
"""添加固定边"""
if from_node not in self.edges:
self.edges[from_node] = []
self.edges[from_node].append(to_node)
def add_conditional_edge(self, from_node: str, condition_fn: Callable):
"""
添加条件边
condition_fn(state) -> str: 返回下一个节点的名称
"""
self.conditional_edges[from_node] = condition_fn
def set_entry_point(self, name: str):
"""设置入口节点"""
self.entry_point = name
async def run(self, initial_state: dict) -> dict:
"""执行图"""
state = dict(initial_state)
current_node = self.entry_point
max_steps = 20 # 防止无限循环
for step in range(max_steps):
if current_node == "END" or current_node is None:
break
# 执行当前节点
node_fn = self.nodes.get(current_node)
if node_fn is None:
raise ValueError(f"节点 '{current_node}' 不存在")
# 节点函数接收 state，返回更新后的 state
state = await node_fn(state) if asyncio.iscoroutinefunction(node_fn) else node_fn(state)
# 决定下一个节点
if current_node in self.conditional_edges:
# 条件边：根据状态决定
current_node = self.conditional_edges[current_node](state)
elif current_node in self.edges:
# 固定边：取第一个（简化）
current_node = self.edges[current_node][0]
else:
current_node = "END"
return state
# ========== 构建一个多功能 Agent 图 ==========
async def classify_task(state: dict) -> dict:
"""分类节点：判断用户任务类型"""
messages = state.get("messages", [])
last_msg = messages[-1]["content"] if messages else ""
# 简单的关键词分类（实际用 LLM）
if any(kw in last_msg for kw in ["代码", "函数", "bug", "实现"]):
state["task_type"] = "code"
elif any(kw in last_msg for kw in ["文档", "解释", "什么是"]):
state["task_type"] = "doc"
elif any(kw in last_msg for kw in ["数据", "统计", "查询", "SQL"]):
state["task_type"] = "data"
else:
state["task_type"] = "general"
return state
async def code_agent(state: dict) -> dict:
"""代码 Agent 节点"""
# 搜索代码库 → 生成代码 → 返回
state["intermediate_results"] = state.get("intermediate_results", [])
state["intermediate_results"].append("代码搜索完成")
state["tools_used"] = state.get("tools_used", []) + ["code_search", "code_gen"]
state["final_answer"] = "生成的代码..."
return state
async def doc_agent(state: dict) -> dict:
"""文档 Agent 节点"""
state["intermediate_results"] = state.get("intermediate_results", [])
state["intermediate_results"].append("文档检索完成")
state["tools_used"] = state.get("tools_used", []) + ["rag_search"]
state["final_answer"] = "文档回答..."
return state
async def data_agent(state: dict) -> dict:
"""数据 Agent 节点"""
state["intermediate_results"] = state.get("intermediate_results", [])
state["intermediate_results"].append("数据查询完成")
state["tools_used"] = state.get("tools_used", []) + ["sql_query"]
state["final_answer"] = "数据分析结果..."
return state
async def review_node(state: dict) -> dict:
"""审查节点：检查结果质量"""
# 简化：实际会用 LLM 评估
state["retry_count"] = state.get("retry_count", 0)
# 假设第一次通过
return state
def route_by_task_type(state: dict) -> str:
"""条件路由：根据任务类型选择 Agent"""
task_type = state.get("task_type", "general")
return {
"code": "code_agent",
"doc": "doc_agent",
"data": "data_agent",
"general": "doc_agent",
}.get(task_type, "doc_agent")
def route_after_review(state: dict) -> str:
"""审查后路由：通过则结束，否则重试"""
if state.get("retry_count", 0) >= 2:
return "END" # 重试次数用完
# 简化：总是通过
return "END"
# 构建图
graph = StateGraph(AgentState)
graph.add_node("classify", classify_task)
graph.add_node("code_agent", code_agent)
graph.add_node("doc_agent", doc_agent)
graph.add_node("data_agent", data_agent)
graph.add_node("review", review_node)
graph.set_entry_point("classify")
graph.add_conditional_edge("classify", route_by_task_type)
graph.add_edge("code_agent", "review")
graph.add_edge("doc_agent", "review")
graph.add_edge("data_agent", "review")
graph.add_conditional_edge("review", route_after_review)
# 运行
# result = asyncio.run(graph.run({
# "messages": [{"role": "user", "content": "帮我写一个排序函数"}],
# }))Agent 编排进阶 — 生产级工作流模式
LangGraph 提供了图编排的基础，但生产环境还需要更多编排模式：
Human-in-the-Loop — 人工审批节点
python
"""
Human-in-the-Loop (HITL) — 关键决策需要人工审批
场景：
- 删除生产数据前需要人工确认
- 发布代码前需要 Code Review
- 成本超过预算时暂停执行
"""
class HITLNode:
"""
人工审批节点
流程：
LLM 执行 → 到达审批节点 → 暂停 → 发通知 → 等待人工 → 继续/拒绝
"""
def __init__(self, notifier=None):
self.notifier = notifier # 通知渠道（企业微信/邮件/Slack）
self.pending_approvals: Dict[str, dict] = {}
async def request_approval(self, task_id: str, context: dict,
timeout: int = 3600) -> bool:
"""
请求人工审批
Args:
task_id: 任务 ID
context: 审批上下文（做了什么、为什么需要审批）
timeout: 超时时间（秒），超时自动拒绝
Returns:
approved: 是否通过
"""
# 1. 保存审批请求
approval_id = f"approval_{task_id}_{int(time.time())}"
self.pending_approvals[approval_id] = {
"task_id": task_id,
"context": context,
"status": "pending",
"created_at": time.time(),
"timeout": timeout,
}
# 2. 发送通知
await self.notifier.send({
"title": f"Agent 任务需要审批 — {task_id}",
"summary": context.get("summary", "Agent 请求审批"),
"details": json.dumps(context, ensure_ascii=False, indent=2),
"actions": [
{"label": "批准", "value": "approve", "style": "primary"},
{"label": "拒绝", "value": "reject", "style": "danger"},
],
"approval_id": approval_id,
})
# 3. 等待审批结果（轮询或 WebSocket）
deadline = time.time() + timeout
while time.time() < deadline:
approval = self.pending_approvals.get(approval_id)
if approval and approval["status"] != "pending":
return approval["status"] == "approved"
await asyncio.sleep(2)
# 超时
self.pending_approvals[approval_id]["status"] = "timeout"
return False
def handle_webhook(self, approval_id: str, action: str):
"""处理审批回调"""
if approval_id in self.pending_approvals:
self.pending_approvals[approval_id]["status"] = (
"approved" if action == "approve" else "rejected"
)
# 在 Agent 图中添加审批节点
class AgentWithApproval:
"""带审批的 Agent"""
def __init__(self):
self.hitl = HITLNode()
self.graph = StateGraph(AgentState)
async def approval_node(self, state: dict) -> dict:
"""审批节点：检查是否需要人工审批"""
task_type = state.get("task_type", "")
action = state.get("planned_action", "")
# 危险操作需要审批
dangerous_actions = ["删除", "发布", "修改生产配置", "执行 SQL"]
needs_approval = any(a in action for a in dangerous_actions)
if needs_approval:
approved = await self.hitl.request_approval(
task_id=state.get("task_id", ""),
context={
"summary": f"Agent 计划执行: {action}",
"risk": "此操作可能影响生产环境",
"alternatives": "可以拒绝并手动执行",
},
)
state["approved"] = approved
if not approved:
state["error"] = "操作被人工拒绝"
return state
# Graph 示例
# graph.add_node("approval_check", approval_node)
# graph.add_conditional_edge("approval_check", lambda s: "execute" if s.get("approved") else "END")工作流编排模式对比
| 模式 | 框架 | 特点 | 适用场景 | 复杂度 |
|---|---|---|---|---|
| DAG (有向无环图) | LangGraph / Airflow | 确定性的任务依赖 | 数据流水线、CI/CD | ⭐⭐ |
| 状态机 | 自研 / XState | 有限状态 + 事件驱动 | 审批流程、订单状态 | ⭐⭐ |
| 动态规划 | LangGraph + LLM | LLM 动态决定下一步 | 复杂 Agent 任务 | ⭐⭐⭐ |
| 工作流引擎 | Temporal / Prefect | 持久化、重试、超时 | 长时间运行的任务 | ⭐⭐⭐ |
| 发布-订阅 | 消息队列 + Agent | 事件驱动、松耦合 | 微服务、异步处理 | ⭐⭐⭐ |
并行 Agent 编排
python
"""
并行 Agent 编排 — 多个 Agent 同时工作，结果合并
场景：
- 同时搜索多个数据源，合并结果
- 同时生成多个候选方案，投票选最佳
- 代码 + 文档同时生成
"""
import asyncio
from typing import List
class ParallelOrchestrator:
"""并行编排器"""
async def run_parallel(self, tasks: List[dict], merge_strategy: str = "concat") -> dict:
"""
并行执行多个 Agent 任务
Args:
tasks: 任务列表 [{"agent": agent_fn, "input": ...}]
merge_strategy: 合并策略
- concat: 拼接结果
- vote: 投票选最佳
- first: 返回第一个完成的结果（race）
"""
# 并发启动所有 Agent
async def run_task(task):
try:
agent_fn = task["agent"]
result = await agent_fn(task["input"])
return {"success": True, "result": result, "task": task}
except Exception as e:
return {"success": False, "error": str(e), "task": task}
coroutines = [run_task(t) for t in tasks]
if merge_strategy == "first":
# Race：返回第一个完成的结果
for coro in asyncio.as_completed(coroutines):
result = await coro
if result["success"]:
# 取消其他任务
return {"winner": result, "strategy": "first"}
# 等待全部完成
results = await asyncio.gather(*coroutines, return_exceptions=True)
successful = [r for r in results if not isinstance(r, Exception) and r["success"]]
if merge_strategy == "concat":
# 拼接所有结果
combined = "\n\n---\n\n".join(
r["result"] for r in successful
)
return {"combined": combined, "results": successful, "strategy": "concat"}
elif merge_strategy == "vote":
# 投票选最佳（需要 LLM-as-Judge）
return await self._vote_best(successful)
async def _vote_best(self, results: List[dict]) -> dict:
"""LLM 投票选出最佳结果"""
# 用 Judge LLM 评估每个结果，选最优
candidates = "\n---\n".join(
f"候选 {i}: {r['result'][:200]}"
for i, r in enumerate(results)
)
# judge_prompt = f"评估以下候选方案，选择最佳的一个:\n{candidates}"
# winner_idx = await llm.judge(judge_prompt)
return {"results": results, "winner_idx": 0, "strategy": "vote"}
# ========== 混合编排：串行 + 并行 ==========
async def hybrid_orchestrate(user_request: str) -> dict:
"""
混合编排示例：串行步骤中包含并行子任务
1. [串行] 分析需求 → 拆解为子任务
2. [并行] 代码生成 + 文档生成 + 测试生成
3. [串行] 整合结果 → 质量审查 → 返回
"""
orchestrator = ParallelOrchestrator()
# Step 1: 分析需求（串行）
analysis = await analyze_requirement(user_request)
# Step 2: 并行执行
parallel_result = await orchestrator.run_parallel([
{"agent": generate_code, "input": analysis},
{"agent": generate_docs, "input": analysis},
{"agent": generate_tests, "input": analysis},
], merge_strategy="concat")
# Step 3: 审查结果（串行）
final = await review_and_merge(parallel_result)
return finalAgent 超时与重试策略
python
"""
Agent 超时与重试 — 生产环境必备
"""
class AgentExecutor:
"""带超时和重试的 Agent 执行器"""
def __init__(self, max_retries: int = 3, timeout: int = 60):
self.max_retries = max_retries
self.timeout = timeout
async def execute_with_retry(self, agent_fn, input_data: dict,
retryable_errors: tuple = (TimeoutError, ConnectionError)):
"""带重试的执行"""
last_error = None
for attempt in range(self.max_retries):
try:
# 带超时的执行
result = await asyncio.wait_for(
agent_fn(input_data),
timeout=self.timeout,
)
return {"success": True, "result": result, "attempts": attempt + 1}
except retryable_errors as e:
last_error = e
wait_time = min(2 ** attempt, 30) # 指数退避，最多 30s
print(f"重试 {attempt + 1}/{self.max_retries}，" +
f"等待 {wait_time}s: {type(e).__name__}")
await asyncio.sleep(wait_time)
except Exception as e:
# 不可重试的错误直接抛出
return {"success": False, "error": str(e), "attempts": attempt + 1}
return {
"success": False,
"error": f"重试 {self.max_retries} 次后仍失败: {last_error}",
"attempts": self.max_retries,
}
# 使用
executor = AgentExecutor(max_retries=3, timeout=60)
# result = await executor.execute_with_retry(my_agent, {"query": "..."})Multi-Agent 系统 — 多个 Agent 协作
为什么需要多 Agent？
单个 Agent 的能力有限。复杂任务需要多个专业 Agent 协作：
mermaid
graph TD
subgraph "Multi-Agent 架构"
USER["用户请求"] --> ORCHESTRATOR["协调者 Agent<br/>任务分解 + 调度"]
ORCHESTRATOR --> CODER["编码 Agent<br/>写代码"]
ORCHESTRATOR --> REVIEWER["审查 Agent<br/>代码审查"]
ORCHESTRATOR --> TESTER["测试 Agent<br/>写测试"]
ORCHESTRATOR --> RESEARCHER["研究 Agent<br/>搜索资料"]
CODER --> ORCHESTRATOR
REVIEWER --> ORCHESTRATOR
TESTER --> ORCHESTRATOR
RESEARCHER --> ORCHESTRATOR
ORCHESTRATOR --> RESULT["最终结果"]
end
style ORCHESTRATOR fill:#9b59b6,color:#fff
style CODER fill:#3498db,color:#fff
style REVIEWER fill:#e74c3c,color:#fff
style TESTER fill:#2ecc71,color:#fffMulti-Agent 通信模式
| 模式 | 说明 | 适用场景 |
|---|---|---|
| 中心化 | 一个协调者分配任务 | 任务明确、流程固定 |
| 去中心化 | Agent 之间直接通信 | 需要灵活协作 |
| 层级式 | 多层管理者 + 执行者 | 大型复杂任务 |
| 辩论式 | 多个 Agent 讨论达成共识 | 需要多角度分析 |
python
"""
Multi-Agent 协作框架
"""
from dataclasses import dataclass
from typing import Dict, List, Optional
from abc import ABC, abstractmethod
@dataclass
class AgentMessage:
"""Agent 间的消息"""
sender: str # 发送者 Agent 名称
receiver: str # 接收者 Agent 名称（"all" 表示广播）
content: str # 消息内容
msg_type: str # 消息类型: task/result/feedback/question
class BaseAgent(ABC):
"""Agent 基类"""
def __init__(self, name: str, role: str, llm_client=None):
self.name = name
self.role = role
self.llm = llm_client
self.inbox: List[AgentMessage] = []
@abstractmethod
async def process(self, task: str) -> str:
"""处理任务"""
...
async def receive(self, message: AgentMessage):
"""接收消息"""
self.inbox.append(message)
def send(self, receiver: str, content: str, msg_type: str = "result") -> AgentMessage:
"""发送消息"""
return AgentMessage(
sender=self.name,
receiver=receiver,
content=content,
msg_type=msg_type,
)
class OrchestratorAgent(BaseAgent):
"""协调者 Agent — 负责任务分解和调度"""
def __init__(self, llm_client=None):
super().__init__("orchestrator", "任务协调者", llm_client)
self.workers: Dict[str, BaseAgent] = {}
def register_worker(self, agent: BaseAgent):
"""注册工作 Agent"""
self.workers[agent.name] = agent
async def process(self, task: str) -> str:
"""分解任务并协调执行"""
# 1. 任务分解
subtasks = await self._decompose_task(task)
# 2. 分配给对应的 Agent
results = {}
for subtask in subtasks:
agent_name = subtask["agent"]
if agent_name in self.workers:
result = await self.workers[agent_name].process(subtask["task"])
results[agent_name] = result
# 3. 汇总结果
return await self._synthesize_results(task, results)
async def _decompose_task(self, task: str) -> List[dict]:
"""使用 LLM 分解任务"""
# 实际实现会调用 LLM
return [
{"agent": "coder", "task": f"实现: {task}"},
{"agent": "tester", "task": f"为以下功能写测试: {task}"},
]
async def _synthesize_results(self, task: str, results: Dict[str, str]) -> str:
"""汇总所有 Agent 的结果"""
return "\n".join(f"[{k}]: {v}" for k, v in results.items())
class CoderAgent(BaseAgent):
"""编码 Agent"""
def __init__(self, llm_client=None):
super().__init__("coder", "资深开发工程师", llm_client)
async def process(self, task: str) -> str:
# 实际会调用 LLM 生成代码
return f"代码实现: {task}"
class ReviewerAgent(BaseAgent):
"""审查 Agent"""
def __init__(self, llm_client=None):
super().__init__("reviewer", "代码审查专家", llm_client)
async def process(self, task: str) -> str:
return f"审查意见: {task}"
class TesterAgent(BaseAgent):
"""测试 Agent"""
def __init__(self, llm_client=None):
super().__init__("tester", "测试工程师", llm_client)
async def process(self, task: str) -> str:
return f"测试用例: {task}"
# ========== 辩论式 Multi-Agent ==========
class DebateSystem:
"""
辩论式 Multi-Agent
多个 Agent 对同一问题给出不同观点，通过辩论达成共识。
适用于需要多角度分析的场景（如架构设计决策）。
"""
def __init__(self, agents: List[BaseAgent], moderator: BaseAgent,
max_rounds: int = 3):
self.agents = agents
self.moderator = moderator
self.max_rounds = max_rounds
async def debate(self, topic: str) -> str:
"""进行辩论"""
history = []
for round_num in range(self.max_rounds):
round_opinions = []
# 每个 Agent 发表观点
for agent in self.agents:
context = f"话题: {topic}\n历史讨论:\n" + "\n".join(history)
opinion = await agent.process(context)
round_opinions.append(f"[{agent.name}]: {opinion}")
history.append(f"第{round_num+1}轮 [{agent.name}]: {opinion}")
# 主持人判断是否达成共识
consensus = await self.moderator.process(
f"判断以下讨论是否达成共识:\n" + "\n".join(round_opinions)
)
if "共识" in consensus or "一致" in consensus:
return consensus
# 未达成共识，由主持人总结
return await self.moderator.process(
f"讨论未达成共识，请总结各方观点并给出建议:\n" + "\n".join(history)
)MCP (Model Context Protocol) — 工具标准化协议
MCP 在 Agent 中的角色
MCP 解决了 Agent 工具调用的标准化问题。不同的 AI 客户端（Cursor、Claude Desktop、自定义 Agent）可以通过统一的 MCP 协议连接同一套工具：
mermaid
graph TD
subgraph "AI 客户端"
C1["Cursor"]
C2["Claude Desktop"]
C3["自定义 Agent"]
end
subgraph "MCP 协议层"
PROTO["JSON-RPC over stdio/SSE<br/>标准化的工具发现+调用"]
end
subgraph "MCP Servers (工具)"
S1["Git Server<br/>代码版本管理"]
S2["Database Server<br/>数据库查询"]
S3["Wiki Server<br/>知识库检索"]
S4["CI/CD Server<br/>构建部署"]
end
C1 <--> PROTO
C2 <--> PROTO
C3 <--> PROTO
PROTO <--> S1
PROTO <--> S2
PROTO <--> S3
PROTO <--> S4
style PROTO fill:#f39c12,color:#fffMCP 协议核心
| 能力 | 说明 | 示例 |
|---|---|---|
| Tools | 可调用的工具函数 | 搜索代码、执行 SQL、发送消息 |
| Resources | 可读取的数据资源 | 文件内容、数据库表结构、API 文档 |
| Prompts | 预定义的提示模板 | 代码审查模板、日志分析模板 |
| Sampling | 请求 LLM 生成内容 | Server 主动请求 AI 分析 |
实现 MCP Server（Go 版本）
go
package main
import (
"bufio"
"encoding/json"
"fmt"
"os"
)
// MCP JSON-RPC 消息格式
type JSONRPCRequest struct {
JSONRPC string `json:"jsonrpc"`
ID interface{} `json:"id"`
Method string `json:"method"`
Params json.RawMessage `json:"params,omitempty"`
}
type JSONRPCResponse struct {
JSONRPC string `json:"jsonrpc"`
ID interface{} `json:"id"`
Result interface{} `json:"result,omitempty"`
Error *RPCError `json:"error,omitempty"`
}
type RPCError struct {
Code int `json:"code"`
Message string `json:"message"`
}
// Tool 定义
type ToolDefinition struct {
Name string `json:"name"`
Description string `json:"description"`
InputSchema interface{} `json:"inputSchema"`
}
// MCP Server
type MCPServer struct {
tools map[string]ToolHandler
}
type ToolHandler func(args json.RawMessage) (string, error)
func NewMCPServer() *MCPServer {
return &MCPServer{
tools: make(map[string]ToolHandler),
}
}
func (s *MCPServer) RegisterTool(name, description string, handler ToolHandler) {
s.tools[name] = handler
}
func (s *MCPServer) Run() {
scanner := bufio.NewScanner(os.Stdin)
for scanner.Scan() {
line := scanner.Text()
var req JSONRPCRequest
if err := json.Unmarshal([]byte(line), &req); err != nil {
continue
}
resp := s.handleRequest(req)
data, _ := json.Marshal(resp)
fmt.Println(string(data))
}
}
func (s *MCPServer) handleRequest(req JSONRPCRequest) JSONRPCResponse {
switch req.Method {
case "initialize":
return JSONRPCResponse{
JSONRPC: "2.0",
ID: req.ID,
Result: map[string]interface{}{
"protocolVersion": "2024-11-05",
"capabilities": map[string]interface{}{
"tools": map[string]bool{"listChanged": true},
},
"serverInfo": map[string]string{
"name": "wiki-agent-tools",
"version": "1.0.0",
},
},
}
case "tools/list":
tools := make([]ToolDefinition, 0)
for name := range s.tools {
tools = append(tools, ToolDefinition{
Name: name,
Description: fmt.Sprintf("Tool: %s", name),
})
}
return JSONRPCResponse{JSONRPC: "2.0", ID: req.ID, Result: map[string]interface{}{"tools": tools}}
case "tools/call":
var params struct {
Name string `json:"name"`
Arguments json.RawMessage `json:"arguments"`
}
json.Unmarshal(req.Params, ¶ms)
handler, ok := s.tools[params.Name]
if !ok {
return JSONRPCResponse{JSONRPC: "2.0", ID: req.ID, Error: &RPCError{Code: -1, Message: "tool not found"}}
}
result, err := handler(params.Arguments)
if err != nil {
return JSONRPCResponse{JSONRPC: "2.0", ID: req.ID, Error: &RPCError{Code: -1, Message: err.Error()}}
}
return JSONRPCResponse{
JSONRPC: "2.0",
ID: req.ID,
Result: map[string]interface{}{
"content": []map[string]string{{"type": "text", "text": result}},
},
}
default:
return JSONRPCResponse{JSONRPC: "2.0", ID: req.ID, Error: &RPCError{Code: -32601, Message: "method not found"}}
}
}
// 使用示例
func main() {
server := NewMCPServer()
// 注册工具
server.RegisterTool("search_wiki", "搜索 Wiki 知识库", func(args json.RawMessage) (string, error) {
var params struct {
Query string `json:"query"`
}
json.Unmarshal(args, ¶ms)
// 实际实现：调用向量数据库搜索
return fmt.Sprintf("搜索结果: %s", params.Query), nil
})
server.RegisterTool("get_file", "读取项目文件", func(args json.RawMessage) (string, error) {
var params struct {
Path string `json:"path"`
}
json.Unmarshal(args, ¶ms)
data, err := os.ReadFile(params.Path)
if err != nil {
return "", err
}
return string(data), nil
})
server.Run()
}Agent 可观测性与调试
Agent 执行追踪
mermaid
graph TD
subgraph "Agent 可观测性"
TRACE["执行追踪<br/>每步的输入/输出/耗时"] --> LOG["结构化日志<br/>JSON 格式，可搜索"]
LOG --> METRIC["指标监控<br/>成功率/延迟/Token 消耗"]
METRIC --> ALERT["告警<br/>异常检测+通知"]
end
style TRACE fill:#3498db,color:#fff
style METRIC fill:#2ecc71,color:#fffpython
"""
Agent 执行追踪系统
"""
import time
import uuid
from dataclasses import dataclass, field
from typing import List, Optional
from contextlib import contextmanager
@dataclass
class TraceSpan:
"""追踪跨度 — 记录一次操作"""
span_id: str = ""
parent_id: str = ""
name: str = "" # 操作名称
start_time: float = 0
end_time: float = 0
duration_ms: float = 0
status: str = "ok" # ok / error
attributes: dict = field(default_factory=dict)
events: List[dict] = field(default_factory=list)
def __post_init__(self):
if not self.span_id:
self.span_id = str(uuid.uuid4())[:8]
@dataclass
class AgentTrace:
"""完整的 Agent 执行追踪"""
trace_id: str = ""
spans: List[TraceSpan] = field(default_factory=list)
total_tokens: int = 0
total_cost: float = 0.0
total_duration_ms: float = 0
def __post_init__(self):
if not self.trace_id:
self.trace_id = str(uuid.uuid4())[:12]
class AgentTracer:
"""Agent 追踪器"""
def __init__(self):
self.current_trace: Optional[AgentTrace] = None
self._span_stack: List[TraceSpan] = []
@contextmanager
def trace(self, name: str):
"""开始一个新的追踪"""
self.current_trace = AgentTrace()
start = time.perf_counter()
try:
yield self.current_trace
finally:
self.current_trace.total_duration_ms = (time.perf_counter() - start) * 1000
@contextmanager
def span(self, name: str, **attributes):
"""记录一个操作跨度"""
span = TraceSpan(
name=name,
start_time=time.perf_counter(),
parent_id=self._span_stack[-1].span_id if self._span_stack else "",
attributes=attributes,
)
self._span_stack.append(span)
try:
yield span
except Exception as e:
span.status = "error"
span.events.append({"error": str(e)})
raise
finally:
span.end_time = time.perf_counter()
span.duration_ms = (span.end_time - span.start_time) * 1000
self._span_stack.pop()
if self.current_trace:
self.current_trace.spans.append(span)
def log_llm_call(self, span: TraceSpan, prompt_tokens: int,
completion_tokens: int, model: str):
"""记录 LLM 调用信息"""
span.attributes.update({
"llm.model": model,
"llm.prompt_tokens": prompt_tokens,
"llm.completion_tokens": completion_tokens,
"llm.total_tokens": prompt_tokens + completion_tokens,
})
if self.current_trace:
self.current_trace.total_tokens += prompt_tokens + completion_tokens
def export(self) -> dict:
"""导出追踪数据（可发送到 Jaeger/Zipkin）"""
if not self.current_trace:
return {}
return {
"trace_id": self.current_trace.trace_id,
"total_duration_ms": self.current_trace.total_duration_ms,
"total_tokens": self.current_trace.total_tokens,
"spans": [
{
"span_id": s.span_id,
"parent_id": s.parent_id,
"name": s.name,
"duration_ms": s.duration_ms,
"status": s.status,
"attributes": s.attributes,
}
for s in self.current_trace.spans
],
}
# 使用示例
# tracer = AgentTracer()
# with tracer.trace("user_query"):
# with tracer.span("classify_task") as span:
# # 分类任务...
# pass
# with tracer.span("llm_call", model="gpt-5.6") as span:
# # 调用 LLM...
# tracer.log_llm_call(span, 500, 200, "gpt-5.6")
# with tracer.span("tool_call", tool="search") as span:
# # 调用工具...
# pass
# print(json.dumps(tracer.export(), indent=2))Agent 技术选型对比
| 框架 | 语言 | 核心特点 | 适用场景 | 学习曲线 |
|---|---|---|---|---|
| LangChain | Python | 组件丰富、生态最大 | 快速原型 | 中 |
| LangGraph | Python | 图结构、状态管理 | 复杂工作流 | 高 |
| CrewAI | Python | Multi-Agent、角色扮演 | 团队协作任务 | 低 |
| AutoGen | Python | 多 Agent 对话 | 研究/实验 | 中 |
| Semantic Kernel | C#/Python | 微软生态、企业级 | .NET 项目 | 中 |
| 自研 (Go/Python) | Go/Python | 完全可控、无依赖 | 生产环境 | 高 |
建议：原型阶段用 LangGraph 快速验证，生产环境用自研框架（如本文的 Go/Python 实现）确保可控性和性能。
Agent 长期记忆 — 持久化的知识与经验
短期记忆（Session 上下文压缩）解决的是单次对话的上下文窗口问题，但当 Agent 需要跨会话记住用户偏好、积累领域知识、从历史任务中学习时，就需要长期记忆系统。
记忆层级架构
mermaid
graph TD
subgraph "Agent 记忆层级"
WM["工作记忆<br/>当前对话上下文<br/>存活: 单次会话"] --> STM["短期记忆<br/>对话摘要 + 关键事实<br/>存活: 数小时到数天"]
STM --> LTM["长期记忆<br/>用户偏好 + 领域知识<br/>存活: 永久"]
LTM --> EM["情景记忆<br/>历史任务经验<br/>存活: 永久"]
end
WM -->|"超出窗口"| COMPRESS["压缩归档"]
COMPRESS --> STM
STM -->|"模式识别"| EXTRACT["知识沉淀"]
EXTRACT --> LTM
LTM -->|"相似任务"| RECALL["经验召回"]
RECALL --> WM
style WM fill:#3498db,color:#fff
style LTM fill:#e74c3c,color:#fff
style EM fill:#9b59b6,color:#fff实现方案：向量记忆 + 结构化事实
python
"""
Agent 长期记忆系统
双层存储:
1. 向量记忆 — 语义检索历史对话片段
2. 结构化事实 — 精确存储用户偏好、实体关系
"""
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Any
from datetime import datetime
import json
@dataclass
class Memory:
"""记忆单元"""
id: str
content: str # 记忆内容
memory_type: str # preference / fact / experience / conversation
importance: float = 0.5 # 重要性评分 0-1
created_at: float = 0.0
last_accessed: float = 0.0
access_count: int = 0
metadata: Dict[str, Any] = field(default_factory=dict)
def __post_init__(self):
if not self.created_at:
self.created_at = datetime.now().timestamp()
class LongTermMemory:
"""长期记忆系统"""
def __init__(self, vector_store, llm_client, kv_store=None):
"""
Args:
vector_store: 向量数据库（Milvus/Qdrant/Chroma），用于语义检索
llm_client: LLM 客户端，用于记忆提取和重要性评分
kv_store: 键值存储（Redis/SQLite），用于结构化事实
"""
self.vector_store = vector_store
self.llm = llm_client
self.kv_store = kv_store or {}
# ===== 记忆写入 =====
async def add_memory(
self,
content: str,
memory_type: str = "conversation",
metadata: Dict = None,
) -> Memory:
"""添加一条记忆"""
# 1. 评估重要性
importance = await self._rate_importance(content, memory_type)
# 2. 创建记忆单元
memory = Memory(
id=f"mem_{datetime.now().strftime('%Y%m%d%H%M%S')}_{hash(content) % 10000}",
content=content,
memory_type=memory_type,
importance=importance,
metadata=metadata or {},
)
# 3. 写入向量存储（语义检索）
await self.vector_store.insert(
id=memory.id,
text=content,
metadata={
"memory_type": memory_type,
"importance": importance,
**memory.metadata,
},
)
# 4. 如果是结构化事实，额外写入 KV 存储
if memory_type == "fact":
structured = await self._structure_fact(content)
if structured:
key = structured["key"]
self.kv_store[key] = {
"value": structured["value"],
"source_memory_id": memory.id,
"updated_at": datetime.now().isoformat(),
}
return memory
async def add_from_conversation(self, messages: List[Dict]) -> List[Memory]:
"""从对话中提取并存储记忆"""
# 用 LLM 从对话中提取值得记住的信息
conversation_text = "\n".join([
f"[{m['role']}]: {m['content'][:300]}" for m in messages
])
prompt = f"""从以下对话中提取值得长期记住的信息。返回 JSON 数组，每条包含 content 和 type。
类型说明:
- preference: 用户偏好（"用户喜欢简洁的回答"）
- fact: 事实信息（"项目A的API地址是xxx"）
- experience: 经验教训（"上次用方案X导致性能问题"）
对话:
{conversation_text}
输出格式:
[{{"content": "...", "type": "preference|fact|experience"}}]
"""
response = await self.llm.chat(prompt, response_format="json")
items = json.loads(response)
memories = []
for item in items:
memory = await self.add_memory(
content=item["content"],
memory_type=item["type"],
)
memories.append(memory)
return memories
# ===== 记忆检索 =====
async def recall(
self,
query: str,
top_k: int = 5,
memory_type: str = None,
min_importance: float = 0.0,
) -> List[Memory]:
"""检索相关记忆"""
# 1. 向量语义检索
results = await self.vector_store.search(
query=query,
top_k=top_k * 2, # 多召回，后续过滤
)
# 2. 按条件和重要性过滤
filtered = []
for r in results:
meta = r.get("metadata", {})
if memory_type and meta.get("memory_type") != memory_type:
continue
if meta.get("importance", 0) < min_importance:
continue
filtered.append(r)
# 3. 更新访问时间
now = datetime.now().timestamp()
for r in filtered[:top_k]:
r_id = r.get("id")
# 在向量存储中更新 last_accessed（实际实现取决于向量数据库）
return filtered[:top_k]
async def recall_facts(self, entity: str) -> List[Dict]:
"""精确召回关于某个实体的结构化事实"""
facts = []
prefix = f"fact:{entity}:"
for key, value in self.kv_store.items():
if key.startswith(prefix):
facts.append({
"key": key.replace(prefix, ""),
"value": value["value"],
"updated_at": value["updated_at"],
})
return facts
# ===== 记忆维护 =====
async def consolidate(self):
"""记忆巩固：定期将短期记忆沉淀为长期知识"""
# 1. 获取最近 24h 内访问过但重要性低、访问次数高的记忆
recent = await self.vector_store.search(
query="", # 按时间过滤
top_k=50,
filter={"last_accessed": {"$gte": datetime.now().timestamp() - 86400}},
)
# 2. 用 LLM 识别可合并/升级的记忆
if recent:
content_list = "\n".join([
f"- [{r.get('metadata', {}).get('memory_type', '?')}] {r['text'][:100]}"
for r in recent
])
prompt = f"""分析以下最近访问的记忆，找出:
1. 可以合并的重复/相似记忆
2. 应该从短期升级为长期的记忆
3. 可以降低重要性的过时信息
记忆列表:
{content_list}
返回 JSON:
{{"merge": [["id1", "id2", "合并后内容"]], "upgrade": ["id3", "id4"], "downgrade": ["id5"]}}
"""
response = await self.llm.chat(prompt, response_format="json")
plan = json.loads(response)
# 3. 执行合并/升级/降级操作
await self._execute_consolidation_plan(plan)
async def forget(self, memory_id: str):
"""删除记忆"""
await self.vector_store.delete(memory_id)
# 清除关联的结构化事实（如果有）
def get_stats(self) -> Dict:
"""记忆系统统计"""
return {
"total_vector_memories": len(self.vector_store),
"total_structured_facts": len(self.kv_store),
}
# ===== 内部方法 =====
async def _rate_importance(self, content: str, memory_type: str) -> float:
"""LLM 评估记忆重要性"""
prompt = f"""评估以下信息的重要性（0-1之间的浮点数）:
类型: {memory_type}
内容: {content[:200]}
评分标准:
- 0.8-1.0: 关键信息（用户核心偏好、系统配置、安全规则）
- 0.5-0.7: 有价值信息（项目细节、经验教训）
- 0.2-0.4: 一般信息（日常对话、临时讨论）
- 0.0-0.1: 琐碎信息
只返回数字。"""
response = await self.llm.chat(prompt)
try:
return float(response.strip())
except ValueError:
return 0.5
async def _structure_fact(self, content: str) -> Optional[Dict]:
"""将自然语言事实转为结构化键值"""
prompt = f"""将以下事实转为结构化键值对，key 格式为 "fact:{实体}:{属性}":
事实: {content}
返回 JSON:
{{"key": "fact:xxx:yyy", "value": "具体值"}}
如果无法结构化，返回 null。"""
response = await self.llm.chat(prompt, response_format="json")
return json.loads(response)
async def _execute_consolidation_plan(self, plan: Dict):
"""执行记忆巩固计划"""
for items in plan.get("merge", []):
# 合并多条记忆为一条
merged_content = items[2] if len(items) > 2 else ""
if merged_content:
await self.add_memory(merged_content, "experience", importance=0.7)
for mid in items[:2]:
await self.forget(mid)
for mid in plan.get("upgrade", []):
# 升级为长期记忆（实际实现取决于向量数据库的元数据更新）
pass
for mid in plan.get("downgrade", []):
# 降级重要性
pass记忆检索集成到 Agent 循环
python
class MemoryAwareAgent:
"""集成长期记忆的 Agent"""
def __init__(self, llm_client, tools, memory: LongTermMemory):
self.llm = llm_client
self.tools = tools
self.memory = memory
self.conversation_history = [] # 短期工作记忆
async def think(self, user_message: str) -> str:
"""带有记忆增强的思考过程"""
# 1. 从长期记忆中检索相关信息
relevant_memories = await self.memory.recall(
query=user_message,
top_k=3,
)
# 2. 检索用户偏好
user_preferences = await self.memory.recall(
query=user_message,
memory_type="preference",
top_k=2,
)
# 3. 检索相关结构化事实
entities = await self._extract_entities(user_message)
structured_facts = {}
for entity in entities:
facts = await self.memory.recall_facts(entity)
if facts:
structured_facts[entity] = facts
# 4. 构建增强的上下文
memory_context = self._build_memory_context(
relevant_memories, user_preferences, structured_facts
)
# 5. 标准 Agent 推理
system_prompt = f"""你是一个有记忆的 AI Agent。参考以下记忆来回答用户问题。
## 用户偏好
{memory_context['preferences']}
## 相关知识
{memory_context['knowledge']}
## 相关经验
{memory_context['experiences']}"""
response = await self.llm.chat(
system=system_prompt,
messages=self.conversation_history + [{"role": "user", "content": user_message}],
)
# 6. 更新工作记忆
self.conversation_history.append({"role": "user", "content": user_message})
self.conversation_history.append({"role": "assistant", "content": response})
# 7. 异步：从对话中提取新记忆（不阻塞主流程）
asyncio.create_task(
self.memory.add_from_conversation(self.conversation_history[-4:])
)
return response
def _build_memory_context(
self, memories, preferences, facts
) -> Dict[str, str]:
"""构建记忆上下文"""
return {
"preferences": "\n".join([
f"- {m.content}" for m in preferences
]) or "无",
"knowledge": "\n".join([
f"- {m.content}" for m in memories if m.memory_type != "experience"
]) or "无",
"experiences": "\n".join([
f"- {m.content}" for m in memories if m.memory_type == "experience"
]) or "无",
}记忆系统的工程考量
| 考量点 | 方案 | 说明 |
|---|---|---|
| 检索延迟 | 异步预取 + 缓存 | 长期记忆检索不阻塞主推理流程 |
| 存储增长 | TTL + 重要性衰减 | 低重要性记忆自动过期，控制存储成本 |
| 隐私合规 | 按用户隔离 + 遗忘权 | 支持用户级记忆删除，符合 GDPR |
| 冷启动 | 默认偏好模板 | 新用户无记忆时使用通用默认值 |
| 多会话一致性 | 分布式 KV 存储 (Redis) | 结构化事实跨服务实例共享 |
核心原则：长期记忆不是"把所有对话都存下来"，而是"提取值得记住的，遗忘不值得的"。好的记忆系统应该像人类一样，重要的记得牢，琐碎的自动淡忘。
原生多模态 Agent — 视觉 + 听觉 + 行动的统一体
范式跃迁：从"文本为主"到"原生多模态"
GPT-5.6 和 Claude Fable 这一代模型的根本不同在于：它们不再把图片/音频当作"文本的附件"，而是原生理解多模态输入。Agent 的感知和行动能力因此大幅扩展：
mermaid
graph TD
subgraph "上一代 Agent (GPT-4o 时代)"
T1["文本输入"] --> AG1["Agent 推理"]
IMG1["图片输入<br/>(需要预处理)"] -.->|"转文本描述"| AG1
AUD1["音频输入<br/>(需要 STT)"] -.->|"转文字"| AG1
AG1 --> OUT1["文本输出<br/>+ 函数调用"]
end
subgraph "新一代 Agent (GPT-5.6 / Claude Fable)"
TXT["文本"] --> AG2["Agent 推理引擎<br/>统一多模态理解"]
IMG["图像/视频帧"] --> AG2
AUD["音频流"] --> AG2
CODE["代码/表格"] --> AG2
AG2 --> TOUT["文本 + 结构化输出"]
AG2 --> VOUT["图像生成/编辑"]
AG2 --> ACT["Computer Use<br/>鼠标/键盘操作"]
AG2 --> AOUT["语音合成输出"]
end
style AG2 fill:#e74c3c,color:#fff
style ACT fill:#9b59b6,color:#fff多模态 Agent 工程实现
python
"""
原生多模态 Agent — 统一处理文本、图像、音频、代码
"""
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Union
from enum import Enum
import base64
import json
class ContentType(Enum):
TEXT = "text"
IMAGE = "image"
AUDIO = "audio"
VIDEO_FRAME = "video_frame"
CODE = "code"
TABLE = "table"
@dataclass
class MultiModalContent:
"""多模态内容块"""
type: ContentType
data: Union[str, bytes] # 文本内容 或 base64 编码的媒体
mime_type: str = "" # image/png, audio/wav 等
metadata: Dict = field(default_factory=dict) # 分辨率、时长等
class MultiModalAgent:
"""原生多模态 Agent"""
def __init__(self, llm_client, tools: List = None):
self.llm = llm_client
self.tools = tools or []
def build_message(
self, user_input: Union[str, List[MultiModalContent]]
) -> dict:
"""构建多模态消息"""
if isinstance(user_input, str):
return {"role": "user", "content": user_input}
# 多模态消息：多个 content block
content_blocks = []
for block in user_input:
if block.type == ContentType.TEXT:
content_blocks.append({
"type": "text",
"text": block.data,
})
elif block.type == ContentType.IMAGE:
# 支持本地文件路径或 base64
if isinstance(block.data, str) and not block.data.startswith("data:"):
with open(block.data, "rb") as f:
img_data = base64.b64encode(f.read()).decode()
else:
img_data = block.data if isinstance(block.data, str) else \
base64.b64encode(block.data).decode()
content_blocks.append({
"type": "image",
"source": {
"type": "base64",
"media_type": block.mime_type or "image/png",
"data": img_data,
},
})
elif block.type == ContentType.AUDIO:
if isinstance(block.data, str):
with open(block.data, "rb") as f:
audio_data = base64.b64encode(f.read()).decode()
else:
audio_data = base64.b64encode(block.data).decode()
content_blocks.append({
"type": "audio",
"source": {
"type": "base64",
"media_type": block.mime_type or "audio/wav",
"data": audio_data,
},
})
elif block.type == ContentType.VIDEO_FRAME:
# 视频帧作为图片序列发送
content_blocks.append({
"type": "image",
"source": {
"type": "base64",
"media_type": "image/jpeg",
"data": block.data,
},
"metadata": {"frame_index": block.metadata.get("frame_index", 0)},
})
return {"role": "user", "content": content_blocks}
async def perceive_and_act(
self, input_contents: List[MultiModalContent], task: str
) -> Dict:
"""多模态感知 + 行动"""
# 1. 构建消息
message = self.build_message(input_contents)
# 2. 添加任务指令
if isinstance(message["content"], list):
message["content"].insert(0, {
"type": "text",
"text": f"[任务] {task}\n\n请分析以上多模态内容并执行任务。",
})
# 3. 调用 LLM（Native Tool Use）
response = await self.llm.chat(
messages=[message],
tools=self.tools,
tool_choice="auto",
)
return response
# ===== 常见多模态场景 =====
async def analyze_ui_screenshot(
self, screenshot_path: str, task: str = "描述这个界面并找出所有可交互元素"
) -> Dict:
"""分析 UI 截图"""
return await self.perceive_and_act(
[MultiModalContent(
type=ContentType.IMAGE,
data=screenshot_path,
mime_type="image/png",
)],
task,
)
async def debug_from_screenshot(
self, screenshot_path: str, error_description: str
) -> Dict:
"""从截图诊断错误"""
return await self.perceive_and_act(
[
MultiModalContent(
type=ContentType.IMAGE, data=screenshot_path, mime_type="image/png",
),
MultiModalContent(
type=ContentType.TEXT,
data=f"用户描述的错误: {error_description}",
),
],
"分析截图中的错误信息，给出修复方案",
)
async def review_design(
self, design_image: str, requirements: str
) -> Dict:
"""设计稿审查"""
return await self.perceive_and_act(
[
MultiModalContent(
type=ContentType.IMAGE, data=design_image, mime_type="image/png",
),
MultiModalContent(
type=ContentType.TEXT, data=f"需求文档: {requirements}",
),
],
"对比设计稿和需求文档，找出不一致之处并给出修改建议",
)
async def transcribe_and_analyze(
self, audio_path: str
) -> Dict:
"""音频转写+分析（原生，不需要 STT 预处理）"""
return await self.perceive_and_act(
[MultiModalContent(
type=ContentType.AUDIO,
data=audio_path,
mime_type="audio/mp3",
)],
"转写这段音频，然后提取关键信息和待办事项",
)
async def analyze_video(
self, video_path: str, analysis_task: str, fps: float = 1.0
) -> Dict:
"""视频分析 — 按帧率采样"""
import cv2
frames = []
cap = cv2.VideoCapture(video_path)
frame_interval = int(cap.get(cv2.CAP_PROP_FPS) / fps) if fps > 0 else 30
frame_idx = 0
while True:
ret, frame = cap.read()
if not ret:
break
if frame_idx % frame_interval == 0:
_, buffer = cv2.imencode('.jpg', frame)
frames.append(MultiModalContent(
type=ContentType.VIDEO_FRAME,
data=base64.b64encode(buffer).decode(),
mime_type="image/jpeg",
metadata={"frame_index": frame_idx, "timestamp": frame_idx / cap.get(cv2.CAP_PROP_FPS)},
))
frame_idx += 1
cap.release()
return await self.perceive_and_act(
frames + [MultiModalContent(type=ContentType.TEXT, data=analysis_task)],
f"分析视频内容（{len(frames)} 帧），完成: {analysis_task}",
)多模态 Agent 场景矩阵
| 场景 | 输入 | Agent 能力 | 典型应用 |
|---|---|---|---|
| UI 自动化测试 | 截图 + 操作描述 | 识别按钮/文本/布局 → 生成操作指令 | 视觉回归测试、无障碍检查 |
| 客服质检 | 聊天截图/录音 | 识别情绪、检测违规用语 | 客服质量分析 |
| 代码审查增强 | 代码 + UI 截图 | 对比代码逻辑与实际渲染 | PR Review 可视化 |
| 会议纪要 | 音频 + 共享屏幕 | 转写 + 提取决策 + 生成待办 | 智能会议助手 |
| 文档理解 | PDF/扫描件（含图表） | 同时理解文字和图表 | 合同审查、发票识别 |
| 视频内容审核 | 视频流 | 逐帧分析 + 时间轴标注 | 内容安全审核 |
Computer Use / Code Agent — 模型直接操作世界的时代
概念
2025-2026 年，Agent 的边界从"调用 API"扩展到"直接操作计算机"。Claude Fable 的 Computer Use 和 OpenAI 的 Operator 将屏幕截图→鼠标键盘操作作为 Agent 的原生能力：
mermaid
graph TD
subgraph "传统 Agent"
U1["用户指令"] --> LC1["LLM 推理"] --> TC1["Tool Call<br/>API/命令行"]
end
subgraph "Computer Use Agent"
U2["用户指令"] --> LC2["LLM 推理"] --> ACT2["操作计算机<br/>🖱️ 移动鼠标<br/>⌨️ 键盘输入<br/>🖥️ 截屏验证"]
ACT2 -->|"获取新截图"| OBS["观察环境变化"]
OBS --> LC2
end
style ACT2 fill:#e74c3c,color:#fff
style OBS fill:#f39c12,color:#fffComputer Use 实现
python
"""
Computer Use Agent — 模型控制鼠标键盘
"""
from dataclasses import dataclass
from typing import List, Dict, Optional, Tuple
import pyautogui
import base64
import io
import time
@dataclass
class ComputerAction:
"""计算机操作"""
action_type: str # click, type, scroll, screenshot, key_press, drag, wait
params: Dict = None
class ComputerUseAgent:
"""Computer Use Agent — 自动化操作桌面/浏览器"""
def __init__(self, llm_client):
self.llm = llm_client
pyautogui.FAILSAFE = True # 鼠标移到角落时停止
pyautogui.PAUSE = 0.5 # 每个操作后暂停
def get_screenshot(self) -> str:
"""截取当前屏幕，返回 base64"""
screenshot = pyautogui.screenshot()
buffer = io.BytesIO()
screenshot.save(buffer, format="PNG")
return base64.b64encode(buffer.getvalue()).decode()
def get_screen_size(self) -> Tuple[int, int]:
"""获取屏幕分辨率"""
return pyautogui.size()
def execute_action(self, action: ComputerAction) -> Dict:
"""执行单个操作"""
result = {"success": True, "action": action.action_type}
try:
if action.action_type == "click":
x, y = action.params["x"], action.params["y"]
# 坐标归一化：模型可能返回 0-1 的相对坐标
if x < 1 and y < 1:
w, h = self.get_screen_size()
x, y = int(x * w), int(y * h)
pyautogui.click(x, y)
result["position"] = (x, y)
elif action.action_type == "type":
text = action.params["text"]
pyautogui.write(text)
result["text"] = text
elif action.action_type == "key_press":
keys = action.params["keys"] # 如 "enter", "ctrl+c"
pyautogui.hotkey(*keys.split("+"))
result["keys"] = keys
elif action.action_type == "scroll":
amount = action.params.get("amount", 3)
pyautogui.scroll(amount)
result["amount"] = amount
elif action.action_type == "drag":
x1, y1 = action.params["x1"], action.params["y1"]
x2, y2 = action.params["x2"], action.params["y2"]
pyautogui.moveTo(x1, y1)
pyautogui.drag(x2 - x1, y2 - y1, duration=0.5)
elif action.action_type == "screenshot":
result["image"] = self.get_screenshot()
elif action.action_type == "wait":
duration = action.params.get("duration", 1)
time.sleep(duration)
elif action.action_type == "move":
x, y = action.params["x"], action.params["y"]
pyautogui.moveTo(x, y)
result["position"] = (x, y)
else:
result["success"] = False
result["error"] = f"Unknown action: {action.action_type}"
except Exception as e:
result["success"] = False
result["error"] = str(e)
return result
async def run_task(self, task: str, max_steps: int = 20) -> Dict:
"""执行一个完整的 Computer Use 任务"""
history = []
screenshot_before = self.get_screenshot()
for step in range(max_steps):
# 1. 发送截图 + 任务 + 历史给模型
message = {
"role": "user",
"content": [
{"type": "text", "text": f"[任务] {task}\n\n你正在操作一台 {self.get_screen_size()} 分辨率的计算机。\n请决定下一步操作。只输出 JSON:\n{{\"action\": \"click|type|scroll|key_press|done\", \"params\": {{...}}, \"reasoning\": \"为什么做这个操作\"}}"},
{"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": screenshot_before}},
],
}
if history:
message["content"].insert(0, {
"type": "text",
"text": f"历史操作: {json.dumps(history[-3:], ensure_ascii=False)}",
})
# 2. 获取模型决策
response = await self.llm.chat(
messages=[message],
response_format="json",
)
decision = json.loads(response)
# 3. 如果任务完成
if decision.get("action") == "done":
return {
"success": True,
"steps": len(history),
"history": history,
"conclusion": decision.get("params", {}).get("result", "Task completed"),
}
# 4. 执行操作
action = ComputerAction(
action_type=decision["action"],
params=decision.get("params", {}),
)
result = self.execute_action(action)
result["reasoning"] = decision.get("reasoning", "")
history.append(result)
# 5. 截取操作后的屏幕
time.sleep(0.5) # 等待界面更新
screenshot_before = self.get_screenshot()
return {
"success": False,
"steps": len(history),
"history": history,
"error": f"Reached max steps ({max_steps}) without completing task",
}
# ===== Code Agent — 模型直接写代码并执行 =====
class CodeAgent:
"""Code Agent — 在隔离环境中编写和执行代码"""
def __init__(self, llm_client, workspace: str = "/tmp/code_agent"):
self.llm = llm_client
self.workspace = workspace
import os
os.makedirs(workspace, exist_ok=True)
async def solve_with_code(self, task: str, max_iterations: int = 5) -> Dict:
"""用代码解决问题：写→执行→观察→修改→再执行"""
iteration = 0
history = []
while iteration < max_iterations:
iteration += 1
# 1. 模型生成/修改代码
if history:
last_error = history[-1].get("error", "")
last_output = history[-1].get("output", "")
prompt = f"""[任务] {task}
上次执行的输出:
{last_output}
上次的错误:
{last_error}
请修改代码以修复错误并完成任务。只输出代码块。"""
else:
prompt = f"[任务] {task}\n\n请写 Python 代码完成任务。只输出代码块。"
response = await self.llm.chat(prompt)
code = self._extract_code(response)
# 2. 保存并执行
filepath = f"{self.workspace}/solution_{iteration}.py"
with open(filepath, "w") as f:
f.write(code)
exec_result = await self._execute_code(filepath)
history.append({
"iteration": iteration,
"file": filepath,
"code": code,
"output": exec_result.get("stdout", ""),
"error": exec_result.get("stderr", ""),
"success": exec_result.get("returncode") == 0,
})
# 3. 检查是否成功
if exec_result.get("returncode") == 0 and not exec_result.get("stderr"):
return {"success": True, "history": history, "final_code": code}
return {"success": False, "history": history, "error": "Max iterations reached"}
def _extract_code(self, text: str) -> str:
import re
m = re.search(r'```(?:python)?\n(.*?)```', text, re.DOTALL)
return m.group(1) if m else text
async def _execute_code(self, filepath: str, timeout: int = 30) -> Dict:
import subprocess
try:
result = subprocess.run(
["python", filepath],
capture_output=True,
text=True,
timeout=timeout,
cwd=self.workspace,
)
return {
"returncode": result.returncode,
"stdout": result.stdout[-2000:],
"stderr": result.stderr[-2000:],
}
except subprocess.TimeoutExpired:
return {"returncode": -1, "stdout": "", "stderr": "Execution timed out"}
except Exception as e:
return {"returncode": -1, "stdout": "", "stderr": str(e)}Computer Use Agent 的安全边界
| 安全层 | 措施 | 说明 |
|---|---|---|
| 沙箱执行 | Docker / VM 隔离 | 在隔离环境中操作，不触碰宿主机 |
| 操作白名单 | 允许的操作类型列表 | 禁止 rm -rf、format 等危险操作 |
| 人类确认 | 关键操作前请求确认 | 涉及文件删除、网络请求时弹窗确认 |
| 操作审计 | 全量录屏 + 日志 | 每次 Computer Use 会话完整录像 |
| 速率限制 | 限制操作频率 | 防止过快操作导致不可逆后果 |
| 回滚能力 | VM 快照 | 操作出错后一键恢复到操作前状态 |
典型应用场景
| 场景 | Agent 类型 | 示例 |
|---|---|---|
| Web 自动化测试 | Computer Use | "打开登录页，输入账号密码，点击登录，验证跳转到首页" |
| 数据录入 | Computer Use | "从这个 Excel 表格读取数据，逐行录入到网页表单" |
| RPA 替代 | Computer Use | "打开邮件客户端，查找未读邮件，提取附件保存" |
| 数据分析 | Code Agent | "读取 CSV，做统计，生成图表，输出结论" |
| 代码迁移 | Code Agent | "将这个 Python 2 项目升级到 Python 3，修复所有兼容性问题" |
| Bug 修复 | Code Agent | "运行测试，定位失败原因，修改代码，确认测试通过" |
核心原则：Computer Use 和 Code Agent 不是要替代所有 API 集成，而是填补"没有 API 可调"的空白。有 API 就用 API（高效、稳定），没有 API 时才用屏幕操作（通用、但慢且脆弱）。
登录后即可发表评论 👇