这个模板解决什么问题
用OpenAI的Function Calling搭Agent，官方文档给了碎片化的示例，但没告诉你怎么把这些碎片拼成一个完整的、可维护的项目。你可能遇到过这些问题：
对话历史怎么管理，全塞进messages数组会越来越长吧？多个工具怎么组织，每个工具写一堆if-else？流式输出和工具调用同时用会冲突吗？
下面这份模板把这些问题都处理了。
项目结构
ai-agent-starter/
├── agent.py # Agent主逻辑
├── tools.py # 工具定义和实现
├── memory.py # 对话记忆管理
├── config.py # 配置
├── requirements.txt # 依赖
└── main.py # 入口requirements.txt
openai>=1.50.0
python-dotenv>=1.0.0
httpx>=0.27.0config.py
import os
from dotenv import load_dotenv
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
MODEL = os.getenv("MODEL", "gpt-4o-mini")
MAX_HISTORY = int(os.getenv("MAX_HISTORY", "20"))
SYSTEM_PROMPT = os.getenv(
"SYSTEM_PROMPT",
"你是一个有帮助的AI助手。根据用户的问题调用合适的工具获取信息，然后给出清晰的回答。"
)tools.py
工具用统一的装饰器注册，新增工具只需要写函数加装饰器，不用改Agent主逻辑。
import json
import httpx
from functools import wraps
TOOLS_REGISTRY = []
TOOLS_FUNCTIONS = {}
def tool(name, description, parameters):
def decorator(func):
TOOLS_REGISTRY.append({
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": parameters,
}
})
TOOLS_FUNCTIONS[name] = func
@wraps(func)
def wrapper(**kwargs):
return func(**kwargs)
return wrapper
return decorator
@tool(
name="get_weather",
description="获取指定城市的当前天气信息",
parameters={
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称，如北京、上海"}
},
"required": ["city"],
},
)
def get_weather(city: str) -> str:
resp = httpx.get(
f"https://wttr.in/{city}",
params={"format": "j1"},
timeout=10,
)
data = resp.json()
current = data["current_condition"][0]
return json.dumps({
"city": city,
"temp_c": current["temp_C"],
"humidity": current["humidity"],
"desc": current["weatherDesc"][0]["value"],
})
@tool(
name="calculate",
description="进行数学计算，支持加减乘除",
parameters={
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式，如 2+3*4"}
},
"required": ["expression"],
},
)
def calculate(expression: str) -> str:
import ast
import operator
ops = {
ast.Add: operator.add, ast.Sub: operator.sub,
ast.Mult: operator.mul, ast.Div: operator.truediv,
}
def eval_node(node):
if isinstance(node, ast.Constant):
return node.value
if isinstance(node, ast.BinOp):
return ops[type(node.op)](eval_node(node.left), eval_node(node.right))
raise ValueError(f"不支持的表达式: {ast.dump(node)}")
tree = ast.parse(expression, mode="eval")
result = eval_node(tree.body)
return json.dumps({"expression": expression, "result": result})
def get_tools_schema():
return TOOLS_REGISTRY
def execute_tool(name, arguments):
func = TOOLS_FUNCTIONS.get(name)
if not func:
return json.dumps({"error": f"工具 {name} 不存在"})
return func(**arguments)memory.py
对话记忆用滑动窗口策略，保留最近N轮对话。超出部分从前面裁掉，但system prompt始终保留。
class ConversationMemory:
def __init__(self, system_prompt, max_history=20):
self.system_prompt = system_prompt
self.max_history = max_history
self.messages = [{"role": "system", "content": system_prompt}]
def add(self, role, content):
self.messages.append({"role": role, "content": content})
self._trim()
def add_tool_message(self, tool_call_id, content):
self.messages.append({
"role": "tool",
"tool_call_id": tool_call_id,
"content": content,
})
self._trim()
def get_messages(self):
return self.messages
def _trim(self):
if len(self.messages) <= self.max_history:
return
system = self.messages[0]
rest = self.messages[1:]
half = self.max_history // 2
self.messages = [system] + rest[-half * 2:]
def clear(self):
self.messages = [{"role": "system", "content": self.system_prompt}]agent.py
Agent主循环处理三件事：调模型、执行工具、返回结果。支持流式和非流式两种模式。
import json
from openai import OpenAI
from config import OPENAI_API_KEY, MODEL, MAX_HISTORY, SYSTEM_PROMPT
from tools import get_tools_schema, execute_tool
from memory import ConversationMemory
class Agent:
def __init__(self):
self.client = OpenAI(api_key=OPENAI_API_KEY)
self.memory = ConversationMemory(SYSTEM_PROMPT, MAX_HISTORY)
self.tools = get_tools_schema()
def chat(self, user_input, stream=False):
self.memory.add("user", user_input)
if stream:
return self._chat_stream()
return self._chat_sync()
def _chat_sync(self):
response = self.client.chat.completions.create(
model=MODEL,
messages=self.memory.get_messages(),
tools=self.tools,
tool_choice="auto",
)
msg = response.choices[0].message
if msg.tool_calls:
self.memory.add("assistant", msg.content or "")
for call in msg.tool_calls:
args = json.loads(call.function.arguments)
result = execute_tool(call.function.name, args)
self.memory.add_tool_message(call.id, result)
return self._chat_sync()
self.memory.add("assistant", msg.content)
return msg.content
def _chat_stream(self):
collected_content = ""
stream = self.client.chat.completions.create(
model=MODEL,
messages=self.memory.get_messages(),
tools=self.tools,
tool_choice="auto",
stream=True,
)
tool_calls_buffer = {}
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
collected_content += delta.content
yield {"type": "content", "text": delta.content}
if delta.tool_calls:
for tc in delta.tool_calls:
idx = tc.index
if idx not in tool_calls_buffer:
tool_calls_buffer[idx] = {"id": "", "name": "", "args": ""}
if tc.id:
tool_calls_buffer[idx]["id"] = tc.id
if tc.function and tc.function.name:
tool_calls_buffer[idx]["name"] = tc.function.name
if tc.function and tc.function.arguments:
tool_calls_buffer[idx]["args"] += tc.function.arguments
if tool_calls_buffer:
self.memory.add("assistant", collected_content or "")
for idx in sorted(tool_calls_buffer.keys()):
tc = tool_calls_buffer[idx]
args = json.loads(tc["args"]) if tc["args"] else {}
result = execute_tool(tc["name"], args)
self.memory.add_tool_message(tc["id"], result)
yield {"type": "tool", "name": tc["name"], "result": result}
yield from self._chat_stream()
else:
self.memory.add("assistant", collected_content)
def create_agent():
return Agent()main.py
入口提供命令行交互模式。
from agent import create_agent
def run_interactive():
agent = create_agent()
print("Agent已启动，输入quit退出。")
while True:
user_input = input("\n你: ").strip()
if user_input.lower() in ("quit", "exit"):
break
if not user_input:
continue
print("\nAI: ", end="", flush=True)
for chunk in agent.chat(user_input, stream=True):
if chunk["type"] == "content":
print(chunk["text"], end="", flush=True)
elif chunk["type"] == "tool":
print(f"\n[调用工具: {chunk['name']}]")
print()
if __name__ == "__main__":
run_interactive()几个设计决策的说明
工具注册用装饰器模式。新增工具时只需要在tools.py里写函数加@tool装饰器，不用动agent.py。这比写一大堆if-elif分支干净得多。
记忆管理用滑动窗口。对话轮次多了，全量传给模型既浪费token又可能超出上下文窗口。MAX_HISTORY设20，实际保留最近10轮对话加system prompt，对大部分场景够用。如果需要更复杂的记忆（比如摘要压缩），在memory.py里改就行。
流式模式下工具调用的处理稍微复杂。OpenAI流式返回tool_calls是分段的，id、name、arguments可能跨多个chunk到达。代码里用tool_calls_buffer累积，等流结束后统一执行。这个细节官方文档没有明确说，踩过坑才知道。
怎么扩展
加新工具：在tools.py里写函数，加@tool装饰器，完事。
换模型：在.env里设MODEL变量。gpt-4o-mini跑Agent够用，成本大约每次请求0.0001美元（按OpenAI定价页gpt-4o-mini输入0.15美元/百万token计算）。
接持久化记忆：把memory.py里的ConversationMemory换成Redis或PostgreSQL存储，接口保持add和get_messages两个方法就行。
加RAG：在chat方法里先做向量检索，把检索结果塞进system prompt或作为上下文消息插入。工具里也可以加一个search_documents工具让Agent主动检索。
运行
# 安装依赖
pip install -r requirements.txt
# 配置环境变量
echo 'OPENAI_API_KEY=sk-your-key-here' > .env
echo 'MODEL=gpt-4o-mini' >> .env
# 启动
python main.py试一下问它"北京今天多少度"或者"帮我算一下 123 乘以 456"，你会看到Agent自己决定调用哪个工具，拿到结果后组织回答，全程流式输出。
这份模板的代码量不到200行，但覆盖了Agent开发的核心问题：工具定义、调用循环、记忆管理、流式输出。拿去改改就能当项目骨架用。别从零开始搭了。
