Agent 框架设计：从选型判断到最小状态机实现

import json, os, sqlite3
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv('DASHSCOPE_API_KEY'),
    base_url='https://dashscope.aliyuncs.com/compatible-mode/v1'
)

# ---- 检查点存储（最小实现：SQLite 持久化）----
class CheckpointStore:
    def __init__(self, db_path='agent_checkpoints.db'):
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        self.conn.execute(
            'CREATE TABLE IF NOT EXISTS cp (thread_id TEXT PRIMARY KEY, state TEXT)'
        )
        self.conn.commit()

    def save(self, thread_id, state):
        self.conn.execute(
            'INSERT OR REPLACE INTO cp VALUES (?, ?)',
            (thread_id, json.dumps(state))
        )
        self.conn.commit()

    def load(self, thread_id):
        row = self.conn.execute(
            'SELECT state FROM cp WHERE thread_id=?', (thread_id,)
        ).fetchone()
        return json.loads(row[0]) if row else None

# ---- 工具定义 ----
def get_time() -> str:
    import datetime
    return datetime.datetime.now().isoformat()

def add(a: int, b: int) -> str:
    return str(a + b)

TOOL_MAP = {'get_time': get_time, 'add': add}

tools_schema = [
    {"type": "function", "function": {
        "name": "get_time", "description": "获取当前时间",
        "parameters": {"type": "object", "properties": {}}
    }},
    {"type": "function", "function": {
        "name": "add", "description": "两数相加",
        "parameters": {
            "type": "object",
            "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
            "required": ["a", "b"]
        }
    }}
]

# ---- 状态机循环 ----
def run_agent(thread_id: str, user_input: str, max_steps: int = 8):
    store = CheckpointStore()
    state = store.load(thread_id) or {"messages": [], "step": 0}
    state["messages"].append({"role": "user", "content": user_input})

    while state["step"] < max_steps:
        resp = client.chat.completions.create(
            model="qwen-plus",
            messages=state["messages"],
            tools=tools_schema
        )
        msg = resp.choices[0].message

        # 终止条件：模型不再请求工具，直接返回文本
        if not msg.tool_calls:
            state["messages"].append({"role": "assistant", "content": msg.content})
            store.save(thread_id, state)
            return msg.content

        # 将模型决策追加到状态（增量更新）
        state["messages"].append(msg)

        # 工具路由与执行
        for tc in msg.tool_calls:
            name = tc.function.name
            args = json.loads(tc.function.arguments)
            if name not in TOOL_MAP:
                result = f"未知工具: {name}"
            else:
                try:
                    result = TOOL_MAP[name](**args)
                except Exception as e:
                    result = f"工具执行错误: {e}"

            state["messages"].append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": str(result)
            })

        state["step"] += 1
        store.save(thread_id, state)  # 每轮落盘

    return "达到最大步数限制，任务未完成。"

# 使用：同一 thread_id 再次调用会从上次状态继续
# print(run_agent("session-1", "现在几点？然后算 3+7"))