下面这段代码演示了如何使用自定义的 SQLAlchemy 子类，实现事务自动提交和回滚：

from flask_sqlalchemy import SQLAlchemy as _SQLAlchemy
from contextlib import contextmanager
class SQLAlchemy:
   @contextmanager
   def auto_commit:
       try:
           yield
           self.session.commit
       except Exception as e:
           self.session.rollback
           raise e
db = SQLAlchemy

随后定义业务实体，例如「App」表：

import uuid
from datetime import datetime
from sqlalchemy import Column, String, Text, DateTime, UUID
class App:
   __tablename__ = 'app'
   id          = Column, primary_key=True,
                        default=uuid.uuid4)
   account_id  = Column, nullable=False)
   name        = Column, default='未命名', nullable=False)
   icon        = Column, default='', nullable=False)
   description = Column
   created_at  = Column(DateTime,
                        default=datetime.utcnow,
                        nullable=False)
   updated_at  = Column(DateTime,
                       default=datetime.utcnow,
                       onupdate=datetime.utcnow,
                       nullable=False)
   def __repr__:
       return f""

from dataclasses import dataclass
from injector import inject
@inject
@dataclass
class AppService:
   db: SQLAlchemy
   def create -> App:
       with self.db.auto_commit:
           app_obj = App(name='Demo 应用',
                         account_id=uuid.uuid4,
                         icon='demo.png',
                         description='演示用')
           self.db.session.add
       return app_obj
   def get -> App | None:
       return db.session.get
   def update_name -> App:
       with self.db.auto_commit:
           app_obj = db.session.get
           if not app_obj:
               raise NotFoundException
           app_obj.name = new_name
       return app_obj
   def delete -> None:
       with self.db.auto_commit:
           obj= db.session.get
           if obj:
               db.session.delete

AIAgent 在执行外部调用时容易抛出各种异常。我们通过自定义基类，把错误信息包装成统一结构：

class CustomException:
   code: str
   message: str
   data: Any | None
   def __init__(self,
                code: str,
                message: str,
                data: Any | None=None):
       super.__init__
       self.code=code;self.message=message;self.data=data
class NotFoundException:
   def __init__:
       super.__init__
# geng多异常类型……

在 Flask 中注册全局错误处理器：

@app.errorhandler
def handle_all:
   if isinstance:
       return fail
   # 未知异常直接返回通用失败码
   return fail)

AIAgent 常常需要参考历史对话才NengZuo出合理决策。LangChain 提供了 ConversationBufferWindowMemory，可限制窗口大小以免磁盘膨胀：

from langchain.memory import ConversationBufferWindowMemory
from langchain.schema import HumanMessage , AIMessage
memory=ConversationBufferWindowMemory(
      k=10,
      input_key='query',
      output_key='answer',
      chat_memory=FileChatMessageHistory
)
# 使用示例：
memory.save_context({'query':'帮我查一下明天北京天气'},
                    {'answer':'好的，我马上查询…'})
history=memory.load_memory_variables
print

。

import pytest
from myproject.app import create_app
@pytest.fixture
def client:
     app=create_app
     app.config=True
     with app.test_client as c:
         yield c
def test_create_app:
     resp=client.post
     assert resp.status_code==200
     assert resp.json=='success'

# 克隆仓库并安装依赖
git clone https://github.com/example/ai-agent-demo.git && cd ai-agent-demo
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
# 创建 .env
cp .env.example .env && vi .env
# 初始化数据库 & 启动服务
flask db upgrade          # alembic migration 
flask run --host=0.0.0.0  # 默认 5000 端口 
curl http://127.0.0.1:5000/api/ping | python -m json.tool  
# => {"code":"success","message":"pong","data":{}}