Deep Read
01 简单Agent搭建
目录+
from pathlib import Path
import os
from dotenv import load_dotenv
from openai import OpenAI
# 获取.env 配置文件
env_path = Path(__file__).resolve().parent.parent / ".env"
load_dotenv(env_path, override=True)
MODEL_ID = os.getenv("MODEL_ID", "deepseek-chat")
# 调用OpenAI 的接口。获取LLM 的客户端
client = OpenAI(
api_key=os.getenv("DEEPSEEK_API"),
base_url=os.getenv("DEEPSEEK_BASE_URL") or None,
)
# 系统提示词
SYSTEM_PROMPT = "You are a helpful AI assistant. Answer questions directly."
def agent_loop() -> None:
messages: list[dict] = []
print(f"Model: {MODEL_ID}")
print("Type 'exit' or 'quit' to leave.")
while True:
try:
# 等待用户输入。
# .strip(): 去掉字符串前后空白字符
user_input = input("请输入>>> ").strip()
# 程序中断 退出
except (KeyboardInterrupt, EOFError):
print("\nExiting...")
break
# 如果用户什么都没输入,就跳过本次循环,重新等待输入。
if not user_input:
continue
# 用户输入转小写,如果是 “exit” 或者 “quit”, 程序退出
if user_input.lower() in {"exit", "quit"}:
print("Goodbye!")
break
# 把用户消息加入聊天历史(messages)中。
messages.append({"role": "user", "content": user_input})
try:
# 调用大模型 API,获取 AI 回复。
response = client.chat.completions.create(
model=MODEL_ID,
max_tokens=2048,
# 发送给模型的聊天上下文, 系统提示词 + 解包的会话历史
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
*messages,
],
)
except Exception as exc:
print(f"Error: {exc}")
messages.pop()
continue
"""
从 API 返回结果中提取:
- AI 回复内容 + 回复结束原因
""" assistant_text = response.choices[0].message.content or ""
finish_reason = response.choices[0].finish_reason
# 如果 AI 不是“正常结束”,就打印结束原因。
if finish_reason and finish_reason != "stop":
print(f"[finish_reason={finish_reason}]")
# 如果 AI 回复内容不为空,就打印出来。
if assistant_text:
print(f"Assistant: {assistant_text}")
# 把 AI 的回复加入聊天历史。
messages.append({"role": "assistant", "content": assistant_text})
# 入口
def main() -> None:
if not os.getenv("DEEPSEEK_API"):
print("Error: DEEPSEEK_API is not set.")
return
agent_loop()
if __name__ == "__main__":
main()
核心讲解:
这是一个最简单的多轮会话Agent,在agent_loop() 是整个Agent 会话的核心,首先定义一个messages: messages: list[dict] = [] 用来存储会话历史,
while True: 多轮会话循环,
user_input = input().strip() 获取用户的输入,进行安全性判断之后将其加入到会话历史messages中
messages.append({"role": "user", "content": user_input})
大模型调用
# 调用大模型 API,获取 AI 回复。
response = client.chat.completions.create(
model=MODEL_ID,
max_tokens=2048,
# 发送给模型的聊天上下文, 系统提示词 + 解包的会话历史
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
*messages,
],
)
- max_tokens: 限制回复的长度
- messages:把系统提示词和会话历史一起发过去
- *messages: 解包操作
最终发送给大模型的就是以下内容:
messages = [
{"role": "system", "content": "你是AI助手"},
{"role": "user", "content": "你好"},
{"role": "assistant", "content": "你好!"},
{"role": "user", "content": "Python是什么"}
]
assistant_text = response.choices[0].message.content or ""
finish_reason = response.choices[0].finish_reason
获取大模型的回复结果和停止回复的原因
常见 finish_reason
1. "stop"
正常结束。
例如:
你好,很高兴帮助你。
模型觉得回答完了。
最常见。
2. "length"
达到 token 上限。
例如:
max_tokens=100
结果回答太长。
模型被强制截断。
3. "content_filter"
内容触发安全过滤。
4. "tool_calls"
模型想调用工具。
Agent 很常见。