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 很常见。