AI Agent 全景：从 LLM 到自主智能体系统的 7 层深度实践

背景与问题定义：为什么“大模型 + Agent”不是噱头

大模型（LLM）具备“世界知识”与“推理”能力，但本身无状态、无行动、无目标。Agent 系统通过

1. 规划（Planning）
2. 记忆（Memory）
3. 工具（Tools）
4. 行动（Action）
   把 LLM 升级为“目标驱动的自主智能体”。2025 年 3 月，OpenAI 发布“Agent Eval”白皮书，指出在真实 API 调用场景下，带记忆与工具的 Agent 任务完成率比纯 LLM 提高 42%。因此，理解并落地 AI Agent 已成为算法与后端工程师的“必修课”。

技术栈总览：7 层能力模型与开源地图

Part-1 单 Agent 基础：ReAct 范式与可执行代码

1.1 ReAct 原理

ReAct = Reason + Act，用“思考→行动→观察”循环把 LLM 推理与外部 API 交织在一起。

1.2 最小可运行代码（200 行内）

环境：Python 3.11，OpenAI SDK 1.3+

# agent.py
import openai, json, os, re
from typing import List, Dict

openai.api_key = os.getenv("OPENAI_API_KEY")

class ReActAgent:
    def __init__(self, tools: List[Dict], max_step=5):
        self.tools = {t["name"]: t for t in tools}
        self.max_step = max_step
        self.history = []

    def _prompt(self, query: str) -> str:
        tool_desc = "\n".join([f"{t['name']}: {t['desc']}" for t in self.tools.values()])
        return (
            "Answer the following question using the ReAct paradigm.\n"
            f"Tools available:\n{tool_desc}\n"
            "Respond in the format:\n"
            "Thought: ...\nAction: tool_name(input)\nObservation: ...\n"
            "Repeat until you have the answer. Final Answer: ...\n"
            f"Question: {query}"
        )

    def _parse(self, text: str):
        thought = re.search(r"Thought: (.*?)\n", text, re.S)
        action = re.search(r"Action: (\w+)\((.*?)\)", text)
        return thought.group(1) if thought else None, \
               (action.group(1), action.group(2)) if action else (None, None)

    def run(self, query: str) -> str:
        prompt = self._prompt(query)
        for _ in range(self.max_step):
            msg = [{"role": "user", "content": prompt}]
            resp = openai.ChatCompletion.create(model="gpt-4", messages=msg, temperature=0)
            text = resp.choices[0].message.content
            self.history.append(text)
            thought, (tool_name, arg) = self._parse(text)
            if not tool_name:
                return re.search(r"Final Answer: (.*)", text).group(1)
            if tool_name in self.tools:
                try:
                    arg = json.loads(arg)
                    observation = self.tools[tool_name]["func"](**arg)
                except Exception as e:
                    observation = str(e)
                prompt += f"\nObservation: {observation}\n"
        return "Max step reached"

# 工具示例：查询天气 & 计算器
def get_weather(city: str) -> str:
    return f"{city} is sunny."

def calc(a: float, b: float, op: str) -> float:
    return {"+": a+b, "-": a-b, "*": a*b, "/": a/b}[op]

if __name__ == "__main__":
    tools = [
        {"name": "get_weather", "desc": "get_weather(city:str)->str", "func": get_weather},
        {"name": "calc", "desc": "calc(a:float,b:float,op:str)->float", "func": calc}
    ]
    agent = ReActAgent(tools)
    print(agent.run("What is the weather in Beijing and 3.5*2?"))

输出：

Final Answer: Beijing is sunny and 3.5*2 = 7.0

Part-2 长期记忆：向量库 + 记忆摘要双通道实现

2.1 记忆分类

• 工作记忆：最近 k 轮对话
• 长期记忆：用户画像、任务历史，用向量检索

2.2 代码实战

依赖：chromadb 0.4、sentence-transformers

# memory.py
import chromadb, uuid
from sentence_transformers import SentenceTransformer

class Memory:
    def __init__(self, emb_model="all-MiniLM-L6-v2"):
        self.emb = SentenceTransformer(emb_model)
        self.chroma = chromadb.Client()
        self.collection = self.chroma.create_collection("long_term")

    def add(self, text: str, meta: dict=None):
        self.collection.add(
            embeddings=[self.emb.encode(text).tolist()],
            documents=[text],
            metadatas=[meta or {}],
            ids=[str(uuid.uuid4())]
        )

    def search(self, query: str, topk=3) -> List[str]:
        emb = self.emb.encode(query).tolist()
        res = self.collection.query(query_embeddings=[emb], n_results=topk)
        return res["documents"][0]

把 Memory 嵌入 ReActAgent：

def run_with_memory(self, query: str) -> str:
    hist = self.memory.search(query)
    prompt = self._prompt(query) + "\nRelevant memory:\n" + "\n".join(hist)
    ...

Part-3 工具调用：Function Calling 与 100+ 工具生态

3.1 OpenAI Function Calling 原生模式

2025 年 6 月，OpenAI 把并发工具数上限提到 30 个，支持 async。

tools_schema = [
    {
        "type": "function",
        "function": {
            "name": "sql_query",
            "description": "Run SQL on Snowflake",
            "parameters": {
                "type": "object",
                "properties": {"sql": {"type": "string"}},
                "required": ["sql"]
            }
        }
    }
]

3.2 Composio：一键接入 GitHub、Notion、Slack

from composio import Action, ComposioToolSet
toolset = ComposioToolSet()
tools = toolset.get_tools(actions=[Action.GITHUB_STAR_REPO])

Part-4 多 Agent 协作：GroupChat 与角色扮演架构

4.1 AutoGen GroupChat

AutoGen 把 Agent 抽象为 ConversableAgent，通过“群聊”完成分布式任务。

from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager

coder = AssistantAgent("coder", system_message="You write Python.")
tester = AssistantAgent("tester", system_message="You write unit test.")
proxy = UserProxyAgent("proxy", code_execution_config={"work_dir": "coding"})

groupchat = GroupChat(agents=[proxy, coder, tester], messages=[], max_round=6)
manager = GroupChatManager(groupchat=groupchat, llm_config={"model": "gpt-4"})

proxy.initiate_chat(manager, message="Write a CLI calculator with tests.")

4.2 角色扮演 vs. 主从架构

• 角色扮演：每个 Agent 有 persona，适合创意写作
• 主从：Boss 分解任务，Worker 执行，适合爬虫、数据分析

Part-5 人类对齐：RLHF + Constitutional AI 的落地

5.1 数据飞轮

1. 收集人类偏好对 (x, y_w, y_l)
2. 训练 Reward Model
3. PPO 微调 Agent

5.2 代码：用 trl 库 30 行启动 RLHF

from transformers import AutoTokenizer, AutoModelForCausalLM
from trl import PPOTrainer, PPOConfig
from trl.core import respond_to_batch

model = AutoModelForCausalLM.from_pretrained("gpt2")
tokenizer = AutoTokenizer.from_pretrained("gpt2")
config = PPOConfig(model_name="gpt2", learning_rate=1.41e-5)
ppo_trainer = PPOTrainer(config, model, None, tokenizer)

query_tensor = tokenizer.encode("Write a summary of AI safety.", return_tensors="pt")
response_tensor = respond_to_batch(model, query_tensor)
reward = torch.tensor([1.5])  # 来自人类或RM
stats = ppo_trainer.step([query_tensor[0]], [response_tensor[0]], reward)

5.3 Constitutional AI

让 Agent 自我批评→修正，减少人类标注量 80%。

Part-6 评估体系：AgentBench 与自定义指标

6.1 AgentBench

覆盖操作系统、数据库、知识图谱、网页浏览等 8 类任务，已集成 1,000+ 环境。

6.2 自定义指标

• Success Rate
• 平均步数
• 工具调用准确率
• 人类满意度（Likert 5）

def evaluate(agent, dataset):
    ok = 0
    for case in dataset:
        out = agent.run(case["query"])
        ok += int(out == case["answer"])
    return ok / len(dataset)

Part-7 生产部署：Docker + FastAPI + 事件总线

7.1 架构图

[Browser] <--WS--> [FastAPI Gateway] <--Kafka--> [Agent Service Pool]  
                                     ^  
                                     |--[Redis 记忆缓存]  
                                     |--[Postgres 日志]  

7.2 Dockerfile（多阶段）

FROM python:3.11-slim as builder
COPY requirements.txt .
RUN pip install --user -r requirements.txt

FROM python:3.11-slim
COPY --from=builder /root/.local /usr/local
COPY . /app
WORKDIR /app
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

7.3 水平扩展

• 每个容器 1 Agent，CPU 限制 2 core
• Kafka partition key = user_id，保证会话一致性

彩蛋：10 行代码把 GPT-4 变成“自主科研助理”

import openai, arxiv
query = "latest RLHF papers"
papers = arxiv.Search(query="cat:cs.LG RLHF", max_results=5).results()
summary = openai.ChatCompletion.create(
    model="gpt-4",
    messages=[{"role": "user", "content": f"Summarize: {[p.title for p in papers]}"}]
)
print(summary.choices[0].message.content)

结语与展望

2025 年，AI Agent 正从“玩具”走向“生产”。随着

• 上下文窗口突破 2M token
• 多模态工具（视觉、代码解释器、浏览器）成熟
• 成本下降 10 倍（MoE + 边缘量化）
  “一人公司”式的 Agent 集群将在 3 年内普及。