这次我们来看一个名为“建立循环的循环：后台智能体新思路”的技术项目。这个项目并非一个具体的软件或模型，而是一种关于构建“后台智能体”（Background Agent）的系统架构设计理念。其核心思想是让智能体能够自我迭代、自我优化，形成一个持续进化的“循环之循环”，从而在无人值守的后台环境中，长期、稳定、自主地处理复杂任务。

对于开发者而言，最值得关注的不是某个现成的工具包，而是这套思路如何落地。它能否降低长期运行智能体的运维成本？能否实现任务的自动化编排与错误自愈？能否在有限的资源下持续学习？本文将围绕这些核心问题，拆解“循环的循环”这一架构的关键组件、实现路径与验证方法。

本文将带你完成以下内容：首先，快速梳理这一思路的核心能力与适用边界；然后，从零开始规划一套可验证的简易后台智能体系统，涵盖环境准备、核心循环逻辑实现、任务持久化与状态恢复；接着，我们会设计测试用例，验证其容错与自我优化能力；最后，探讨如何为其添加API接口、监控告警，并总结在工程化实践中常见的陷阱与最佳实践。

如果你正在研究智能体的长期运行、自动化运维或多智能体协作系统，这篇文章提供的架构思路和实现样板将为你节省大量前期设计时间。

1. 核心能力速览

“建立循环的循环”这一理念，旨在构建一个高阶的管理循环，来监督和优化底层一个或多个执行智能体的工作循环。下表概括了其核心设计目标与能力特征：

能力项 说明
核心理念 实现智能体系统的“元管理”，即一个智能体负责监控、评估和调整另一个智能体的策略与行为，形成双层循环结构。
核心功能 1. 任务持久化与状态恢复 ：智能体中断后可从检查点重启。
2. 性能监控与评估 ：自动收集执行指标（成功率、耗时、资源消耗）。
3. 策略动态调整 ：根据评估结果，自动修改提示词、工作流或调用工具的策略。
4. 错误处理与重试 ：定义异常处理逻辑，实现任务级或步骤级重试。
技术栈倾向 Python（主流），可能涉及LangChain、AutoGen、LlamaIndex等智能体框架，以及Redis/Celery用于任务队列，SQLite/PostgreSQL用于状态存储。
资源需求 取决于底层智能体模型（LLM）。若使用云API（如GPT-4），则关注网络与费用；若本地部署大模型，则需相应GPU显存。管理循环本身资源消耗极低。
启动方式 通常以守护进程（Daemon）或计划任务（Cron）形式后台运行，可通过命令行或系统服务（systemd/supervisor）管理。
接口能力 可对外提供REST API或消息队列接口，用于提交新任务、查询状态、注入评估规则。
适合场景 需7x24小时运行的自动化客服、内容审核、数据爬取与清洗、市场监控、自动化测试、DevOps运维等场景。

2. 适用场景与使用边界

适合谁？解决什么问题？

这套架构特别适合两类开发者：

  1. 业务自动化开发者 ：需要构建一个能长期运行、遇到问题能自己“想办法”的自动化流程，而不仅仅是执行固定脚本。
  2. 智能体系统研究者 ：希望探索智能体的长期记忆、持续学习和元认知能力，需要一个可扩展的实验框架。

它能解决的核心痛点是 智能体运行的脆弱性 。传统一次性运行的智能体，遇到未预见的输入、工具调用失败或上下文过长等问题时就会崩溃。而“循环的循环”通过上层监控和调整，赋予了系统 韧性 和 适应性 。

不适合什么场景？

  • 简单、确定性的任务 ：如果任务逻辑完全固定且稳定，用常规脚本或工作流引擎更简单高效。
  • 对实时性要求极高的场景 ：双层循环引入了一定的决策开销，不适合毫秒级响应的交易系统。
  • 资源极度受限的环境 ：如果连运行一个基础LLM都困难，叠加管理循环只会增加负担。

合规与安全边界

  • 数据安全 ：后台智能体可能长期接触业务数据，必须确保数据存储、传输加密，并遵守相关数据隐私法规。
  • 操作权限 ：智能体被赋予的API或系统工具调用权限必须遵循最小权限原则，防止越权操作。
  • 内容合规 ：对于内容生成或审核类智能体，必须内置过滤机制，并定期审核其输出，避免产生违规内容。
  • 可控性 ：必须保留人工介入和紧急停止的“红色按钮”，确保智能体不会进入不可控的循环。

3. 环境准备与前置条件

在开始构建之前，请确保你的开发环境满足以下基础要求。我们将以一个基于Python和轻量级LLM的本地化方案为例。

  1. 操作系统 ：Linux (Ubuntu 20.04+)、macOS 或 Windows (WSL2推荐)。生产环境建议Linux。
  2. Python环境 ：Python 3.9+。强烈建议使用 conda 或 venv 创建独立的虚拟环境。 
    # 创建并激活虚拟环境
python -m venv agent_loop_env
source agent_loop_env/bin/activate  # Linux/macOS
# agent_loop_env\Scripts\activate  # Windows
  3. 基础依赖 ：安装必要的Python包。 
    pip install openai langchain langchain-community sqlalchemy redis celery psutil
    注：这里以LangChain为例，你可根据喜好替换为其他框架。
  4. LLM基础 ：选择一种LLM供给方式。
    • 方案A（云API，简单） ：准备有效的OpenAI、Anthropic或国内合规大模型的API Key。
    • 方案B（本地模型，可控） ：安装Ollama或LM Studio，并拉取一个轻量级模型（如 qwen2.5:7b 、 llama3.2:3b ）。 
      # 以Ollama为例
curl -fsSL https://ollama.com/install.sh | sh
ollama pull qwen2.5:7b
  5. 持久化存储 ：我们需要一个数据库来存储任务状态和循环日志。SQLite（本地）或PostgreSQL（生产）均可。 
    # 如果使用PostgreSQL，请确保已安装并运行
# sudo apt-get install postgresql postgresql-contrib # Ubuntu
  6. 任务队列（可选，用于解耦） ：对于复杂任务流，建议使用Redis + Celery。 
    # 安装Redis
# sudo apt-get install redis-server # Ubuntu
# 或使用Docker: docker run -d -p 6379:6379 redis:alpine

4. 系统架构与核心循环实现

我们来构建一个最小可行系统（MVS）。该系统包含两个核心组件：

  • 执行智能体（Worker Agent） ：负责执行具体任务（例如，分析一篇新闻的情感）。
  • 管理智能体（Manager Agent） ：负责监控Worker的表现，并根据历史数据调整其策略（例如，修改提示词）。

4.1 数据库模型设计

首先，定义存储任务、执行历史和评估结果的数据模型。

# models.py
from sqlalchemy import create_engine, Column, Integer, String, DateTime, Float, Text, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime

Base = declarative_base()

class Task(Base):
    __tablename__ = 'tasks'
    id = Column(Integer, primary_key=True)
    name = Column(String(255), nullable=False)
    input_data = Column(Text)  # 任务输入，如URL、文本
    status = Column(String(50), default='pending')  # pending, running, success, failed, retrying
    created_at = Column(DateTime, default=datetime.utcnow)
    started_at = Column(DateTime)
    completed_at = Column(DateTime)
    result = Column(Text)  # 任务执行结果
    error_message = Column(Text)

class ExecutionLog(Base):
    __tablename__ = 'execution_logs'
    id = Column(Integer, primary_key=True)
    task_id = Column(Integer, index=True)
    agent_type = Column(String(50))  # 'worker' or 'manager'
    action = Column(String(255))  # 执行的动作描述
    prompt_used = Column(Text)  # 本次使用的提示词（快照）
    metrics = Column(Text)  # JSON格式的性能指标，如耗时、token数
    created_at = Column(DateTime, default=datetime.utcnow)

class Strategy(Base):
    __tablename__ = 'strategies'
    id = Column(Integer, primary_key=True)
    name = Column(String(255), unique=True)  # 策略名称，如“情感分析提示词v1”
    config = Column(Text)  # JSON格式的策略配置，如提示词模板、工具列表
    is_active = Column(Boolean, default=True)
    performance_score = Column(Float, default=0.0)  # 由管理智能体评估的分数
    created_at = Column(DateTime, default=datetime.utcnow)

# 初始化数据库连接
engine = create_engine('sqlite:///agent_loop.db')  # 生产环境可换为PostgreSQL连接字符串
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(bind=engine)

4.2 执行智能体（Worker Agent）实现

这是一个简单的任务执行者，它从数据库获取 pending 状态的任务，使用当前激活的策略执行。

# worker_agent.py
import logging
from langchain.chat_models import ChatOpenAI  # 或ChatOllama
from langchain.schema import HumanMessage, SystemMessage
from models import SessionLocal, Task, ExecutionLog, Strategy
import json
import time

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class WorkerAgent:
    def __init__(self, llm):
        self.llm = llm
        self.session = SessionLocal()

    def get_active_strategy(self):
        """获取当前激活的策略"""
        strategy = self.session.query(Strategy).filter_by(is_active=True).first()
        if not strategy:
            # 默认策略
            default_config = {
                "system_prompt": "你是一个专业的分析助手。请根据用户输入，完成指定的分析任务。",
                "max_tokens": 500
            }
            strategy = Strategy(name="default", config=json.dumps(default_config), is_active=True)
            self.session.add(strategy)
            self.session.commit()
        return json.loads(strategy.config)

    def execute_task(self, task_id):
        """执行一个具体的任务"""
        task = self.session.query(Task).get(task_id)
        if not task or task.status != 'pending':
            return

        task.status = 'running'
        task.started_at = datetime.utcnow()
        self.session.commit()

        strategy_config = self.get_active_strategy()
        system_prompt = strategy_config.get("system_prompt", "")
        
        try:
            # 构建LLM请求
            messages = [
                SystemMessage(content=system_prompt),
                HumanMessage(content=f"请分析以下内容：\n{task.input_data}")
            ]
            start_time = time.time()
            response = self.llm.invoke(messages)
            elapsed_time = time.time() - start_time

            # 记录执行日志
            log = ExecutionLog(
                task_id=task.id,
                agent_type='worker',
                action=f'execute_task_{task.name}',
                prompt_used=system_prompt,
                metrics=json.dumps({"time_elapsed": elapsed_time, "response_length": len(response.content)})
            )
            self.session.add(log)

            # 更新任务状态
            task.status = 'success'
            task.result = response.content
            task.completed_at = datetime.utcnow()
            self.session.commit()
            logger.info(f"Task {task_id} executed successfully.")

        except Exception as e:
            logger.error(f"Task {task_id} failed: {e}")
            task.status = 'failed'
            task.error_message = str(e)
            self.session.commit()

    def run_loop(self):
        """Worker的主循环，持续从数据库拉取任务执行"""
        logger.info("Worker agent started.")
        while True:
            pending_tasks = self.session.query(Task).filter_by(status='pending').limit(5).all()
            for task in pending_tasks:
                self.execute_task(task.id)
            time.sleep(5)  # 每5秒检查一次新任务

if __name__ == "__main__":
    # 初始化LLM (示例：使用Ollama本地模型)
    # from langchain_community.chat_models import ChatOllama
    # llm = ChatOllama(model="qwen2.5:7b", temperature=0.1)
    
    # 示例：使用OpenAI API (需设置环境变量OPENAI_API_KEY)
    from langchain_openai import ChatOpenAI
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.1)

    agent = WorkerAgent(llm)
    agent.run_loop()

4.3 管理智能体（Manager Agent）实现

管理智能体定期运行，检查 ExecutionLog 和 Task 表，评估Worker的表现，并决定是否调整策略。

# manager_agent.py
import logging
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage
from models import SessionLocal, ExecutionLog, Task, Strategy
import json
import statistics
from datetime import datetime, timedelta

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ManagerAgent:
    def __init__(self, llm):
        self.llm = llm
        self.session = SessionLocal()

    def evaluate_performance(self, hours=1):
        """评估过去一段时间内任务的执行表现"""
        since_time = datetime.utcnow() - timedelta(hours=hours)
        logs = self.session.query(ExecutionLog).filter(
            ExecutionLog.agent_type == 'worker',
            ExecutionLog.created_at >= since_time
        ).all()

        success_tasks = self.session.query(Task).filter(
            Task.status == 'success',
            Task.completed_at >= since_time
        ).count()
        total_tasks = self.session.query(Task).filter(Task.completed_at >= since_time).count()
        success_rate = success_tasks / total_tasks if total_tasks > 0 else 0

        times = []
        for log in logs:
            metrics = json.loads(log.metrics) if log.metrics else {}
            if 'time_elapsed' in metrics:
                times.append(metrics['time_elapsed'])
        avg_time = statistics.mean(times) if times else 0

        return {
            "success_rate": success_rate,
            "avg_execution_time": avg_time,
            "sample_size": total_tasks
        }

    def analyze_and_adjust(self, performance_metrics):
        """分析性能指标，并决定是否调整策略"""
        # 如果成功率太低或耗时太长，则尝试优化策略
        if performance_metrics['sample_size'] > 10 and (
            performance_metrics['success_rate'] < 0.8 or 
            performance_metrics['avg_execution_time'] > 30.0
        ):
            logger.warning(f"Performance below threshold: {performance_metrics}. Considering strategy adjustment.")
            
            # 调用LLM分析日志，生成新的提示词建议
            recent_logs = self.session.query(ExecutionLog).filter_by(agent_type='worker').order_by(ExecutionLog.id.desc()).limit(10).all()
            log_context = "\n".join([f"- {log.action}: {log.prompt_used[:100]}..." for log in recent_logs])
            
            analysis_prompt = f"""
            作为智能体系统管理员，我观察到以下性能问题：
            - 任务成功率: {performance_metrics['success_rate']:.2%}
            - 平均执行时间: {performance_metrics['avg_execution_time']:.2f}秒
            最近的部分执行日志如下：
            {log_context}
            请分析可能导致问题的原因（例如：提示词不清晰、任务类型复杂），并生成一个改进后的系统提示词（System Prompt），用于指导执行智能体更好地完成任务。新的提示词应更具体、更具指导性。
            只返回改进后的提示词文本。
            """
            messages = [
                SystemMessage(content="你是一个资深的AI提示词工程师和系统优化专家。"),
                HumanMessage(content=analysis_prompt)
            ]
            try:
                new_prompt = self.llm.invoke(messages).content.strip()
                
                # 创建新策略
                new_strategy = Strategy(
                    name=f"optimized_v{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}",
                    config=json.dumps({"system_prompt": new_prompt}),
                    performance_score=0.0  # 初始分数，后续由效果评估更新
                )
                self.session.add(new_strategy)
                
                # 可选：停用旧策略，或设置新策略为活跃
                # old_strategy = self.session.query(Strategy).filter_by(is_active=True).first()
                # if old_strategy:
                #     old_strategy.is_active = False
                # new_strategy.is_active = True
                
                self.session.commit()
                logger.info(f"New strategy created: {new_strategy.name}")
                return new_strategy
            except Exception as e:
                logger.error(f"Failed to generate new strategy: {e}")
        else:
            logger.info(f"Performance is acceptable: {performance_metrics}. No adjustment needed.")
        return None

    def run_loop(self, evaluation_interval_minutes=30):
        """Manager的主循环，定期评估和调整"""
        logger.info("Manager agent started.")
        while True:
            import time
            time.sleep(evaluation_interval_minutes * 60)  # 转换为秒
            
            logger.info("Starting performance evaluation cycle...")
            metrics = self.evaluate_performance(hours=1)
            self.analyze_and_adjust(metrics)

if __name__ == "__main__":
    # 初始化LLM，Manager可以使用更强的模型
    from langchain_openai import ChatOpenAI
    llm = ChatOpenAI(model="gpt-4", temperature=0.1)  # 使用更强的模型进行分析
    
    agent = ManagerAgent(llm)
    agent.run_loop()

5. 系统启动与功能测试

5.1 启动后台服务

我们需要将Worker和Manager作为独立的守护进程启动。可以使用 supervisor 或 systemd 管理，这里先用简单的脚本演示。

创建一个启动脚本 start_agents.sh ：

#!/bin/bash
# start_agents.sh
source /path/to/your/agent_loop_env/bin/activate

# 启动Worker Agent， 输出日志到文件
python worker_agent.py > worker.log 2>&1 &
WORKER_PID=$!
echo "Worker Agent started with PID: $WORKER_PID"

# 启动Manager Agent， 输出日志到文件
python manager_agent.py > manager.log 2>&1 &
MANAGER_PID=$!
echo "Manager Agent started with PID: $MANAGER_PID"

# 保存PID以便后续管理
echo $WORKER_PID > worker.pid
echo $MANAGER_PID > manager.pid

echo "Agents started. Check worker.log and manager.log for details."

停止脚本 stop_agents.sh ：

#!/bin/bash
# stop_agents.sh
if [ -f worker.pid ]; then
    kill $(cat worker.pid) && rm worker.pid
    echo "Worker Agent stopped."
fi
if [ -f manager.pid ]; then
    kill $(cat manager.pid) && rm manager.pid
    echo "Manager Agent stopped."
fi

5.2 功能测试：验证循环的循环

现在，我们来模拟真实场景，测试整个系统是否工作。

步骤1：插入测试任务 编写一个脚本向数据库添加几个待分析的任务。

# test_insert_tasks.py
from models import SessionLocal, Task
from datetime import datetime

session = SessionLocal()

test_tasks = [
    {"name": "情感分析_新闻1", "input_data": "公司今日发布了年度财报，利润大幅增长，市场反应积极。"},
    {"name": "情感分析_新闻2", "input_data": "新产品发布后出现严重漏洞，用户投诉激增，股价应声下跌。"},
    {"name": "情感分析_新闻3", "input_data": "这是一段中性的文本，描述了一个普通的天气情况，晴转多云，气温25度。"},
]

for task_data in test_tasks:
    task = Task(**task_data)
    session.add(task)

session.commit()
print(f"Inserted {len(test_tasks)} test tasks.")
session.close()

步骤2：启动系统并观察

  1. 运行 bash start_agents.sh 启动Worker和Manager。
  2. 观察 worker.log 文件，应该能看到Worker拉取任务并执行。 
    INFO:__main__:Worker agent started.
INFO:__main__:Task 1 executed successfully.
INFO:__main__:Task 2 executed successfully.
INFO:__main__:Task 3 executed successfully.
  3. 查询数据库，确认任务状态变为 success ，并且 result 字段有内容。 
    sqlite3 agent_loop.db "SELECT id, name, status, substr(result, 1, 50) as preview FROM tasks;"

步骤3：触发管理循环

  1. Manager默认每30分钟评估一次。为了快速测试，可以修改 manager_agent.py 中的 run_loop 函数，将间隔临时改为60秒。
  2. 或者，手动插入一些“失败”的任务（例如，将 input_data 设置为空字符串或乱码），以降低成功率，触发Manager的调整机制。
  3. 观察 manager.log 文件，当检测到性能不达标时，会看到类似日志： 
    WARNING:__main__:Performance below threshold: {'success_rate': 0.5, ...}. Considering strategy adjustment.
INFO:__main__:New strategy created: optimized_v20231026_143022
  4. 检查 strategies 表，会发现新增了一条策略记录，其 config 字段包含了LLM生成的新提示词。

步骤4：验证策略更新效果

  1. 手动将新创建的策略设置为 is_active=True （或在Manager代码中取消注释自动激活的代码）。
  2. 再次插入新的测试任务。
  3. 观察Worker是否使用了新的提示词（通过 execution_logs 表的 prompt_used 字段查看），并评估任务成功率是否有所改善。

至此，一个具备“循环的循环”雏形的后台智能体系统就完成了核心功能的验证。Worker负责执行，Manager负责监控和优化，形成了一个自我迭代的闭环。

6. 接口API与批量任务集成

为了让系统更易用，我们需要提供API来提交任务和查询状态。同时，支持批量任务提交是生产环境的必备能力。

6.1 使用FastAPI构建REST API

创建一个 api_server.py 文件：

# api_server.py
from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from models import SessionLocal, Task
from datetime import datetime
import uuid

app = FastAPI(title="后台智能体API服务")

class TaskCreateRequest(BaseModel):
    name: str
    input_data: str

class BatchTaskRequest(BaseModel):
    tasks: List[TaskCreateRequest]

@app.post("/api/tasks", status_code=202)
async def create_task(task_req: TaskCreateRequest, background_tasks: BackgroundTasks):
    """提交单个任务"""
    session = SessionLocal()
    db_task = Task(name=task_req.name, input_data=task_req.input_data)
    session.add(db_task)
    session.commit()
    task_id = db_task.id
    session.close()
    # 在实际项目中，这里可能会触发一个Celery任务，而非直接依赖Worker轮询
    return {"task_id": task_id, "message": "Task accepted", "status": "pending"}

@app.post("/api/tasks/batch", status_code=202)
async def create_batch_tasks(batch_req: BatchTaskRequest):
    """批量提交任务"""
    session = SessionLocal()
    task_ids = []
    for task_req in batch_req.tasks:
        db_task = Task(name=task_req.name, input_data=task_req.input_data)
        session.add(db_task)
        session.flush()  # 获取id
        task_ids.append(db_task.id)
    session.commit()
    session.close()
    return {"task_ids": task_ids, "message": f"{len(task_ids)} tasks accepted", "status": "pending"}

@app.get("/api/tasks/{task_id}")
async def get_task_status(task_id: int):
    """查询任务状态和结果"""
    session = SessionLocal()
    task = session.query(Task).get(task_id)
    session.close()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return {
        "task_id": task.id,
        "name": task.name,
        "status": task.status,
        "result": task.result,
        "error": task.error_message,
        "created_at": task.created_at,
        "completed_at": task.completed_at
    }

@app.get("/api/tasks")
async def list_tasks(limit: int = 100, status: Optional[str] = None):
    """列出任务，支持按状态过滤"""
    session = SessionLocal()
    query = session.query(Task)
    if status:
        query = query.filter_by(status=status)
    tasks = query.order_by(Task.id.desc()).limit(limit).all()
    session.close()
    return [{
        "task_id": t.id,
        "name": t.name,
        "status": t.status,
        "created_at": t.created_at
    } for t in tasks]

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

启动API服务： python api_server.py 。现在可以通过 http://localhost:8000/docs 访问交互式API文档。

6.2 批量任务提交与监控示例

使用Python客户端批量提交任务并轮询结果：

# batch_client.py
import requests
import time
import json

API_BASE = "http://localhost:8000/api"

def submit_batch_from_file(file_path):
    """从文件读取任务并批量提交"""
    with open(file_path, 'r', encoding='utf-8') as f:
        # 假设文件每行是一个任务输入
        tasks = [{"name": f"line_{i}", "input_data": line.strip()} for i, line in enumerate(f) if line.strip()]
    
    batch_payload = {"tasks": tasks}
    resp = requests.post(f"{API_BASE}/tasks/batch", json=batch_payload)
    if resp.status_code == 202:
        data = resp.json()
        print(f"Batch submitted. Task IDs: {data['task_ids']}")
        return data['task_ids']
    else:
        print(f"Submission failed: {resp.text}")
        return []

def monitor_tasks(task_ids):
    """监控一批任务直到全部完成"""
    completed = set()
    while len(completed) < len(task_ids):
        for tid in task_ids:
            if tid in completed:
                continue
            resp = requests.get(f"{API_BASE}/tasks/{tid}")
            if resp.status_code == 200:
                task_info = resp.json()
                if task_info['status'] in ['success', 'failed']:
                    print(f"Task {tid} finished with status: {task_info['status']}")
                    completed.add(tid)
        time.sleep(2)  # 每2秒检查一次
    print("All tasks completed.")

if __name__ == "__main__":
    # 示例：从input.txt文件读取任务
    submitted_ids = submit_batch_from_file("input.txt")
    if submitted_ids:
        monitor_tasks(submitted_ids)

7. 资源占用、性能观察与优化

7.1 资源占用观察

  • Worker进程 ：主要资源消耗在于LLM调用。如果使用本地模型，GPU显存是主要瓶颈；如果使用API，则是网络延迟和Token费用。可以通过 psutil 库在代码中监控。 
    import psutil, os
process = psutil.Process(os.getpid())
print(f"Memory RSS: {process.memory_info().rss / 1024 / 1024:.2f} MB")
  • Manager进程 ：资源消耗极低，主要是轻量级的数据库查询和偶尔的LLM调用（用于分析）。
  • 数据库 ：SQLite在轻量级使用下内存和CPU占用可忽略。PostgreSQL在任务量极大时需要适当配置。

7.2 性能优化建议

  1. Worker并发 ：当前的Worker是单线程轮询。对于I/O密集型（调用API）的任务，可以使用 asyncio 或 Celery 实现并发执行，显著提升吞吐量。
  2. 数据库连接池 ：使用 SQLAlchemy 的 scoped_session 和连接池，避免频繁创建连接。
  3. LLM调用批处理 ：如果任务相似，可以将多个任务合并为一个LLM请求，减少调用次数（需LLM支持）。
  4. 缓存 ：对频繁查询且变化不大的数据（如活跃策略）进行内存缓存。
  5. 评估频率 ：Manager的评估间隔 ( evaluation_interval_minutes ) 需要根据业务节奏调整。太频繁浪费资源，太迟钝则响应慢。

7.3 降低显存/内存占用

  • 使用量化模型 ：如果运行本地LLM，使用4-bit或8-bit量化的模型版本。
  • 卸载策略 ：使用 transformers 库的 device_map="auto" 或 accelerate 将模型层卸载到CPU或磁盘。
  • 限制上下文长度 ：在策略配置中限制 max_tokens ，避免处理过长的文本。

8. 常见问题与排查方法

问题现象 可能原因 排查方式 解决方案
Worker不处理任务 1. 数据库连接失败。
2. LLM初始化失败（API Key错误、模型未加载）。
3. 任务状态非 pending 。		 1. 检查 worker.log 错误日志。
2. 检查数据库文件路径和权限。
3. 手动查询 tasks 表状态。		 1. 确认数据库URL正确，文件可写。
2. 检查API Key或本地模型服务（Ollama）是否运行。
3. 重置任务状态为 pending 。
Manager未创建新策略 1. 评估间隔未到。
2. 性能指标未触发阈值（成功率>0.8且耗时<30s）。
3. Manager的LLM调用失败。		 1. 检查 manager.log ，看是否有评估日志。
2. 检查 evaluate_performance 函数返回的指标。
3. 检查Manager的LLM配置和网络。		 1. 临时缩短评估间隔进行测试。
2. 插入一些必定失败的任务来降低成功率。
3. 确保Manager的LLM可用。
API服务无法访问 1. 端口冲突（默认8000）。
2. uvicorn 未正确安装或启动。
3. 防火墙规则阻止。		 1. 使用 netstat -tlnp | grep :8000 检查端口。
2. 查看API服务启动日志。		 1. 修改 api_server.py 中的端口号。
2. 使用 pip install fastapi uvicorn 安装依赖。
3. 检查本地防火墙设置。
批量任务卡住 1. Worker进程挂起或崩溃。
2. 某个任务陷入死循环或LLM无响应。
3. 数据库锁。		 1. 检查Worker进程是否存活 ( ps aux | grep worker )。
2. 查看 worker.log 是否有超时或异常。
3. 检查数据库文件是否被独占锁定。		 1. 重启Worker进程。
2. 为LLM调用设置超时 ( timeout 参数)。
3. 使用任务超时机制，将长时间运行的任务标记为失败。
策略切换后效果更差 1. LLM生成的新提示词质量不佳。
2. 评估样本太少，指标有噪声。		 1. 查看 strategies 表的新提示词内容。
2. 检查评估周期内的任务样本是否具有代表性。		 1. 在Manager的 analyze_and_adjust 中增加对新提示词的验证逻辑（例如A/B测试）。
2. 增加评估样本量或延长评估周期。
数据库文件过大 执行日志和任务历史未清理。 检查 agent_loop.db 文件大小。 增加日志清理脚本，定期归档或删除旧数据。

9. 最佳实践与工程化建议

  1. 从简单开始 ：先用一个固定的、简单的策略让整个流程跑通，再逐步增加Manager的优化逻辑。
  2. 完善的日志 ：除了执行日志，记录详细的调试信息（如LLM的输入输出），这对排查问题至关重要。
  3. 版本化管理策略 ： strategies 表的设计支持版本化。每次调整都生成新版本，并记录性能分数，便于回滚和对比分析。
  4. 引入A/B测试 ：不要立即用新策略替换旧策略。可以同时运行A/B两组任务，对比效果后再决定是否切换。
  5. 设置安全边界 ：
    • 为Manager调整策略的频率和幅度设置上限，防止出现“策略震荡”。
    • 对LLM生成的新提示词进行安全检查（如关键词过滤），避免生成有害指令。
    • 关键操作（如停用核心策略）需要人工确认或设置多重验证。
  6. 监控与告警 ：集成Prometheus、Grafana等监控工具，对任务队列长度、成功率、平均耗时、LLM API调用失败率设置告警。
  7. 数据备份 ：定期备份数据库，尤其是 strategies 表，这是系统经验的结晶。

10. 总结与下一步

“建立循环的循环”这一后台智能体新思路，其核心价值在于将一次性的智能体任务，升级为一个可以 持续运行、自我观察、自我调整 的智能系统。本文通过一个从数据库设计到API暴露的完整可运行示例，演示了如何实现这一架构的最小核心。

最值得尝试的点 ：

  • 韧性提升 ：系统具备了从错误中恢复和调整的基础能力。
  • 持续优化 ：无需人工干预，提示词和工作流可以自动迭代。
  • 可观测性 ：所有决策和结果都被记录，便于分析和审计。

最先应该验证的功能 ：

  1. 基础循环 ：确保Worker能稳定处理任务，Manager能定期运行。
  2. 策略调整触发 ：通过制造“失败”场景，验证Manager能否正确生成新策略。
  3. API与批量处理 ：确认外部系统能方便地提交和获取任务结果。

最容易踩的坑 ：

  • 数据库连接泄露 ：确保每个请求或循环迭代后正确关闭Session。
  • LLM调用超时与限流 ：必须设置合理的超时和重试机制，并处理供应商的速率限制。
  • 评估指标的设计 ：简单的成功率可能不够，需要结合业务设计更精细的评估指标（如结果质量评分）。

后续扩展方向 ：

  • 多Worker协作 ：引入多个不同专长的Worker，由Manager进行任务路由。
  • 工具学习 ：让Manager不仅能调整提示词，还能决定为Worker增加或删除哪些工具（如搜索、计算）。
  • 长期记忆 ：引入向量数据库，让系统能记住历史上的成功与失败案例，进行更精准的调整。
  • 成本控制 ：集成成本监控，在优化效果的同时，将Token消耗或API费用也纳入评估体系。

这个框架是一个起点，你可以根据具体的业务需求，无限扩展其深度和广度。建议先将这个最小系统部署起来，感受“循环”运转起来的力量，然后再逐步添加更复杂的功能。

# 系统架构
Logo
码道开发者社区

码道开发者社区，聚焦华为云码道 CodeArts 代码智能体，沉淀 Agent、Skill、鸿蒙开发实战内容，供开发者查阅资料、交流技术、分享工程实践

更多推荐

Manim 数学动画完全指南：五分钟从零到成片

Manim 是用 Python 驱动的数学动画引擎，把抽象的公式与曲线转成逐帧可控的动态可视化。适合想制作 Python 数学可视化视频的编程新手、数学学生和科普创作者。## 数学为什么需要"动"起来你试过用一张静态图讲清导数的概念吗？切线、割线、极限位置都能画出来，但"割线逼近切线"发生的那一下，观众只能靠脑补。课本插图大多是"结论截图"：只留下最终状态，过程被压缩成一句"易见"。函

avatar 码道开发者社区
cover

Claude API + Python实战：结构化输出后，如何拦住“看起来正确”的JSON

avatar 码道开发者社区
cover

GLM-5.3-Flash深度解析：3200亿参数只卖Opus四十分之一，国产算力首次跑通前沿模型

avatar 码道开发者社区