Agent的本质：从工具到伙伴
Agent的本质：从工具到伙伴
从技术的角度来看，Agent 可能只是一个不断调用 LLM API 的死循环脚本。然而，从交互哲学的角度来看，它标志着人机关系的根本性重构：从“命令与控制”转向“意图与协作”。
从命令到对话：交互原语的升维
在这段对话中，Agent 并非在“执行命令”，而是在“理解意图”“提供建议”和“参与讨论”。它更像一位经验丰富的同事，而非一个机械的命令行工具。
工具的根本特征是被动性与机械性：它们等待指令、执行指令，然后继续等待下一条指令。
Agent打破了这一范式：它们主动感知环境，自主思考与规划，采取行动，并能够从经验中持续学习。
- 在CLI（Command-Line Interface，命令行界面）时代，人机交互的基本单 元是“指令”。
- 在GUI时代，人机交互的基本单元演变为“点击”。这两种模式共享一个根本前提：人类必须负责“怎么做”（How），而机器仅负责执行。
例如，用户与机器的操作分别如下。 - 用户：打开文件→选择文字→点击“加粗”按钮→点击“保存”按钮。
- 机器（如Word软件）：依次执行上述操作。
在Agent 时代，人机交互的基本单元已转变为“对话”，更准确地说，是“意图”。
例如，用户与 Agent 的操作分别如下。 - 用户：把这份文档改得正式一点。
- Agent：理解“正式”的语境含义→读取文档内容一分析段落结构→重写文本→ 保存结果。
意图的理解：模糊与精确的桥梁
传统软件不仅缺乏主动性，而且拘泥于字面含义。例如，少写一个分号，编译器就会报错——它无法理解你的“言外之意”。
Agent的核心能力在于充当一个“模糊翻译器”，搭建起“人类模糊的自然语言”与
“机器精确的 AP| 调用” 之间的桥梁。
- 输入：模糊的意图，例如，“帮我查下这周还有没有去上海的便宜机票”。
- 处理：进行推理、消歧，并补全必要参数。
- 输出：生成精确的调用指令。
语义理解
提取用户表达背后的动作、对象、范围和限制条件。
例如，当用户说“我想把昨天的日志整理一下”，其中“整理”是指分析、压缩、归
档还是其他操作？这都需要进一步澄清。
目的识别
不仅要理解“你说什么”，更要理解“你想实现什么”。
例如，当用户说“这段代码总是返回None”，其目的往往并非仅仅调试，而是希望
“获得正确的返回结果”。
Google、OpenAl、DeepSeek 等头部 Al 公司的最新研究普遍强调，Agent 应能
从“显性指令”中推断出“隐性目标”。以OpenAl o1的“Reasoning Tokens”为例，
虽然其表面是在进行推理，但本质上是为了更准确地捕捉用户的隐含意图。
上下文关联
Agent必须将意图“落地”到具体环境中，这包括以下几个关键要素。
■用户的代码结构。
■当前文件目录。
■用户偏好（例如，倾向于使用yield语句还是列表推导式）。
■安全约束（例如，禁止删除系统文件）。
■历史任务上下文（例如，上次正在编写的数据管道）。
这正是上下文工程的核心价值所在：Agent必须基于当前世界状态进行判断和决策。
风险评估与澄清
Agent必须在“意图模糊”时主动提问，而非贸然执行，以避免误操作，这样既确保
任务能够准确完成，又符合安全策略。
记忆与身份：Agent的自我连续性
工具是“用完即走”的，但伙伴是有记忆的。试想，如果你的同事每天早上见到你，
都像初次见面一样问：“你是谁？你要干什么？”你恐怕会崩溃。
每次打开工具时，都像是面对一块空白的白板，但 Agent 则不同，它拥有记忆。
Agent 在每次对话之间延续其“身份”，并逐渐形成对用户的深入理解，如图3-3所示。
记忆：连续性的幻觉
LLM本身是无状态的。Agent 所谓的“记忆”，实际上是通过上
下文工程构建出的一种“幻觉”。
短期记忆（SessionMemory）：记录当前会话的上下文，包括用户上一句话的内容、最近讨论的话题、代码片段，以及用户此前纠正过的误解等。
短期记忆使Agent 能够实现以下目标。
■在多轮对话中保持一致性。
■避免重复提问。
■进行上下文连续的推理。
如今，最先进的上下文工程技术会对短期记忆进行“摘要、过滤与结构化”处理，使
Agent 能在高达 20 万至 100 万个 Token 的大上下文中保持精准推理。
长期记忆（Long-termMemory)：借助向量数据库等技术，记录用户跨会话的偏
好信息，例如以下内容。
■用户偏好的代码风格。
■常用工具栈（如 Python、LangChain、Spark ）。
■历史项目结构。
■过往提出的问题类型。
■用户的角色设定（例如，“你更偏好可读性高的代码”）。
这使Agent具备了“持续服务能力”：下次打开时，它能自然延续上一次的思路。
目前，Mem0、GraphRAG、Hybrid Memory Systems 等架构已将长期记忆实
现为可建模、可检索、可索引的形式，使Agent 能够像人类一样“内化知识”。
System Prompt的灵魂
你可以给 Agent 编写一段 System Prompt：“你是一个刻薄但专业的代码审查员。”
顷刻之间，这个无形的概率模型便坍缩成一个具体的人格（persona）。
身份（identity）远不止于人设，它是一套价值观与行为准则的集合，决定了Agent
在面临两难选择时（如效率与安全之间的权衡）会倾向于哪一方。
记忆是“知道你是谁”，身份则是“知道自己是谁”。
一个成熟的Agent 会在以下方面保持稳定。
表达风格。
■专业领域定位。
■价值观倾向（例如，偏向保守操作或注重安全）。
■行动策略偏好（例如，先检查，再执行）。
这些设计让用户感受到：“这是同一个 Agent，也是我的 Agent。”
记忆=连续性，
身份=稳定性。
记忆赋予Agent连续性与个性化能力，使其能够跨越对话和任务保持一致性与相
关性。在工具时代，这种能力并不存在，而这正是Agent从“软件”进化为“协作者”的
根本基础。
上下文工程：理解世界状态
意图理解与身份信息使 Agent 明确“你希望达成什么目标”，而上下文工程则使
Agent 理解“当前世界的状况是怎样的”，如图3-4所示。
LLM本身并不具备主动感知世界的能力，它只能依据输入的文本进行推理。因此，
对任何 Agent 而言，都受制于这样一句话：“Garbage in, garbage out"(输入的是垃圾，输出的也必然是垃圾）。Agent失败的主要原因，往往并非模型的能力不足，而是上下文信息提供有误。
对 Agent 而言，上下文就是它的整个世界。上下文工程的目标：将Agent完成任务所需的最关键信息，以最合适的结构提供给
模型。架构师的新职责是设计一套机制，以便能够实时将现实世界的状态序列化为文本，并将其投射到 Agent 的上下文中。这正是上下文工程的本质——为 Agent 搭建数字感官。
上下文工程的3个核心来源：
■用户上下文（UserContext）：涵盖用户的开发风格、历史偏好（如偏好的列表
推导式）、常用技术栈（如 Spark、LangChain、FastAPI）以及当前工作内容
（如正在编写爬虫或构建RAG系统）。一个优秀的Agent 能够利用这些信息，自
动调整回答的风格与推理深度。
■环境上下文（Environment Context)：用于理解“世界状态”，包括文件结
构、代码库内容、当前分支、日志、错误栈、变量值、API响应以及运行时状态
（如 CPU、GPU°使用情况和任务状态）。调试型 Agent 必须先看到堆栈跟踪
（Stack Trace），才能给出准确的建议。
■任务上下文（TaskContext)：Agent 在执行任务时，必须掌握前序任务的结
果、当前子任务的进度、依赖项是否已加载，以及工具调用的历史尝试记录。这
是支持多步任务和复杂 Agent 工作流（如LangGraph）所必需的能力。
随着上下文窗口扩展到20 万个至 100万个Token，上下文工程（尤其是信息的筛选与压缩）变得更加重要：不能简单地将所有内容塞入上下文，而需要通过选择、摘要和结构化等手段，有效降低干扰。
上下文工程领域呈现出技术新趋势
■选择性上下文（Selective Context)：精准筛选与当前意图紧密相连的内容，
剔除无关信息，聚焦关键要素。
■结构化摘要（StructuredSummaries）：针对历史内容生成结构化摘要，包括
标题、关键发现和失败点等核心要素。
■图结构上下文（Graph-basedContext)：将知识库构建为图结构，并依据节
点重要性进行检索。
■上下文优先级排序（ContextPrioritization）：由模型自主判断不同内容的权重
高低。例如，有的模型引入了“思考模式”，赋予重要信息更高的优先级，使信
息处理更加符合用户需求和场景要求，优化了决策依据的生成过程。
上述技术的本质都服务于一个共同目标：最大限度地减少噪声，同时尽可能突出关键信息。
上下文工程并非简单地要求“喂信息”，而是要“喂结构”。优秀的上下文并非取决于“文字量多”，而在于“让模型立即理解”。当模型接收到高质量的结构化输入时，其推理精度可提升数倍。
上下文工程，正是打造高质量Agent 的真正秘诀。它的本质是将多维、动态的现实
世界，精准地投射为模型可理解的线性文本序列。如果将 Agent 比作一个在漆黑房间中解题的人，那么上下文工程如同那束手电筒的光——光所照之处，决定了Agent聚焦于代码库的哪一部分、日志的哪几行，以及用户的哪段历史，从而直接决定其能否成功解题。
任务分解：把“目标”变为“计划”
如果给一位初级工程师布置任务：建一座房子。他很可能会不知所措。
而给一位资深架构师同样的任务，他会迅速拆解出关键步骤：选址、设计图纸、打地基、建造主体、装修等。因此，任务分解（Task Decomposition）赋予了 Agent 将“目标”转化为具体“步骤”的能力，是衡量其智能化水平的关键指标。
任务分解包含两个关键环节。
■规划（Planning）：将总体目标拆解为若干子目标。
■映射（Mapping)：将每个子目标转化为具体的行动或可调用的工具。
这种“化整为零”的能力，使 Agent 能够处理远超单次 AP| 调用能力范围的长程复
杂任务。
在工具时代，复杂度的负担主要落在人身上。用户必须在脑海中将“修复bug”这一
目标拆解为“定位文件一修改代码一运行测试”等多个步骤，并依次手动执行。
Agent不再是单纯的单步执行者，而是具备规划能力的规划者（Planner)，其核心
在于引入了一个关键的中间层——规划层。
■理解（Understand)：将模糊的高层目标（如“优化数据管道”）转化为明确的
技术需求。
■规划（Plan）：生成可操作的多步方案，如“I/O分析→算法优化→缓存策略”。
■执行与反思（Execute and Critique)：这是最接近人类智能的环节——Agent
在执行完第一步后，可能基于新信息主动调整计划。例如，“我查看了日志，发
现I/O并非瓶颈，因此决定暂停原计划，转而排查CPU占用问题”。这种动态规划能力，是Agent与传统自动化脚本的根本区别所在。用户只需要说：“帮我优化这个数据管道，我希望速度快一点。”传统工具会回应：“请告诉我具体要修改哪些函数、怎么改。”而Agent则会主动回应：“我先分析当前的性能瓶颈，列出几个可能的优化方向，然后为你生成一个可执行的优化计划。”
1.任务分解的三步法
现代 Agent 内部普遍采用三步策略。
■理解目标：这并非仅仅解析字面意思，而是要洞察“真实需求”。例如，当用户
说“清理一下昨天的日志”，其真实意图可能是压缩、分析或归档日志，而非简
单地删除文件。
■确定路径：将目标转化为清晰、可执行的流程。例如，加载日志文件一按日期分
组一统计错误数量→生成分析报告→压缩原始日志。一个优秀的计划应具备以下
特征：步骤简洁、无歧义、可成功执行、支持中断、便于检查，并能根据执行反
馈动态调整。
■管理执行：包括工具调用、失败重试、异常处理、结果验证，以及向用户汇报进
度和动态调整计划。这一环节正是LangGraph、AutoGen、CrewAl 等 Agent
框架的核心价值所在——让Agent从单纯的“聊天模型”真正升级为可靠的“任
务执行者”。
2.学会自我审查
近期Agent研究领域呈现出一个明显趋势：在执行前对计划进行自我审查。
例如，DeepSeek、Qwen3 等模型引入的“思考模式”，均强调先生成初步计划，
再对该计划进行批判性分析，确认其合理性与可行性后，才进入执行阶段。
这一点非常关键，它使 Agent 的潜在错误在执行前就被识别和修正，从而避免在运行时酿成不可逆的事故。
3.任务分解与上下文工程互相强化
没有上下文工程，就无法正确分解任务；而没有任务分解，上下文工程也无从聚焦与取舍。二者相互依存，共同构成了Agent 完整的“执行链路”。
这正是 Agent 与传统工具的根本区别：它不只是回答问题，而是真正开展工作。
■上下文工程让 Agent“看得清楚”—准确理解环境与需求。
■任务分解让Agent“做得正确”——将目标转化为可执行、可调整的行动路径。
具体参考代码如下：
import os
import operator
from typing import Annotated, List, Tuple, Union, Literal
from typing_extensions import TypedDict
# --- 优先使用新版 tavily，若未安装则回退并提示 ---
try:
from langchain_tavily import TavilySearch # 新版
print("使用 langchain-tavily")
except ImportError:
from langchain_community.tools.tavily_search import TavilySearchResults as TavilySearch
print("提示：请运行 `pip install -U langchain-tavily` 消除弃用警告")
from pydantic import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langgraph.graph import StateGraph, START, END
from dotenv import load_dotenv
load_dotenv()
os.environ["TAVILY_API_KEY"] = os.getenv("TAVILY_API_KEY")
os.environ["DASHSCOPE_API_KEY"] = os.getenv("DASHSCOPE_API_KEY")
tools = [TavilySearch(max_results=3)]
llm = ChatOpenAI(
model="qwen3-max",
api_key=os.environ.get("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
temperature=0,
)
prompt = "You are a helpful assistant."
agent_executor = create_react_agent(llm, tools, prompt=prompt)
# --- 数据结构定义 ---
class Plan(BaseModel):
"""未来要执行的计划"""
steps: List[str] = Field(description="需要按顺序执行的不同步骤")
class PlanExecute(TypedDict):
input: str
plan: List[str]
past_steps: Annotated[List[Tuple], operator.add]
response: str
class Response(BaseModel):
"""对用户的响应"""
response: str
class Act(BaseModel):
"""要执行的动作"""
action: Union[Response, Plan] = Field(
description="要执行的动作。Response 表示直接回复，Plan 表示需要继续执行新计划。"
)
# --- Prompt 模板 ---
planner_prompt = ChatPromptTemplate.from_messages([
("system", """针对给定的目标，制定一个简单的逐步计划。
该计划应包含独立的任务，如果正确执行将得到正确答案。不要添加任何多余的步骤。
最后一步的结果就是最终答案。确保每个步骤都包含所有必要的信息——不要跳过步骤。"""),
("placeholder", "{messages}"),
])
planner = planner_prompt | llm.with_structured_output(Plan)
replanner_prompt = ChatPromptTemplate.from_template("""
针对给定的目标，制定一个简单的逐步计划。
该计划应包含独立的任务，如果正确执行将得到正确答案。不要添加任何多余的步骤。
最后一步的结果就是最终答案。确保每个步骤都包含所有必要的信息——不要跳过步骤。
你的原始目标是：
{input}
你最初的计划是：
{plan}
目前已经完成的步骤：
{past_steps}
请据此更新你的计划。**如果所有步骤均已成功完成，请使用 Response 直接给出最终答案**；否则，仅添加仍需完成的步骤作为新的 Plan。
""")
replanner = replanner_prompt | llm.with_structured_output(Act)
# --- 节点函数 ---
async def plan_step(state: PlanExecute):
"""根据用户输入生成初始计划"""
plan = await planner.ainvoke({"messages": [("user", state["input"])]})
return {"plan": plan.steps}
async def execute_step(state: PlanExecute):
"""执行计划中的第一个未完成步骤"""
plan = state["plan"]
if not plan:
return {"past_steps": []}
plan_str = "\n".join(f"{i + 1}. {step}" for i, step in enumerate(plan))
task = plan[0]
task_formatted = f"""针对以下计划：
{plan_str}\n\n你需要执行第1步：{task}。"""
agent_response = await agent_executor.ainvoke(
{"messages": [("user", task_formatted)]}
)
result = agent_response["messages"][-1].content
return {"past_steps": [(task, result)]}
async def replan_step(state: PlanExecute):
"""更新计划或生成最终响应"""
output = await replanner.ainvoke(state)
if isinstance(output.action, Response):
return {"response": output.action.response}
elif isinstance(output.action, Plan):
# 如果模型返回了一个空的计划（steps=[]），说明任务已完成，自动从 past_steps 提取最终答案
if not output.action.steps and state["past_steps"]:
final_answer = state["past_steps"][-1][1] # 取最后一步的结果
return {"response": final_answer}
else:
return {"plan": output.action.steps}
else:
# 兜底情况
return {"plan": []}
def should_end(state: PlanExecute):
"""判断是否结束"""
if state.get("response"):
return END
if not state.get("plan"):
return END
return "agent"
# --- 构建工作流 ---
workflow = StateGraph(PlanExecute)
workflow.add_node("planner", plan_step)
workflow.add_node("agent", execute_step)
workflow.add_node("replan", replan_step)
workflow.add_edge(START, "planner")
workflow.add_edge("planner", "agent")
workflow.add_edge("agent", "replan")
workflow.add_conditional_edges(
"replan",
should_end,
["agent", END],
)
app = workflow.compile()
# --- 运行测试 ---
async def main():
config = {"recursion_limit": 10}
inputs = {"input": "查询 2024 年男子澳大利亚网球公开赛冠军的家乡?"}
res = await app.ainvoke(inputs, config=config)
print("=== 最终结果 ===")
if res.get("response"):
print(res["response"])
else:
# 如果 response 仍为空，打印最后一条 past_steps
if res["past_steps"]:
print(res["past_steps"][-1][1])
else:
print("未找到答案。")
print("\n完整状态：", res)
if __name__ == "__main__":
import asyncio
asyncio.run(main())
结果如下：
工具、技能与行动策略：真正影响世界
如果没有工具，LLM不过是一个被困在服务器里的“缸中之脑”——它能写出完美的
SQL查询，却无法连接数据库；它还能制订周密的旅行计划，却无法预订一张机票。要真正成为人类的伙伴，它必须拥有“手”：能够调用工具、操作环境、执行动作。
工具使用（Tool Use)，或更学术地称为行动（Action)，是 Agent 获得 “数字具身性”（Digital Embodiment）的关键。它使 Agent 从一个“被动的观察者”转变为“主
动的参与者”。本质上，工具调用以语言模型为“大脑”，以各类工具为“身体”—通过语言理解与规划驱动外部的能力，实现对数字环境的真实干预，如图3-5所示。Agent与传统聊天模型最根本的分野，不在于“回答更准确”，而在于“能够行动”。要实现行动能力，Agent必须具备以下两项核心能力。
■选择合适工具的能力：根据任务目标和上下文，精准匹配并调用最有效的工具。
■制定安全、可靠、可预测的行动策略的能力：旨在规划清晰、可控且具备容错与
调整能力的执行路径。
这两项核心能力共同构成了长尾任务执行的基础：没有工具调用能力的Agent只能
“说”；而具备工具调用能力的 Agent才真正能够“做”。
具体代码参考：
#!/usr/bin/env python3
"""
本示例展示了如何实现一个带有工具调用能力的 AI Agent 循环。
Agent 循环的核心逻辑保持不变，只是在工具数组中添加了工具定义，
并通过一个分发映射表（dispatch map）来路由工具调用请求。
数据流向图:
+----------+ +-------+ +------------------+
| User | ---> | LLM | ---> | Tool Dispatch |
| prompt | | | | { |
+----------+ +---+---+ | bash: run_bash |
^ | read: run_read |
| | write: run_wr |
+----------+ edit: run_edit |
tool_result| } |
+------------------+
"""
# =============================================================================
# 标准库导入
# =============================================================================
import os # 操作系统接口，用于环境变量和路径操作
import subprocess # 子进程管理，用于执行 shell 命令
import json # JSON 解析，用于解析工具调用的参数
from pathlib import Path # 面向对象的路径操作
import logging
# 配置日志格式和级别
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# =============================================================================
# 第三方库导入
# =============================================================================
from openai import OpenAI # OpenAI 官方 Python SDK
from dotenv import load_dotenv # 从 .env 文件加载环境变量
# =============================================================================
# 环境配置
# =============================================================================
# 加载 .env 文件中的环境变量
# override=True 表示即使系统已有同名环境变量，也用 .env 文件中的值覆盖
load_dotenv(override=True)
# 工作目录：当前脚本运行的位置
# 所有文件操作都会相对于这个目录进行
WORKDIR = Path.cwd()
# 初始化 OpenAI 客户端
# 注意：虽然变量名叫 ANTHROPIC_API_KEY，但这里实际是 DashScope 的 API Key
# 这种命名是为了方便切换不同的后端服务
client = OpenAI(
api_key=os.getenv("ANTHROPIC_API_KEY"), # 从环境变量获取 API 密钥
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", # DashScope 兼容接口地址
)
# 模型配置：从环境变量获取，默认使用 qwen-plus
# 可选值包括：qwen-turbo, qwen-plus, qwen-max 等
MODEL = os.getenv("MODEL_ID", "qwen-plus")
# System Prompt: 定义 Agent 的角色和行为准则
# 这里告诉模型它是一个编码代理，应该使用工具来完成任务，并且行动优于解释
SYSTEM = f"You are a coding agent at {WORKDIR}. Use tools to solve tasks. Act, don't explain."
# =============================================================================
# 安全工具函数
# =============================================================================
def safe_path(p: str) -> Path:
"""
安全路径解析函数 - 防止路径遍历攻击
将用户提供的相对路径转换为绝对路径，并验证该路径是否在允许的工作目录内。
这可以防止恶意用户尝试访问工作目录之外的敏感文件。
参数:
p: 用户提供的文件路径（相对路径）
返回:
Path: 解析后的安全绝对路径对象
异常:
ValueError: 如果路径试图逃逸工作目录
示例:
>>> safe_path("src/main.py") # 正常路径
Path("/home/user/project/src/main.py")
>>> safe_path("../../../etc/passwd") # 恶意路径
ValueError: Path escapes workspace: ../../../etc/passwd
"""
# 将相对路径解析为绝对路径
# WORKDIR / p 会将路径拼接到工作目录
# .resolve() 会解析所有符号链接和相对路径组件（如 ..）
path = (WORKDIR / p).resolve()
# 安全检查：确保解析后的路径仍然是工作目录的子路径
# is_relative_to() 方法检查路径是否在指定目录内
if not path.is_relative_to(WORKDIR):
raise ValueError(f"Path escapes workspace: {p}")
return path
# =============================================================================
# 工具执行函数
# =============================================================================
def run_bash(command: str) -> str:
"""
Bash 命令执行工具
在 shell 中执行指定的命令，并返回输出结果。
包含安全检查，阻止危险命令的执行。
参数:
command: 要执行的 shell 命令字符串
返回:
str: 命令的标准输出和标准错误输出（合并），最多返回 50000 字符
安全措施:
1. 阻止危险命令：rm -rf /, sudo, shutdown, reboot, > /dev/
2. 设置超时限制：120 秒
3. 输出截断：防止返回过多数据
示例:
>>> run_bash("ls -la")
"total 48\ndrwxr-xr-x 2 user user 4096 ..."
>>> run_bash("sudo rm -rf /")
"Error: Dangerous command blocked"
"""
# 危险命令黑名单
# 这些命令可能对系统造成严重损害，因此被完全阻止
dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]
# 检查命令是否包含任何危险关键词
if any(d in command for d in dangerous):
return "Error: Dangerous command blocked"
try:
# 执行 shell 命令
# shell=True: 通过 shell 执行命令（支持管道、通配符等 shell 特性）
# cwd=WORKDIR: 在工作目录中执行命令
# capture_output=True: 捕获 stdout 和 stderr
# text=True: 将输出解码为字符串而非字节
# timeout=120: 设置 120 秒超时限制
r = subprocess.run(
command,
shell=True,
cwd=WORKDIR,
capture_output=True,
text=True,
timeout=120
)
# 合并标准输出和标准错误输出
# .strip() 移除首尾空白字符
out = (r.stdout + r.stderr).strip()
# 截断输出，防止返回过多数据
# 如果输出为空，返回提示信息
return out[:50000] if out else "(no output)"
except subprocess.TimeoutExpired:
# 命令执行超时
return "Error: Timeout (120s)"
def run_read(path: str, limit: int = None) -> str:
"""
文件读取工具
读取指定文件的内容，支持限制读取行数。
包含路径安全检查，防止读取工作目录外的文件。
参数:
path: 文件路径（相对于工作目录）
limit: 可选参数，限制读取的行数。如果文件行数超过限制，
会在末尾添加省略提示
返回:
str: 文件内容，最多返回 50000 字符
示例:
>>> run_read("README.md")
"# Project Name\nThis is a readme file..."
>>> run_read("large_file.txt", limit=10)
"Line 1\nLine 2\n... Line 10\n... (990 more lines)"
"""
try:
# 使用 safe_path 进行安全路径解析
# .read_text() 读取整个文件内容为字符串
text = safe_path(path).read_text()
# 按行分割文本
lines = text.splitlines()
# 如果设置了行数限制且文件行数超过限制
if limit and limit < len(lines):
# 截取指定行数
lines = lines[:limit]
# 添加省略提示，显示剩余行数
lines = lines + [f"... ({len(lines) - limit} more lines)"]
# 重新拼接为字符串，并截断到 50000 字符
return "\n".join(lines)[:50000]
except Exception as e:
# 捕获并返回任何错误（如文件不存在、权限不足等）
return f"Error: {e}"
def run_write(path: str, content: str) -> str:
"""
文件写入工具
将内容写入指定文件。如果文件不存在会创建新文件，
如果文件所在目录不存在也会自动创建目录。
参数:
path: 文件路径（相对于工作目录）
content: 要写入的文件内容
返回:
str: 操作结果信息，包含写入的字节数和文件路径
安全措施:
1. 使用 safe_path 防止路径遍历攻击
2. 自动创建父目录
示例:
>>> run_write("output.txt", "Hello, World!")
"Wrote 13 bytes to output.txt"
"""
try:
# 安全路径解析
fp = safe_path(path)
# 创建父目录（如果不存在）
# parents=True: 创建所有必要的父目录
# exist_ok=True: 如果目录已存在不报错
fp.parent.mkdir(parents=True, exist_ok=True)
# 写入文件内容
fp.write_text(content)
# 返回操作结果
return f"Wrote {len(content)} bytes to {path}"
except Exception as e:
return f"Error: {e}"
def run_edit(path: str, old_text: str, new_text: str) -> str:
"""
文件编辑工具
在文件中查找并替换精确匹配的文本。只会替换第一个匹配项。
参数:
path: 文件路径（相对于工作目录）
old_text: 要查找的原始文本（必须精确匹配）
new_text: 替换后的新文本
返回:
str: 操作结果信息
注意:
- 使用精确字符串匹配，区分大小写
- 只替换第一个匹配项，不会替换所有出现
示例:
>>> run_edit("config.py", "DEBUG = False", "DEBUG = True")
"Edited config.py"
>>> run_edit("config.py", "not_exist", "new_value")
"Error: Text not found in config.py"
"""
try:
# 安全路径解析
fp = safe_path(path)
# 读取当前文件内容
content = fp.read_text()
# 检查要替换的文本是否存在
if old_text not in content:
return f"Error: Text not found in {path}"
# 执行替换操作
# .replace(old, new, 1) 中的 1 表示只替换第一个匹配项
fp.write_text(content.replace(old_text, new_text, 1))
return f"Edited {path}"
except Exception as e:
return f"Error: {e}"
# =============================================================================
# 工具分发映射表
# =============================================================================
# 工具名称到处理函数的映射
# 当 LLM 调用某个工具时，通过这个字典找到对应的处理函数
# 使用 lambda 函数来统一参数传递方式（从字典中提取参数）
#
# 工作原理:
# 1. LLM 返回工具调用请求，包含工具名称和参数
# 2. 通过 TOOL_HANDLERS[tool_name] 获取对应的 lambda 函数
# 3. 调用 lambda 函数，传入参数字典 **kwargs
# 4. lambda 函数将参数分发给实际的执行函数
TOOL_HANDLERS = {
"bash": lambda **kw: run_bash(kw["command"]),
"read_file": lambda **kw: run_read(kw["path"], kw.get("limit")),
"write_file": lambda **kw: run_write(kw["path"], kw["content"]),
"edit_file": lambda **kw: run_edit(kw["path"], kw["old_text"], kw["new_text"]),
}
# =============================================================================
# 工具定义 (OpenAI 格式)
# =============================================================================
# OpenAI 工具定义格式说明:
# - 每个工具是一个字典，包含 "type" 和 "function" 两个键
# - "type" 目前只支持 "function"
# - "function" 包含:
# - name: 工具名称（必须与 TOOL_HANDLERS 中的键一致）
# - description: 工具描述（帮助 LLM 理解工具用途）
# - parameters: JSON Schema 格式的参数定义
TOOLS = [
# -------------------------------------------------------------------------
# bash 工具: 执行 shell 命令
# -------------------------------------------------------------------------
{
"type": "function", # 工具类型，OpenAI 目前只支持 "function"
"function": {
"name": "bash", # 工具名称，必须与 TOOL_HANDLERS 中的键匹配
"description": "Run a shell command.", # 工具描述，LLM 会根据这个决定是否使用
"parameters": { # 参数定义，使用 JSON Schema 格式
"type": "object", # 参数必须是一个对象（字典）
"properties": {
"command": {
"type": "string", # 参数类型为字符串
"description": "The shell command to execute" # 参数描述
}
},
"required": ["command"] # 必需参数列表
}
}
},
# -------------------------------------------------------------------------
# read_file 工具: 读取文件内容
# -------------------------------------------------------------------------
{
"type": "function",
"function": {
"name": "read_file",
"description": "Read file contents.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "The file path to read"
},
"limit": {
"type": "integer", # 整数类型
"description": "Maximum number of lines to read"
# 注意: limit 不在 required 中，是可选参数
}
},
"required": ["path"] # 只有 path 是必需的
}
}
},
# -------------------------------------------------------------------------
# write_file 工具: 写入文件
# -------------------------------------------------------------------------
{
"type": "function",
"function": {
"name": "write_file",
"description": "Write content to file.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "The file path to write"
},
"content": {
"type": "string",
"description": "The content to write"
}
},
"required": ["path", "content"] # 两个参数都是必需的
}
}
},
# -------------------------------------------------------------------------
# edit_file 工具: 编辑文件（查找替换）
# -------------------------------------------------------------------------
{
"type": "function",
"function": {
"name": "edit_file",
"description": "Replace exact text in file.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "The file path to edit"
},
"old_text": {
"type": "string",
"description": "The text to replace"
},
"new_text": {
"type": "string",
"description": "The new text"
}
},
"required": ["path", "old_text", "new_text"] # 三个参数都是必需的
}
}
},
]
# =============================================================================
# Agent 循环核心函数
# =============================================================================
def agent_loop(messages: list):
"""
Agent 主循环函数
这是整个系统的核心，实现了 LLM 与工具之间的交互循环：
1. 将对话历史发送给 LLM
2. 检查 LLM 是否需要调用工具
3. 如果需要，执行工具并将结果返回给 LLM
4. 重复直到 LLM 返回最终回复
参数:
messages: 对话历史列表，包含 user 和 assistant 的消息
工作流程:
┌─────────────────────────────────────┐
│ 发送消息给 LLM (包含工具定义) │
└─────────────────┬───────────────────┘
▼
┌─────────────────────────────────────┐
│ LLM 返回响应 │
└─────────────────┬───────────────────┘
▼
┌─────────────────┐
│ 有工具调用? │
└────────┬────────┘
┌────┴────┐
是 否
│ │
▼ ▼
┌──────────────┐ ┌──────────────┐
│ 执行所有工具 │ │ 返回最终回复 │
│ 返回结果给LLM │ │ │
└──────┬───────┘ └──────────────┘
│
└──────► 返回步骤 1 (循环)
OpenAI 特有细节:
- tool_choice="auto": 让模型自动决定是否使用工具
- tool_calls: 包含所有工具调用的列表
- tool_call_id: 每个工具调用都有唯一 ID，结果需要关联
"""
while True:
# =====================================================================
# 步骤 1: 调用 OpenAI API
# =====================================================================
response = client.chat.completions.create(
model=MODEL, # 使用的模型
# 构建消息列表: system prompt + 对话历史
# 注意: OpenAI 将 system 作为 messages 列表的一部分
messages=[{"role": "system", "content": SYSTEM}] + messages,
tools=TOOLS, # 工具定义列表
tool_choice="auto", # 让模型自动决定是否使用工具
max_tokens=8000, # 最大生成 token 数
)
# =====================================================================
# 步骤 2: 获取助手消息并添加到历史
# =====================================================================
# response.choices[0].message 是助手的回复消息对象
assistant_message = response.choices[0].message
# ⚠️ 重要: 必须将 Pydantic 模型对象转换为字典格式
# 如果直接添加 assistant_message 对象，当包含 tool_calls 时，
# 下次 API 调用会报错: TypeError: argument 'by_alias': 'NoneType' object
# 使用 model_dump(exclude_none=True) 序列化为字典
messages.append(assistant_message.model_dump(exclude_none=True))
logger.info(f"当前回应大模型的数据消息: {assistant_message.model_dump(exclude_none=True)}")
# =====================================================================
# 步骤 3: 检查是否需要调用工具
# =====================================================================
# tool_calls 属性为 None 或空列表表示模型不需要调用工具
# 这意味着模型已经生成了最终回复
if not assistant_message.tool_calls:
# 没有工具调用，打印最终回复并返回
if assistant_message.content:
print(assistant_message.content)
return # 退出循环，返回到调用者
# =====================================================================
# 步骤 4: 处理所有工具调用
# =====================================================================
tool_results = [] # 存储所有工具执行结果
# 遍历每个工具调用请求
# 注意: 模型可能一次请求调用多个工具（并行调用）
for tool_call in assistant_message.tool_calls:
# -----------------------------------------------------------------
# 提取工具信息
# -----------------------------------------------------------------
# tool_call.function.name: 工具名称
tool_name = tool_call.function.name
# tool_call.function.arguments: 工具参数（JSON 字符串）
# 需要用 json.loads() 解析为 Python 字典
tool_args = json.loads(tool_call.function.arguments)
# -----------------------------------------------------------------
# 执行工具
# -----------------------------------------------------------------
# 从分发映射表中获取对应的处理函数
handler = TOOL_HANDLERS.get(tool_name)
logger.info(f"工具调用请求: {tool_name} with args {tool_args}")
if handler:
# 调用处理函数，传入参数
# **tool_args 将字典解包为关键字参数
output = handler(**tool_args)
else:
# 未知工具，返回错误信息
output = f"Unknown tool: {tool_name}"
# 打印工具执行结果（截取前 200 字符）
print(f"> {tool_name}: {output[:200]}")
# -----------------------------------------------------------------
# 构建工具结果消息
# -----------------------------------------------------------------
# OpenAI 要求工具结果包含以下字段:
# - tool_call_id: 关联回原始工具调用的唯一 ID
# - role: 必须是 "tool"
# - name: 工具名称（可选但推荐）
# - content: 工具执行的输出结果
tool_results.append({
"tool_call_id": tool_call.id, # 关联 ID，必须与 tool_call.id 匹配
"role": "tool", # 固定为 "tool"
"name": tool_name, # 工具名称
"content": output # 工具输出
})
# =====================================================================
# 步骤 5: 将工具结果添加到对话历史
# =====================================================================
# 将所有工具结果追加到消息列表
# 下一次循环时，这些结果会随对话历史一起发送给 LLM
messages.extend(tool_results)
# 循环继续，返回步骤 1...
# =============================================================================
# 主程序入口
# =============================================================================
if __name__ == "__main__":
# 初始化对话历史列表
# 用于保存完整的对话上下文
history = []
# 主交互循环
while True:
try:
# 获取用户输入
# \033[36m 是 ANSI 颜色代码，显示青色
# \033[0m 重置颜色
query = input("\033[36ms02 >> \033[0m")
except (EOFError, KeyboardInterrupt):
# EOFError: 用户按了 Ctrl+D (Unix) 或 Ctrl+Z (Windows)
# KeyboardInterrupt: 用户按了 Ctrl+C
break
# 检查退出命令
# strip() 移除首尾空白
# lower() 转换为小写
if query.strip().lower() in ("q", "exit", ""):
break
# 将用户消息添加到历史
history.append({"role": "user", "content": query})
# 调用 agent 循环处理请求
# agent_loop 可能会执行多轮工具调用
agent_loop(history)
# 打印空行分隔不同的交互
print()
1.工具的解剖学：不仅仅是API
在传统软件中，调用一个函数只需要了解其签名（Signature），即参数类型与返回值。然而，在 Agent 架构中，工具必须附带语义（Semantics）。Agent 不需要知晓内存地址或底层实现细节，但它必须清楚“这个工具能用来做什么”以及“在什么情境下应当
使用它”。
为 Agent 设计的工具，通常包含以下3个维度的描述。
■功能描述：这是写给模型看的“说明书”，用于说明工具的能力和适用场景，帮
助模型判断是否应选择该工具。
■参数约束：这是写给程序的“契约”，明确规定调用时所需参数的结构与类型，
模型必须生成符合该规范的JSON输入。
■副作用声明：这是写给安全模块的“警示”，用于标明操作的潜在影响。
2.从“工具”到“技能”：企业的数字本体
在推出 MCP 之后，Anthropic 于 2025 年 10 月发布了 Agent Skills，再次在
Agent 工程领域产生了重要影响。
■工具（TooI）是通用且原子化的操作，如“执行SQL操作”“发送 HTTP 请求”
“读写文件”等。
■技能（Skill）是面向业务场景、经过封装的能力单元，如“生成月度财报”“执
行代码审查”“处理客户退款”等。
简言之，技能=工具＋领域知识+标准作业程序。
为什么这一区别如此重要？因为单纯的API是冰冷且缺少上下文的。如果直接向
Agent 提供一个 delete_user 的 APl，不仅存在危险，还可能引发不可逆的操作失误。然而，如果将该 API封装为一项“用户注销技能”，并在其中内嵌标准作业程序（例如，备份数据→检查欠款→发送通知一删除操作），那么Agent获得的不只是一个原始功能，而是一种具备业务语义和安全约束的业务能力。
技能包构成了企业的数字本体。
■当企业将内部流程、规范和最佳实践封装为Agent可调用的“技能”时，便实现
了知识的资产化。
■Agent 不再只是一个通用模型，而是继承了企业 DNA 的“数字员工”：它不仅知
道“怎么做”（调用API），更懂得“如何正确地做”（遵循标准作业程序）。
3.工具和技能的选择
传统工具调用是机械式的：只要用户说“翻译”，系统就直接调用翻译工具，不加判
断、不问上下文。
而在 Agent 时代，工具和技能的选择更像一位资深工程师的决策过程：是否需要调用工具？现在是不是恰当的调用时机？该选用哪个工具？当前上下文是否充分？是否存在潜在副作用？是否需要先验证输入？例如，当用户说“把昨天的日志分析一下”时，Agent 不会盲目调用 analyse_logs,而是会主动澄清意图、确认上下文。例如，Agent会问：“昨天的日志存储在哪个目录？”“你希望分析错误趋势、性能瓶颈，还是生成汇总报告？”“日志中是否包含敏感信息？是否需要脱敏处理？”Agent 能够在上百个可用工具和多种技能中，精准识别出最适合当前任务的那一个；随后，它会从用户对话中提取关键信息，结合该工具预定义的参数约束，生成结构合规的输入；最后，以这些参数调用工具并执行操作。
优秀的工具和技能选择能力不仅使Agent的行为更加可靠、可控、可解释，同时还支撑了多工具协同（Multi-tool Orchestration）——Agent 能够自动规划调用I顺序，例如，先使用搜索工具查找相关文档，再调用代码执行工具运行代码示例，最后借助数学工具计算并验证结果。这种动态、上下文感知的工具编排能力，正是LangGraph、AutoGen和MCP等生态当前重点发展的方向。
4.核心循环：ReAct
工具调用并非孤立的事件，而是推理链条中的关键一环。Agent会根据工具的返回结果（成功、失败或报错）动态调整后续行动。
DeepMind 提出的 ReAct 范式已成为现代 Agent 行动的标准心智模型，其核心循环
如下。
思考（Thought）一行动（Action）一观察（Observation）
这改变了传统“输入一输出”的线性逻辑，形成了一个动态闭环的推理与行动循环。
■思考：用户让我分析销售下降的原因。我应该先查一下销售数据。
■行动： 调用 query_sales_db(“SELECT ＊ FROM sales WHERE date >
'2024-01-01”)。
■观察：（数据库返回了1000 条原始JSON记录）数据量太大，难以直接分析。
■思考：我犯了个错误，不应拉取明细数据，而应先按月聚合，观察整体趋势。
■行动：调用 query_sales_db(“SELECT month, SUM(amount)…GROUP
BY month”)。
这种“行动增强推理"（Action-Augmented Reasoning）使 Agent 具备了试错和
自我修正的能力。它不再依赖一次性给出完美答案，而是通过与环境的持续交互，逐步逼近正确的解决方案。
5.行动策略：安全与边界
当软件开始“行动”时，风险也随之而来。一个由LLM 驱动、却拥有 rm一f 权限的
Agent，无异于一只在服务器前随意敲击键盘的猴子——其意图或许合理，但后果可能是灾难性的。
行动策略（ActionStrategy）决定了Agent 是鲁莽地直接执行任务，还是谨慎行事
并先由人类确认。其核心不在于如何执行任务，而在于何时暂停以进行验证或获取批准。
为此，架构师必须设计以下3道防线。
■只读沙箱（Read-Only Sandbox）：在探索与信息收集阶段，仅允许 Agent 调
用无副作用的工具（如搜索、读取日志、查询数据库）。这相当于为 Agent 设置
了一道“婴儿围栏”，确保其在理解任务和制订计划时不会对系统造成任何不可
逆影响。
■人类介入（Human-In-The-Loop，HITL）：在执行高风险写操作（如转账、
删除文件、发送外部邮件）前，系统强制中断流程，清晰地展示拟执行的操作内
容、上下文及潜在后果，等待人类明确批准后方可继续。
■确定性护栏（DeterministicGuardrails）：即使获得用户授权，底层执行层仍
需要嵌入硬性规则校验（如“禁止删除根目录”）。
行动策略是Agent的“行为规范”，包含以下3个核心层次。
■安全：在执行任何操作前，Agent必须具备“暂停判断”的能力，主动识别高风
险行为（如删除系统文件、修改生产数据库等），并拒绝执行超出安全边界的危
险操作。这是所有行业级 Agent必须坚守的底线。
■可解释：用户必须清晰地理解 Agent“为什么这么做”。Agent 应主动展示其计
划、说明工具调用的原因，并报告每一步的执行结果。这不仅有助于建立用户信
任，也为审计、追踪与调试提供了必要依据。
■可控：在执行关键或不可逆步骤时，Agent 应支持用户介入。例如，提前展示即
将执行的操作，明确询问“是否继续”，确保人类始终保有最终决策权。
工具使用赋予 Agent 影响现实世界的能力，技能为其提供职业素养，而行动策略则确保其行为是安全、高效且可信的。
三者共同构成了Agent从“知识系统”迈向“执行系统”的最后一块关键拼图。
这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容，并连接云服务、办公空间等稀缺资源，助你专注创造，无忧运营。
更多推荐
7
0- 0
已为社区贡献1条内容
所有评论(0)