本文介绍如何基于 OpenAI Python SDK 开发 AI Agent 核心功能。内容涵盖模型接入与配置、系统提示词定义、流式输出响应处理，以及工具调用（Function Calling）的完整逻辑闭环。此外，本文还介绍了多轮对话中的记忆管理机制与上下文剪裁策略（包含滑动窗口、Token/字符剪裁以及历史摘要化），为构建稳定可靠的 AI Agent 提供具体实现方式与工程参考。

模型接入

通过 OpenAI SDK 调用大语言模型（LLM）时，配置 base_url 参数可以在不修改业务代码的前提下切换底层模型服务商（例如接入与 OpenAI API 协议兼容的第三方云服务或本地推理服务）。

import os
import openai

# 初始化客户端
# 可接入 OpenAI 原生接口或兼容 OpenAI API 协议的服务商（如阿里云 DashScope）
client = openai.OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_BASE_URL")
)

def get_response(prompt: str) -> str:
    response = client.chat.completions.create(
        model="qwen-turbo",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

系统提示词配置

系统提示词（System Prompt）用于定义智能体的指导原则、角色设定、回答风格与行为约束。规范的 System Prompt 通常包含以下三要素：

• 身份设定（Identity）：指定智能体的角色定位与身份说明。
• 行为规则（Rules）：设定答复限制、禁忌主题及输出格式要求。
• 具体任务（Tasks）：说明智能体需要处理的具体业务目标。

SYSTEM_PROMPT = """你是一个专业的 Python 编程助手。
1. 风格：简洁、高效，多使用代码块。
2. 规则：严禁回答政治、娱乐等非编程相关问题。
3. 身份：你是助教，请勿声明自己是其他公司开发的通用模型。
"""

messages = [
    {"role": "system", "content": SYSTEM_PROMPT},
    {"role": "user", "content": "怎么实现冒泡排序？"}
]

流式输出响应

在生成较长文本时，启用流式输出（Streaming）允许客户端实时接收并处理分片数据（chunks），从而降低首字响应延迟（Time To First Token, TTFT）。

在 Python 控制台中，配合 flush=True 参数可绕过标准输出缓冲区，实现字符级的即时刷新显示。

response = client.chat.completions.create(
    model="qwen-turbo",
    messages=messages,
    stream=True  # 启用流式响应
)

print("Assistant: ", end="", flush=True)
for chunk in response:
    content = chunk.choices[0].delta.content
    if content:
        print(content, end="", flush=True)
print()

工具调用

工具调用（Function Calling）使智能体能够根据用户需求调用外部函数或 API 接口。在 OpenAI SDK 中，工具调用的基本逻辑由“意图识别-本地执行-结果反馈”构成闭环。

智能体循环（Agent Loop）工作流程

智能体处理涉及外部数据查询的请求时，工作流程如下：

1. 意图识别：将用户输入及定义的工具列表（JSON Schema）发送至 LLM。
2. 指令生成：LLM 判断需要调用工具时，返回包含工具名称与参数的 tool_calls 结构。
3. 本地执行：客户端解析 tool_calls 中的函数名与参数，并在本地环境中执行该函数。
4. 结果反馈：将函数执行结果构造为 role: "tool" 的消息添加至对话历史，回传给 LLM。
5. 响应生成：LLM 结合原始问题与工具执行结果，生成最终的答复。

工具闭环示例代码

import json

# 1. 定义本地工具函数
def get_article_content(title: str) -> str:
    """模拟从数据库或文档系统中查询文章内容"""
    return f"【{title}】的文章内容是：AI Agent 开发的核心在于工具调用的闭环设计..."

# 2. 定义工具的 JSON Schema 描述
tools = [{
    "type": "function",
    "function": {
        "name": "get_article_content",
        "description": "根据文章标题获取博客文章的详细全文内容",
        "parameters": {
            "type": "object",
            "properties": {
                "title": {"type": "string", "description": "文章标题"}
            },
            "required": ["title"]
        }
    }
}]

# 3. 智能体单步循环控制逻辑
def run_agent_step(user_query: str) -> str:
    messages = [
        {"role": "system", "content": "你是一个博客助手。"},
        {"role": "user", "content": user_query}
    ]

    # 第一轮：由模型判断是否需要调用工具
    response = client.chat.completions.create(
        model="qwen-turbo",
        messages=messages,
        tools=tools
    )
    
    msg = response.choices[0].message
    messages.append(msg)  # 将包含 tool_calls 的 assistant 消息保存至上下文

    # 检查是否存在工具调用指令
    if msg.tool_calls:
        print(f"[Log] 模型发起工具调用请求，数量: {len(msg.tool_calls)}")
        
        for tool_call in msg.tool_calls:
            # 解析工具名称与参数
            function_name = tool_call.function.name
            args = json.loads(tool_call.function.arguments)
            
            # 执行本地对应函数
            if function_name == "get_article_content":
                result = get_article_content(args["title"])
                
                # 将工具执行结果追加至消息历史
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,  # 需与请求中的 tool_call_id 保持一致
                    "name": function_name,
                    "content": result
                })
        
        # 第二轮：将工具返回结果提交给模型以生成最终答复
        final_response = client.chat.completions.create(
            model="qwen-turbo",
            messages=messages
        )
        return final_response.choices[0].message.content
        
    return msg.content

print(run_agent_step("帮我看看关于智能体开发的那篇文章写了什么？"))

工具调用注意事项

• 平行工具调用（Parallel Function Calling）：模型可能在单次响应中返回多个工具调用指令（如并发查询多个目标），客户端需遍历处理 tool_calls 列表中的每一项。
• 上下文一致性：提交工具执行结果时，历史消息中必须完整包含模型上一轮返回的 assistant 消息（含 tool_calls 属性），否则将违反 API 校验协议。
• 循环次数限制：对于支持多轮连续工具调用的场景，应使用循环并设置明确的最大迭代次数限制，防止出现无限递归调用。

记忆与状态管理

由于 LLM API 是无状态的，客户端需要在请求间维护完整的对话历史列表，以支持多轮交互。

上下文维护对比

维护方式	传入消息示例	说明
忽略历史消息	[{"role": "user", "content": "续写上文"}]	模型缺乏前文上下文，无法准确回答。
维护完整消息队列	[System, User(1), Assistant(1), User(2)]	模型获得完整上下文，能实现连贯交互。

基于 Session 的会话管理类实现

class AgentSession:
    def __init__(self, session_id: str):
        self.session_id = session_id
        # 初始化包含系统提示词的消息列表
        self.messages = [{"role": "system", "content": SYSTEM_PROMPT}]

    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})

    def get_history(self) -> list:
        return self.messages

# 使用示例
session = AgentSession("user_001")
session.add_message("user", "我的名字叫张三")
# 处理模型响应后记录: session.add_message("assistant", "你好，张三。")
session.add_message("user", "我叫什么名字？")
# 将 session.get_history() 作为 Payload 传给模型即可维持上下文

持久化方案

在生产环境中，内存维护的列表随进程重启而清空。推荐将消息列表序列化为 JSON 格式后持久化：

• Redis：适合高速会话读取与自动化 TTL 过期策略。
• 关系型数据库（如 MySQL/PostgreSQL）：适合长期审计、日志记录及分析。

上下文剪裁策略

长时间对话会导致消息历史长度超出模型最大 Token 窗口，增加接口成本并延长首字响应延迟。常用的剪裁策略如下：

策略 A：滑动窗口剪裁（Sliding Window）

滑动窗口策略按消息条数丢弃早期历史，保持固定窗口大小 $K$，且固定保留首条 System Prompt。

def sliding_window_trim(messages: list, k: int = 6) -> list:
    """
    保留包含 System Prompt 在内的最近 k 条消息。
    建议 k 设定为偶数（加上 System Prompt 后共 k 条），确保问答对完整。
    """
    if len(messages) <= k:
        return messages
    
    system_msg = messages[0]
    recent_msgs = messages[-(k - 1):]
    return [system_msg] + recent_msgs

# 在对话循环中的应用示例
history = [{"role": "system", "content": "你是一个助手"}]

while True:
    user_input = input("User: ")
    if user_input == "quit":
        break
    
    history.append({"role": "user", "content": user_input})
    
    # 限制发送至 API 的历史消息数量
    trimmed_history = sliding_window_trim(history, k=6)
    
    response = client.chat.completions.create(
        model="qwen-turbo",
        messages=trimmed_history
    )
    
    answer = response.choices[0].message.content
    print(f"AI: {answer}")
    
    # 完整历史中仍记录模型响应内容
    history.append({"role": "assistant", "content": answer})

• 特性：逻辑简单，无需额外的计算开销。
• 隐患：可能中断未完成的工具调用上下文（如截断了包含 tool_calls 的请求但保留了反馈），需额外校验消息角色完整性。

策略 B：基于 Token 或字符数的限制剪裁

根据 Token 或字符总数设定限制阈值，超出时逐轮丢弃队头最早的问答消息。

1. 基于 tiktoken 的分词截断（适用于 OpenAI 系列模型）

import tiktoken

def trim_by_tokens(messages: list, max_tokens: int = 4000, model: str = "gpt-4o") -> list:
    encoding = tiktoken.encoding_for_model(model)
    while True:
        total_text = "".join([m["content"] for m in messages if isinstance(m.get("content"), str)])
        num_tokens = len(encoding.encode(total_text))
        if num_tokens <= max_tokens or len(messages) <= 2:
            break
        # 丢弃早期的一轮问答消息（保留 System Prompt）
        messages.pop(1)
    return messages

2. 基于字符数的保守估算截断（通用方案）

对于非 OpenAI 模型（如 Qwen 或 Llama），其 Tokenizer 计算规则不同。可采用字符数保守估计方案：

def estimate_tokens(messages: list) -> int:
    """按字符数进行保守估算"""
    total = 0
    for m in messages:
        content = m.get("content", "")
        if isinstance(content, str):
            total += len(content)
    return total

def trim_by_estimate(messages: list, max_limit: int = 8000) -> list:
    while estimate_tokens(messages) > max_limit and len(messages) > 2:
        messages.pop(1)
    return messages

策略 C：历史消息摘要化（Summarization）

当对话历史较长且需要保留早期关键事实时，可以定期调用轻量级模型对旧历史进行摘要提炼，并将摘要注入系统上下文。

def get_summary(old_messages: list) -> str:
    """调用轻量级模型对旧对话生成摘要"""
    prompt = f"请总结以下对话的核心内容，保留关键事实与决策：\n{str(old_messages)}"
    response = client.chat.completions.create(
        model="qwen-turbo",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

# 当总字符数超过阈值时触发摘要处理
history = [...]  # 积累的对话历史

if estimate_tokens(history) > 10000:
    system_msg = history[0]
    to_summarize = history[1:-10]
    recent_msgs = history[-10:]
    
    summary_text = get_summary(to_summarize)
    
    # 将摘要注入系统指导信息中
    summary_msg = {
        "role": "system", 
        "content": f"【历史对话摘要】：{summary_text}"
    }
    
    # 重构后的历史结构: 原始 System + 历史摘要 + 近期原始消息
    history = [system_msg, summary_msg] + recent_msgs