AI Agent 完整关键词体系

一份覆盖 Agent 全生命周期的关键词体系，每个层次都配有 Python 示例，聚焦"这项能力在代码里长什么样"。

完整关键词体系

分组说明

基础六项（必备项）

• ReAct Agent
  ：怎么想
• Planning
  ：怎么拆任务
• Memory
  ：记住什么
• Function Calling / Tool Use
  ：怎么下指令、执行动作
• MCP
  ：工具怎么标准化接入
• FastAPI
  ：怎么对外服务

扩展五项（完善项）

• Multi-Agent Collaboration
  ：单 Agent 不够时，怎么分工协作
• Learning & Evolution
  ：怎么从执行中持续变强
• AgentOps
  ：怎么可运维、可观测、可优化
• Security
  ：怎么控制权限、防止过度代理、保证可审计
• Evaluation
  ：怎么衡量它到底做得好不好

一句话串起来

ReAct 是思考循环，Planning 和 Memory 是能力支柱，Function Calling/Tool Use 是执行接口，MCP 是连接标准，FastAPI 是服务容器；在此之上，Multi-Agent 扩展规模，Learning 驱动进化，AgentOps 保障运行，Security 守住底线，Evaluation 验证效果。

1. 思考模式：ReAct Agent

核心是"Thought → Action → Observation"循环。下面用最朴素的方式手写一个 ReAct 循环，不依赖框架，方便看清本质。

import re

defreact_agent(question, tools, llm, max_steps=5):
"""
    tools: dict[str, callable]
    llm: callable(prompt) -> str
    """
    scratchpad = ""
for step in range(max_steps):
        prompt = f"""你是一个 ReAct Agent。可用工具：{list(tools.keys())}

问题：{question}
{scratchpad}

请严格按以下格式输出：
Thought: 你的推理
Action: 工具名[输入]
"""
        output = llm(prompt)
        print(output)

# 解析 Action
        match = re.search(r"Action:\s*(\w+)\[(.*?)\]", output)
ifnot match:
return output  # 没有动作，视为最终回答

        tool_name, tool_input = match.group(1), match.group(2)
if tool_name notin tools:
            observation = f"错误：未知工具 {tool_name}"
else:
            observation = tools[tool_name](tool_input)

        scratchpad += f"\n{output}\nObservation: {observation}\n"

return"达到最大步数，未完成"

# 示例工具
tools = {
"search": lambda q: f"关于'{q}'的搜索结果：...",
"calculator": lambda expr: str(eval(expr)),
}

# 伪 LLM，仅演示
deffake_llm(prompt):
if"搜索"notin prompt:
return"Thought: 我需要先搜索\nAction: search[AI Agent]"
return"Thought: 已获得信息\nAction: finish[AI Agent 是一种智能系统]"

print(react_agent("什么是 AI Agent？", tools, fake_llm))

关键点：Thought 是显式推理，Action 是结构化调用，Observation 回填到 scratchpad，下一轮继续。

2. 规划：Planning (CoT / ToT / Plan-Execute)

这里给 Plan-and-Execute 的示例：先让 LLM 产出一个计划列表，再逐步执行，执行中可动态修改计划。

defplan_and_execute(goal, llm, tools):
# 1. 制定计划
    plan_prompt = f"""目标：{goal}
可用工具：{list(tools.keys())}
请输出一个步骤列表，每行一个步骤，格式：步骤编号. 动作描述"""
    plan_text = llm(plan_prompt)
    plan = [line.strip() for line in plan_text.splitlines() if line.strip()]
    print("初始计划：", plan)

    results = []
for step in plan:
# 2. 为每一步选择工具并执行
        action_prompt = f"当前步骤：{step}\n可用工具：{list(tools.keys())}\n请输出：工具名[输入]"
        action = llm(action_prompt)
        tool_name = action.split("[")[0]
        tool_input = action.split("[")[1].rstrip("]")

if tool_name in tools:
            result = tools[tool_name](tool_input)
            results.append(result)

# 3. 动态重规划（简化示意）
if"失败"in str(result):
            plan.append("重试上一步或换工具")

return results

deffake_llm(prompt):
if"步骤列表"in prompt:
return"1. 搜索资料\n2. 总结要点"
if"搜索"in prompt:
return"search[AI Agent 架构]"
return"summarize[资料]"

tools = {
"search": lambda q: f"搜索结果：{q}",
"summarize": lambda t: f"总结：{t}",
}
print(plan_and_execute("写一篇 AI Agent 简介", fake_llm, tools))

关键点：计划与执行分离，执行结果可以触发计划更新（这里用 append 简化示意）。CoT 是单链推理，ToT 是在每步生成多个候选并评估，Plan-Execute 是先全局规划再逐步落地。

3. 记忆：Memory (短期 / 长期 / 工作)

短期记忆用对话窗口，长期记忆用向量库，工作记忆用任务状态字典。下面是一个综合示意。

from collections import deque

classAgentMemory:
def__init__(self, short_term_size=5):
# 短期记忆：最近几轮对话
        self.short_term = deque(maxlen=short_term_size)
# 长期记忆：伪向量库，实际可用 Chroma / Milvus
        self.long_term = []
# 工作记忆：当前任务中间状态
        self.working = {}

defadd_dialogue(self, role, content):
        self.short_term.append({"role": role, "content": content})

defadd_knowledge(self, text, embedding=None):
# 真实场景：embedding = model.encode(text)
        self.long_term.append({"text": text, "embedding": embedding})

defretrieve(self, query, top_k=2):
# 真实场景：向量相似度检索
return [item["text"] for item in self.long_term[:top_k]]

defset_state(self, key, value):
        self.working[key] = value

defget_context(self, query):
return {
"short_term": list(self.short_term),
"long_term": self.retrieve(query),
"working": self.working,
        }

mem = AgentMemory()
mem.add_dialogue("user", "帮我订一张去北京的票")
mem.add_dialogue("assistant", "请问出发日期？")
mem.add_knowledge("用户偏好：靠窗座位")
mem.set_state("task", "订票")
mem.set_state("destination", "北京")

print(mem.get_context("订票"))

关键点：短期记忆控制上下文长度，长期记忆跨会话检索，工作记忆保存任务执行中的结构化状态。

4. 工具接口：Function Calling / Tool Use

以 OpenAI 风格的 function calling 为例，模型返回结构化的工具调用指令，由你的代码执行。

import json

# 1. 定义工具 Schema
tools_schema = [
    {
"type": "function",
"function": {
"name": "get_weather",
"description": "查询指定城市的当前天气",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名，如北京"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
                },
"required": ["city"],
            },
        },
    }
]

# 2. 实际执行函数
defget_weather(city, unit="celsius"):
return {"city": city, "temp": 22, "unit": unit, "condition": "晴"}

# 3. 模拟模型返回的 tool_call
deffake_llm_tool_call(user_input):
return {
"tool_calls": [
            {
"id": "call_1",
"function": {
"name": "get_weather",
"arguments": json.dumps({"city": "北京", "unit": "celsius"}),
                },
            }
        ]
    }

# 4. 解析并执行
defhandle_tool_call(response):
    results = []
for call in response["tool_calls"]:
        name = call["function"]["name"]
        args = json.loads(call["function"]["arguments"])
if name == "get_weather":
            results.append(get_weather(**args))
return results

resp = fake_llm_tool_call("北京天气怎么样")
print(handle_tool_call(resp))

关键点：Schema 里的 description 决定模型"何时选这个工具"，required 和 enum 约束参数，执行和错误处理由你的代码负责。

5. 连接标准：MCP

MCP 是 Client-Server 协议，Server 暴露工具，Client（Agent）发现并调用。下面用官方 Python SDK 风格写一个最小 MCP Server 和 Client 调用示意。

MCP Server（暴露工具）

# server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("demo-server")

@mcp.tool()
defadd(a: int, b: int) -> int:
"""两数相加"""
return a + b

@mcp.tool()
defsearch_docs(query: str) -> str:
"""搜索内部文档"""
returnf"关于 {query} 的文档结果..."

if __name__ == "__main__":
    mcp.run()  # 默认 stdio 传输

MCP Client（Agent 侧调用）

# client.py
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

asyncdefmain():
    params = StdioServerParameters(command="python", args=["server.py"])
asyncwith stdio_client(params) as (read, write):
asyncwith ClientSession(read, write) as session:
await session.initialize()

# 1. 发现工具
            tools = await session.list_tools()
            print("可用工具：", [t.name for t in tools.tools])

# 2. 调用工具
            result = await session.call_tool("add", {"a": 3, "b": 5})
            print("调用结果：", result.content)

asyncio.run(main())

关键点：Server 用 @mcp.tool() 声明能力，Client 先 list_tools 发现、再 call_tool 调用，传输层可以是 stdio 或 HTTP/SSE。这样 Agent 不需要为每个工具写定制集成。

6. 服务化：FastAPI

把 Agent 封装成 HTTP 服务，管理会话和请求。

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Dict
import uuid

app = FastAPI(title="Agent Service")

# 简易会话存储，生产可用 Redis
sessions: Dict[str, list] = {}

classChatRequest(BaseModel):
    session_id: str | None = None
    message: str

classChatResponse(BaseModel):
    session_id: str
    reply: str

defrun_agent(message: str, history: list) -> str:
# 这里接入你的 ReAct / Plan-Execute / LLM
returnf"Agent 收到：{message}（历史 {len(history)} 条）"

@app.post("/chat", response_model=ChatResponse)
asyncdefchat(req: ChatRequest):
    session_id = req.session_id or str(uuid.uuid4())
    history = sessions.get(session_id, [])

try:
        reply = run_agent(req.message, history)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))

    history.append({"role": "user", "content": req.message})
    history.append({"role": "assistant", "content": reply})
    sessions[session_id] = history

return ChatResponse(session_id=session_id, reply=reply)

@app.get("/health")
asyncdefhealth():
return {"status": "ok"}

# 运行：uvicorn main:app --reload

关键点：session_id 维护会话，Pydantic 做请求校验，异常转 HTTP 错误码，/health 供探活。生产环境再加鉴权、限流、超时和流式响应。

7. 协作：Multi-Agent Collaboration

多个专用 Agent 分工，由一个 Orchestrator 调度。下面示意"研究员 + 写作者 + 审校者"的流水线。

classAgent:
def__init__(self, name, role, llm):
        self.name = name
        self.role = role
        self.llm = llm

defrun(self, task):
        prompt = f"你是{self.role}。任务：{task}"
return self.llm(prompt)

deffake_llm(prompt):
if"研究员"in prompt:
return"研究结果：AI Agent 的关键技术包括 ReAct、Memory、MCP..."
if"写作者"in prompt:
return"文章草稿：AI Agent 正在改变软件形态..."
if"审校者"in prompt:
return"审校意见：结构清晰，建议补充安全部分。"
return"完成"

classOrchestrator:
def__init__(self, agents):
        self.agents = agents

defrun(self, goal):
        context = goal
for agent in self.agents:
            result = agent.run(context)
            print(f"[{agent.name}] {result}")
            context = result  # 上游输出作为下游输入
return context

agents = [
    Agent("researcher", "研究员", fake_llm),
    Agent("writer", "写作者", fake_llm),
    Agent("reviewer", "审校者", fake_llm),
]
print(Orchestrator(agents).run("写一篇 AI Agent 介绍"))

关键点：每个 Agent 有独立角色和提示词，Orchestrator 决定顺序、并行或分层调度。复杂场景可引入协商、投票或共享黑板。

8. 学习进化：Learning & Evolution

Agent 从执行反馈中更新策略或知识。下面示意"失败后把经验写入长期记忆，下次检索复用"。

classEvolvingAgent:
def__init__(self):
        self.experience = []  # 经验库

defact(self, task):
# 1. 检索相似经验
        relevant = [e for e in self.experience if task in e["task"]]
if relevant:
            print(f"复用经验：{relevant[-1]['lesson']}")
return"成功（基于经验）"

# 2. 无经验，尝试执行
        success = self.try_execute(task)
ifnot success:
            lesson = f"任务'{task}'失败，应换用工具 B"
            self.experience.append({"task": task, "lesson": lesson})
            print(f"记录教训：{lesson}")
return"成功"if success else"失败"

deftry_execute(self, task):
returnFalse# 模拟首次失败

agent = EvolvingAgent()
agent.act("订机票")
agent.act("订机票")  # 第二次复用经验

关键点：经验以结构化形式沉淀（任务、教训、成功路径），下次执行前检索复用。更高级的做法包括反思（Reflexion）、自我批评和参数微调。

9. 工程化：AgentOps / 全生命周期

用装饰器给 Agent 加上追踪、耗时统计和日志，这是 AgentOps 的最小落地形态。

import time, logging, functools

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("agentops")

# 简易 trace 存储，生产可用 LangSmith / Langfuse / OpenTelemetry
traces = []

deftraced(func):
    @functools.wraps(func)
defwrapper(*args, **kwargs):
        start = time.time()
        trace = {"func": func.__name__, "input": str(args)[:100]}
try:
            result = func(*args, **kwargs)
            trace["status"] = "success"
return result
except Exception as e:
            trace["status"] = "error"
            trace["error"] = str(e)
raise
finally:
            trace["latency_ms"] = round((time.time() - start) * 1000, 2)
            traces.append(trace)
            logger.info(f"{trace}")
return wrapper

@traced
defcall_llm(prompt):
    time.sleep(0.1)
returnf"回答：{prompt}"

@traced
defrun_tool(name, arg):
if name == "bad":
raise ValueError("工具执行失败")
returnf"{name} 结果"

call_llm("你好")
try:
    run_tool("bad", "x")
except ValueError:
pass

print("Traces:", traces)

关键点：每次 LLM 调用和工具调用都留下 trace，包含输入、状态、延迟、错误。生产系统再接入集中式观测平台、成本统计、告警和 CI/CD。

10. 安全：Excessive Agency / 最小权限 / HITL

用"权限白名单 + 人工审批闸门 + 成本熔断"约束 Agent 的自主性。

classSecureAgent:
def__init__(self):
# 最小权限：只允许这些工具
        self.allowed_tools = {"search", "read_file"}
# 高风险工具需要人工审批
        self.sensitive_tools = {"delete_file", "send_email"}
        self.max_cost = 1.0
        self.cost = 0.0

defrequest_tool(self, tool_name, args):
# 1. 权限校验
if tool_name notin self.allowed_tools and tool_name notin self.sensitive_tools:
returnf"拒绝：{tool_name} 不在权限白名单"

# 2. 成本熔断
if self.cost >= self.max_cost:
return"熔断：已超出成本预算"

# 3. 高风险操作人工审批（HITL）
if tool_name in self.sensitive_tools:
ifnot self.human_approve(tool_name, args):
returnf"拒绝：{tool_name} 未获人工批准"

        self.cost += 0.1
return self.execute(tool_name, args)

defhuman_approve(self, tool_name, args):
# 真实场景：发审批请求，等待人工确认
        print(f"[HITL] 请求执行 {tool_name}，参数 {args}")
returnFalse# 模拟人工拒绝

defexecute(self, tool_name, args):
returnf"已执行 {tool_name}"

agent = SecureAgent()
print(agent.request_tool("search", "AI"))
print(agent.request_tool("delete_file", "/data"))
print(agent.request_tool("unknown_tool", "x"))

关键点：白名单收敛权限，敏感工具走 HITL 审批，成本熔断防止失控循环，所有拒绝都要记录审计日志。OWASP 的 Excessive Agency 正是针对"功能过多、权限过大、自主过高"。

11. 评估：Evaluation

从任务完成率、工具调用正确率、延迟、安全合规多维度打分。

classAgentEvaluator:
def__init__(self):
        self.cases = []

defadd_case(self, task, expected_tool, expected_output):
        self.cases.append({
"task": task,
"expected_tool": expected_tool,
"expected_output": expected_output,
        })

defevaluate(self, agent_fn):
        total = len(self.cases)
        task_success = 0
        tool_correct = 0
        latencies = []

for case in self.cases:
import time
            start = time.time()
            result = agent_fn(case["task"])  # 返回 {"tool": ..., "output": ...}
            latencies.append(time.time() - start)

if result["output"] == case["expected_output"]:
                task_success += 1
if result["tool"] == case["expected_tool"]:
                tool_correct += 1

return {
"task_success_rate": task_success / total,
"tool_accuracy": tool_correct / total,
"avg_latency_s": sum(latencies) / total,
"total_cases": total,
        }

# 被测 Agent（简化）
defmy_agent(task):
if"天气"in task:
return {"tool": "get_weather", "output": "晴"}
return {"tool": "search", "output": "未知"}

evaluator = AgentEvaluator()
evaluator.add_case("北京天气", "get_weather", "晴")
evaluator.add_case("查资料", "search", "结果")

print(evaluator.evaluate(my_agent))

关键点：评估用例要覆盖正常、边界和对抗场景；指标不仅看最终答案，还要看工具选择是否正确、延迟是否可接受、是否触发安全拒绝。生产环境通常结合确定性测试（如 SWE-bench）和 LLM-as-Judge 做多维度评分。

整体串联

这 11 项在代码里是逐层叠加的：

1. ReAct
    提供循环骨架
2. Planning
    和 Memory 增强单次执行能力
3. Function Calling
    和 MCP 解决工具接入
4. FastAPI
    把能力暴露成服务
5. Multi-Agent
    扩展规模
6. Learning
    让系统持续改进
7. AgentOps
    保证可观测可运维
8. Security
    约束边界
9. Evaluation
    验证效果

实际项目不必一次全上，可以从 ReAct + Function Calling + FastAPI 起步，再按需引入其余模块。