手写Agent（简短Agent+hello-agent的第一章-firstAgent）
·
简短代码实现的Agent
可以使用本地命令进行创建等操作
from openai import OpenAI
import os
# ============ Dashscope兼容模式初始化，只实例一次client ============
client = OpenAI(
#api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope/compatible-mode/v1"
)
messages = [
{
"role": "system",
"content": """
你的目标是完成用户的任务，你必须选择下面的其中一种格式进行回复：
1. 如果你认为需要执行命令，则输出'命令：XXX'，XXX 为命令本身，不要用任何格式，不要解释
2. 如果你认为不需要执行命令，则输出'完成：XXX'，XXX 为你的总结信息
"""
}
]
#外层循环，接收用户输入
while True:
user_input = input("\n【你】")
messages.append({"role": "user", "content": user_input})
print("\n-------- Agent 循环开始 --------")
#Agent内层思考循环
while True:
# 重点！Dashscope兼容模式使用 create，不是 send
response = client.chat.completions.create(
model="qwen-plus", #更换模型名即可切换
messages=messages
)
reply = response.choices[0].message.content
messages.append({"role": "assistant", "content": reply})
print(f"【AI】{reply}")
reply_stripped = reply.strip()
if reply_stripped.startswith("完成："):
print("\n-------- Agent 循环结束 --------")
final_text = reply_stripped.split("完成：")[1].strip()
print(f"【AI】{final_text}")
break
try:
command = reply_stripped.split("命令：")[1].strip()
command_result = os.popen(command).read()
except Exception as e:
print(f"[Agent执行出错] {e}")
break
content = f"执行完毕 {command_result}"
print(f"【Agent】{content}")
messages.append({"role": "user", "content": content})
接入skill.md
from openai import OpenAI
import os
# ============ Dashscope兼容模式初始化，只实例一次client ============
client = OpenAI(
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url=""
)
agentmd=open("Easy-Agent\Agent.md","r",encoding="utf-8").read()
skillmd=open("Easy-Agent\Skill.md","r",encoding="utf-8").read()
messages = [{"role": "system", "content": agentmd+skillmd}]
#外层循环，接收用户输入
while True:
user_input = input("\n【你】")
messages.append({"role": "user", "content": user_input})
print("\n-------- Agent 循环开始 --------")
#Agent内层思考循环
while True:
# 重点！Dashscope兼容模式使用 create，不是 send
response = client.chat.completions.create(
model="qwen-plus", #更换模型名即可切换
messages=messages
)
reply = response.choices[0].message.content
messages.append({"role": "assistant", "content": reply})
print(f"【AI】{reply}")
reply_stripped = reply.strip()
if reply_stripped.startswith("完成："):# 如果大模型的回复以"完成："开头，表示任务已经完成，跳出循环
print("\n-------- Agent 循环结束 --------")
final_text = reply_stripped.split("完成：")[1].strip()
print(f"【AI】{final_text}")
break
try:
command = reply_stripped.split("命令：")[1].strip()# 如果大模型的回复中包含"命令："，则提取命令并执行
command_result = os.popen(command).read()
except Exception as e:
print(f"[Agent执行出错] {e}")
break
content = f"执行完毕 {command_result}"
print(f"【Agent】{content}")
messages.append({"role": "user", "content": content})
Agent.md
你的目标是完成用户的任务，你必须选择下面的其中一种格式进行回复：
1. 如果你认为需要执行命令，则输出'命令：XXX'，XXX 为命令本身，不要用任何的格式，不要解释，每次只进行一条命令，不要多条放一起
2. 如果你认为不需要执行命令，则输出'完成：XXX'，XXX 为你的总结信息
我们的沟通过程是个循环，你每次回复一条命令后，要等我给你返回命令执行的结果，然后你再继续回复。
比如：
用户：创建一个 hello.txt 和 helloworld.txt 文件
AI：命令：echo "hello" > hello.txt
用户：执行成功
AI：命令：echo "hello world" > helloworld.txt
用户：执行成功
AI：完成：我已经完成了用户的需求，hello.txt 和 helloworld.txt 文件已经创建成功
Skill.md
如果需要获取新闻，使用如下命令，XXX为用户希望搜索的关键词
curl -L -A "Mozilla/5.0" "http://news.baidu.com/n?cmd=7&class=news&rn=20&word=XXX"
效果：
【AI】你好！有什么我可以帮你的吗？
【你】创建hello.txt，内容是 hello world
-------- Agent 循环开始 --------
【AI】命令：echo “hello world” > hello.txt
【Agent】执行完毕
【AI】完成：文件 hello.txt 已成功创建，内容为 “hello world”。
-------- Agent 循环结束 --------
【AI】文件 hello.txt 已成功创建，内容为 “hello world”。
hello-agent第一个Agent
目标是构建一个能处理分步任务的智能旅行助手。需要解决的用户任务定义为："你好，请帮我查询一下今天北京的天气，然后根据天气推荐一个合适的旅游景点。"要完成这个任务，智能体必须展现出清晰的逻辑规划能力。它需要先调用天气查询工具，并将获得的观察结果作为下一步的依据。在下一轮循环中，它再调用景点推荐工具，从而得出最终建议。
（1）指令模板
驱动真实 LLM 的关键在于提示工程（Prompt Engineering）。我们需要设计一个“指令模板”，告诉 LLM 它应该扮演什么角色、拥有哪些工具、以及如何格式化它的思考和行动。这是我们智能体的“说明书”，它将作为system_prompt传递给 LLM。
AGENT_SYSTEM_PROMPT = """
你是一个智能旅行助手。你的任务是分析用户的请求，并使用可用工具一步步地解决问题。
# 可用工具:
- `get_weather(city: str)`: 查询指定城市的实时天气。
- `get_attraction(city: str, weather: str)`: 根据城市和天气搜索推荐的旅游景点。
# 输出格式要求:
你的每次回复必须严格遵循以下格式，包含一对Thought和Action：
Thought: [你的思考过程和下一步计划]
Action: [你要执行的具体行动]
Action的格式必须是以下之一：
1. 调用工具：function_name(arg_name="arg_value")
2. 结束任务：Finish[最终答案]
# 重要提示:
- 每次只输出一对Thought-Action
- Action必须在同一行，不要换行
- 当收集到足够信息可以回答用户问题时，必须使用 Action: Finish[最终答案] 格式结束
请开始吧！
"""
（2）工具 1：查询真实天气
使用免费的天气查询服务 wttr.in，它能以 JSON 格式返回指定城市的天气数据。
格式化成自然语言返回 return f"{city}当前天气:{weather_desc}，气温{temp_c}摄氏度"
import requests
def get_weather(city: str) -> str:
"""
通过调用 wttr.in API 查询真实的天气信息。
"""
# API端点，我们请求JSON格式的数据
url = f"https://wttr.in/{city}?format=j1"
try:
# 发起网络请求
response = requests.get(url)
# 检查响应状态码是否为200 (成功)
response.raise_for_status()
# 解析返回的JSON数据
data = response.json()
# 提取当前天气状况
current_condition = data['current_condition'][0]
weather_desc = current_condition['weatherDesc'][0]['value']
temp_c = current_condition['temp_C']
# 格式化成自然语言返回
return f"{city}当前天气:{weather_desc}，气温{temp_c}摄氏度"
except requests.exceptions.RequestException as e:
# 处理网络错误
return f"错误:查询天气时遇到网络问题 - {e}"
except (KeyError, IndexError) as e:
# 处理数据解析错误
return f"错误:解析天气数据失败，可能是城市名称无效 - {e}"
（3）工具 2：搜索并推荐旅游景点
根据城市和天气状况，在互联网上搜索合适的景点
（amazing！还有这种API!?？）
import os
from tavily import TavilyClient
def get_attraction(city: str, weather: str) -> str:
"""
根据城市和天气，使用Tavily Search API搜索并返回优化后的景点推荐。
"""
# 1. 从环境变量中读取API密钥
api_key = os.environ.get("TAVILY_API_KEY")
if not api_key:
return "错误:未配置TAVILY_API_KEY环境变量。"
# 2. 初始化Tavily客户端
tavily = TavilyClient(api_key=api_key)
# 3. 构造一个精确的查询
query = f"'{city}' 在'{weather}'天气下最值得去的旅游景点推荐及理由"
try:
# 4. 调用API，include_answer=True会返回一个综合性的回答
response = tavily.search(query=query, search_depth="basic", include_answer=True)
# 5. Tavily返回的结果已经非常干净，可以直接使用
# response['answer'] 是一个基于所有搜索结果的总结性回答
if response.get("answer"):
return response["answer"]
# 如果没有综合性回答，则格式化原始结果
formatted_results = []
for result in response.get("results", []):
formatted_results.append(f"- {result['title']}: {result['content']}")
if not formatted_results:
return "抱歉，没有找到相关的旅游景点推荐。"
return "根据搜索，为您找到以下信息:\n" + "\n".join(formatted_results)
except Exception as e:
return f"错误:执行Tavily搜索时出现问题 - {e}"
（4）执行行动循环
核心循环流程：
用户请求 → 拼接历史上下文 → LLM思考输出Thought+Action → 正则解析Action → 执行工具函数 → 得到Observation结果写入历史 → 再次交给LLM思考，最多循环5次，遇到Finish工具结束任务
以下输出完整地展示了一个成功的智能体执行流程。通过对这个三轮循环的分析，我们可以清晰地看到智能体解决问题的核心能力。
用户输入: 你好，请帮我查询一下今天北京的天气，然后根据天气推荐一个合适的旅游景点。
========================================
--- 循环 1 ---
正在调用大语言模型...
大语言模型响应成功。
模型输出:
Thought: 首先需要获取北京今天的天气情况，之后再根据天气情况来推荐旅游景点。
Action: get_weather(city="北京")
Observation: 北京当前天气:Sunny，气温26摄氏度
========================================
--- 循环 2 ---
# 把【用户请求 + 上一轮 Thought‑Action+Observation】全部输入模型
正在调用大语言模型...
大语言模型响应成功。
模型输出:
Thought: 现在已经知道了北京今天的天气是晴朗且温度适中，接下来可以基于这个信息来推荐一个适合的旅游景点了。
Action: get_attraction(city="北京", weather="Sunny")
Observation: 北京在晴天最值得去的旅游景点是颐和园，因其美丽的湖景和古建筑。另一个推荐是长城，因其壮观的景观和历史意义。
========================================
--- 循环 3 ---
正在调用大语言模型...
大语言模型响应成功。
模型输出:
Thought: 已经获得了两个适合晴天游览的景点建议，现在可以根据这些信息给用户提供满意的答复。
Action: Finish[今天北京的天气是晴朗的，气温26摄氏度，非常适合外出游玩。我推荐您去颐和园欣赏美丽的湖景和古建筑，或者前往长城体验其壮观的景观和深厚的历史意义。希望您有一个愉快的旅行！]
任务完成，最终答案: 今天北京的天气是晴朗的，气温26摄氏度，非常适合外出游玩。我推荐您去颐和园欣赏美丽的湖景和古建筑，或者前往长城体验其壮观的景观和深厚的历史意义。希望您有一个愉快的旅行！
import re
# --- 1. 配置LLM客户端 ---
# 请根据您使用的服务，将这里替换成对应的凭证和地址
API_KEY = "YOUR_API_KEY"
BASE_URL = "YOUR_BASE_URL"
MODEL_ID = "YOUR_MODEL_ID"
TAVILY_API_KEY="YOUR_Tavily_KEY"
os.environ['TAVILY_API_KEY'] = "YOUR_TAVILY_API_KEY"
llm = OpenAICompatibleClient(
model=MODEL_ID,
api_key=API_KEY,
base_url=BASE_URL
)
# --- 2. 初始化 ---
user_prompt = "你好，请帮我查询一下今天北京的天气，然后根据天气推荐一个合适的旅游景点。"
prompt_history = [f"用户请求: {user_prompt}"]
print(f"用户输入: {user_prompt}\n" + "="*40)
# --- 3. 运行主循环 ---
for i in range(5): # 设置最大循环次数
print(f"--- 循环 {i+1} ---\n")
# 3.1. 构建Prompt
full_prompt = "\n".join(prompt_history)
# 3.2. 调用LLM进行思考
llm_output = llm.generate(full_prompt, system_prompt=AGENT_SYSTEM_PROMPT)
# 模型可能会输出多余的Thought-Action，需要截断
match = re.search(r'(Thought:.*?Action:.*?)(?=\n\s*(?:Thought:|Action:|Observation:)|\Z)', llm_output, re.DOTALL)
if match:
truncated = match.group(1).strip()
if truncated != llm_output.strip():
llm_output = truncated
print("已截断多余的 Thought-Action 对")
print(f"模型输出:\n{llm_output}\n")
prompt_history.append(llm_output)
# 3.3. 解析并执行行动
action_match = re.search(r"Action: (.*)", llm_output, re.DOTALL)
if not action_match:
observation = "错误: 未能解析到 Action 字段。请确保你的回复严格遵循 'Thought: ... Action: ...' 的格式。"
observation_str = f"Observation: {observation}"
print(f"{observation_str}\n" + "="*40)
prompt_history.append(observation_str)
continue
action_str = action_match.group(1).strip()
if action_str.startswith("Finish"):
final_answer = re.match(r"Finish\[(.*)\]", action_str).group(1)
print(f"任务完成，最终答案: {final_answer}")
break
tool_name = re.search(r"(\w+)\(", action_str).group(1)
args_str = re.search(r"\((.*)\)", action_str).group(1)
kwargs = dict(re.findall(r'(\w+)="([^"]*)"', args_str))
if tool_name in available_tools:
observation = available_tools[tool_name](**kwargs)
else:
observation = f"错误:未定义的工具 '{tool_name}'"
# 3.4. 记录观察结果
observation_str = f"Observation: {observation}"
print(f"{observation_str}\n" + "="*40)
prompt_history.append(observation_str)
完整代码：（我没跑通因为’tavily’这部分没弄还需要key啥的）
AGENT_SYSTEM_PROMPT = """
你是一个智能旅行助手。你的任务是分析用户的请求，并使用可用工具一步步地解决问题。
# 可用工具:
- `get_weather(city: str)`: 查询指定城市的实时天气。
- `get_attraction(city: str, weather: str)`: 根据城市和天气搜索推荐的旅游景点。
# 输出格式要求:
你的每次回复必须严格遵循以下格式，包含一对Thought和Action：
Thought: [你的思考过程和下一步计划]
Action: [你要执行的具体行动]
Action的格式必须是以下之一：
1. 调用工具：function_name(arg_name="arg_value")
2. 结束任务：Finish[最终答案]
# 重要提示:
- 每次只输出一对Thought-Action
- Action必须在同一行，不要换行
- 当收集到足够信息可以回答用户问题时，必须使用 Action: Finish[最终答案] 格式结束
请开始吧！
"""
import requests
def get_weather(city: str) -> str:
"""
通过调用 wttr.in API 查询真实的天气信息。
"""
# API端点，我们请求JSON格式的数据
url = f"https://wttr.in/{city}?format=j1"
try:
# 发起网络请求
response = requests.get(url)
# 检查响应状态码是否为200 (成功)
response.raise_for_status()
# 解析返回的JSON数据
data = response.json()
# 提取当前天气状况
current_condition = data['current_condition'][0]
weather_desc = current_condition['weatherDesc'][0]['value']
temp_c = current_condition['temp_C']
# 格式化成自然语言返回
return f"{city}当前天气：{weather_desc}，气温{temp_c}摄氏度"
except requests.exceptions.RequestException as e:
# 处理网络错误
return f"错误：查询天气时遇到网络问题 - {e}"
except (KeyError, IndexError) as e:
# 处理数据解析错误
return f"错误：解析天气数据失败，可能是城市名称无效 - {e}"
import os
from tavily import TavilyClient
def get_attraction(city: str, weather: str) -> str:
"""
根据城市和天气，使用Tavily Search API搜索并返回优化后的景点推荐。
"""
# 从环境变量或主程序配置中获取API密钥
api_key = os.environ.get("TAVILY_API_KEY") # 推荐方式
# 或者，我们可以在主循环中传入，如此处代码所示
if not api_key:
return "错误：未配置TAVILY_API_KEY。"
# 2. 初始化Tavily客户端
tavily = TavilyClient(api_key=api_key)
# 3. 构造一个精确的查询
query = f"'{city}' 在'{weather}'天气下最值得去的旅游景点推荐及理由"
try:
# 4. 调用API，include_answer=True会返回一个综合性的回答
response = tavily.search(query=query, search_depth="basic", include_answer=True)
# 5. Tavily返回的结果已经非常干净，可以直接使用
# response['answer'] 是一个基于所有搜索结果的总结性回答
if response.get("answer"):
return response["answer"]
# 如果没有综合性回答，则格式化原始结果
formatted_results = []
for result in response.get("results", []):
formatted_results.append(f"- {result['title']}: {result['content']}")
if not formatted_results:
return "抱歉，没有找到相关的旅游景点推荐。"
return "根据搜索，为您找到以下信息：\n" + "\n".join(formatted_results)
except Exception as e:
return f"错误：执行Tavily搜索时出现问题 - {e}"
# 将所有工具函数放入一个字典，方便后续调用
available_tools = {
"get_weather": get_weather,
"get_attraction": get_attraction,
}
from openai import OpenAI
class OpenAICompatibleClient:
"""
一个用于调用任何兼容OpenAI接口的LLM服务的客户端。
"""
def __init__(self, model: str, api_key: str, base_url: str):
self.model = model
self.client = OpenAI(api_key=api_key, base_url=base_url)
def generate(self, prompt: str, system_prompt: str) -> str:
"""调用LLM API来生成回应。"""
print("正在调用大语言模型...")
try:
messages = [
{'role': 'system', 'content': system_prompt},
{'role': 'user', 'content': prompt}
]
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
stream=False
)
answer = response.choices[0].message.content
print("大语言模型响应成功。")
return answer
except Exception as e:
print(f"调用LLM API时发生错误: {e}")
return "错误：调用语言模型服务时出错。"
import re
# --- 1. 配置LLM客户端 ---
# 请根据您使用的服务，将这里替换成对应的凭证和地址
API_KEY = "YOUR_API_KEY"
BASE_URL = "YOUR_BASE_URL"
MODEL_ID = "YOUR_MODEL_ID"
os.environ['TAVILY_API_KEY'] = "YOUR_TAVILY_API_KEY"
llm = OpenAICompatibleClient(
model=MODEL_ID,
api_key=API_KEY,
base_url=BASE_URL
)
# --- 2. 初始化 ---
user_prompt = "你好，请帮我查询一下今天北京的天气，然后根据天气推荐一个合适的旅游景点。"
prompt_history = [f"用户请求: {user_prompt}"]
print(f"用户输入: {user_prompt}\n" + "="*40)
# --- 3. 运行主循环 ---
for i in range(5): # 设置最大循环次数
print(f"--- 循环 {i+1} ---\n")
# 3.1. 构建Prompt
full_prompt = "\n".join(prompt_history)
# 3.2. 调用LLM进行思考
llm_output = llm.generate(full_prompt, system_prompt=AGENT_SYSTEM_PROMPT)
# 模型可能会输出多余的Thought-Action，需要截断
match = re.search(r'(Thought:.*?Action:.*?)(?=\n\s*(?:Thought:|Action:|Observation:)|\Z)', llm_output, re.DOTALL)
if match:
truncated = match.group(1).strip()
if truncated != llm_output.strip():
llm_output = truncated
print("已截断多余的 Thought-Action 对")
print(f"模型输出:\n{llm_output}\n")
prompt_history.append(llm_output)# 添加到历史记录中
# 3.3. 解析并执行行动
action_match = re.search(r"Action: (.*)", llm_output, re.DOTALL)
if not action_match:
observation = "错误: 未能解析到 Action 字段。请确保你的回复严格遵循 'Thought: ... Action: ...' 的格式。"
observation_str = f"Observation: {observation}"
print(f"{observation_str}\n" + "="*40)
prompt_history.append(observation_str)
continue
action_str = action_match.group(1).strip()
if action_str.startswith("Finish"):
final_answer = re.match(r"Finish\[(.*)\]", action_str).group(1)
print(f"任务完成，最终答案: {final_answer}")
break
tool_name = re.search(r"(\w+)\(", action_str).group(1)
args_str = re.search(r"\((.*)\)", action_str).group(1)
kwargs = dict(re.findall(r'(\w+)="([^"]*)"', args_str))
if tool_name in available_tools:
observation = available_tools[tool_name](**kwargs)
else:
observation = f"错误：未定义的工具 '{tool_name}'"
# 3.4. 记录观察结果
observation_str = f"Observation: {observation}"
print(f"{observation_str}\n" + "="*40)
prompt_history.append(observation_str)
智能体应用的协作模式
基于智能体在任务中的角色和自主性程度，其协作模式主要分为两种：一种是作为高效工具，深度融入我们的工作流；另一种则是作为自主的协作者，与其他智能体协作完成复杂目标。
1.作为开发者工具的智能体
直接举例比如市面上热门的Claude Code为开发者提供了强大的命令行编程体验，GitHubCopilot 深度集成于 Visual Studio Code 等主流编辑器中，以其强大的代码自动补全能力而闻名，Trae 专注于为开发者提供智能化的代码生成和优化服务，
2.作为自主协作者的智能体
在这种模式下，我们不再是手把手地指导 AI 完成每一步，而是将一个高层级的目标委托给它。智能体会像一个真正的项目成员一样，独立地进行规划、推理、执行和反思，直到最终交付成果。
3.Workflow 和 Agent 的差异
Workflow 是让 AI 按部就班地执行指令，而 Agent 则是赋予 AI 自由度去自主达成目标
工作流是一种传统的自动化范式，其核心是对一系列任务或步骤进行预先定义的、结构化的编排。它本质上是一个精确的、静态的流程图，规定了在何种条件下、以何种顺序执行哪些操作。整个过程的每一步、每一个判断条件都被精确地预先设定
智能体是一个具备自主性的、以目标为导向的系统。在这个过程中，没有任何写死的if天气=晴天 then 推荐颐和园的规则。基于实时信息进行动态推理和决策的能力，正是 Agent 的核心价值所在。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
3
0- 0
已为社区贡献2条内容
所有评论(0)