📑 目录

🤖 AgentGPT 简介

📐 项目背景与定位

🏗️ 核心架构总览

🛠️ 技术栈全景

⚛️ 前端架构 - Next.js

📁 前端目录结构

next/
├── src/
│   ├── app/                    # App Router
│   │   ├── (auth)/            # 认证相关页面
│   │   │   ├── login/
│   │   │   └── register/
│   │   ├── (workspace)/       # 工作区页面
│   │   │   ├── agent/[id]/    # Agent 详情
│   │   │   └── new/           # 创建 Agent
│   │   ├── api/               # API 代理路由
│   │   ├── layout.tsx         # 根布局
│   │   └── page.tsx           # 首页
│   ├── components/            # 组件库
│   │   ├── agent/             # Agent 相关组件
│   │   ├── ui/                # 通用 UI 组件
│   │   └── layout/            # 布局组件
│   ├── stores/                # Zustand 状态
│   ├── hooks/                 # 自定义 Hooks
│   ├── types/                 # TypeScript 类型
│   └── utils/                 # 工具函数
├── public/                    # 静态资源
├── tailwind.config.ts
├── next.config.js
└── package.json

🐍 后端架构 - FastAPI

📁 后端目录结构

platform/
├── reworkd_platform/
│   ├── __init__.py
│   ├── __main__.py            # 入口
│   ├── settings.py            # 配置（Pydantic BaseSettings）
│   ├── constants.py           # 常量定义
│   ├── timer.py               # 计时器
│   ├── logging.py             # 日志配置
│   ├── web/
│   │   ├── application.py     # FastAPI app 创建
│   │   ├── lifetime.py        # 启动/关闭生命周期
│   │   └── api/
│   │       ├── v1/            # API v1 路由
│   │       │   ├── agent.py   # Agent 相关 API
│   │       │   ├── auth.py    # 认证 API
│   │       │   └── tool.py    # 工具 API
│   │       └── router.py      # 路由注册
│   ├── services/
│   │   ├── anthropic.py       # Anthropic 集成
│   │   ├── aws/               # AWS 服务
│   │   ├── pinecone/          # 向量存储
│   │   ├── tokenizer/         # Token 计数
│   │   ├── security.py        # 安全服务
│   │   └── oauth_installers.py
│   ├── db/                    # 数据库模型
│   └── schemas/               # Pydantic schemas
├── pyproject.toml             # Poetry 配置
├── Dockerfile
└── entrypoint.sh

🔄 Agent 核心循环

💻 Agent 循环源码

class AgentExecutor:
    """AgentGPT 核心 Agent 执行器"""

    def __init__(
        self,
        llm: BaseLanguageModel,
        tools: list[BaseTool],
        memory: BaseMemory,
        max_iterations: int = 15,
    ):
        self.llm = llm
        self.tools = tools
        self.memory = memory
        self.max_iterations = max_iterations

    async def execute(self, goal: str) -> AsyncIterator[AgentStep]:
        """执行 Agent 循环，流式返回每一步"""
        context = await self._build_context(goal)

        for i in range(self.max_iterations):
            # 1. 思考：让 LLM 分析当前状态
            thought = await self._think(context)

            # 2. 判断是否完成
            if thought.is_final_answer:
                yield AgentStep(
                    type="final",
                    output=thought.output,
                    iterations=i + 1,
                )
                return

            # 3. 行动：执行选中的工具
            observation = await self._act(thought)

            # 4. 更新上下文
            context = self._update_context(
                context, thought, observation
            )

            yield AgentStep(
                type="action",
                thought=thought,
                observation=observation,
            )

🧠 思考-行动-观察

{
  "tool": "web_search",
  "input": {
    "query": "AgentGPT latest features"
  }
}

🧩 LLM 集成层

🔵 OpenAI 集成

# settings.py - 模型配置
class Settings(BaseSettings):
    openai_api_key: str = ""
    openai_organization: str | None = None
    openai_model: str = "gpt-4"
    openai_temperature: float = 0.7
    openai_max_tokens: int = 1000

    # Token 限制
    model_token_limit: int = 8192

    class Config:
        env_prefix = ""  # 直接读取环境变量

# services/llm_factory.py
from langchain.chat_models import ChatOpenAI

def create_openai_llm(settings: Settings) -> ChatOpenAI:
    """创建 OpenAI LLM 实例"""
    return ChatOpenAI(
        model=settings.openai_model,
        temperature=settings.openai_temperature,
        max_tokens=settings.openai_max_tokens,
        openai_api_key=settings.openai_api_key,
        openai_organization=settings.openai_organization,
        streaming=True,  # 支持流式输出
    )

🟣 Anthropic 集成

# services/anthropic.py
from langchain.chat_models import ChatAnthropic

def create_anthropic_llm(settings: Settings) -> ChatAnthropic:
    """创建 Anthropic Claude LLM 实例"""
    return ChatAnthropic(
        model="claude-3-opus-20240229",
        temperature=settings.anthropic_temperature,
        max_tokens=settings.anthropic_max_tokens,
        anthropic_api_key=settings.anthropic_api_key,
        streaming=True,
    )

# LLM 工厂 - 根据配置自动选择
def create_llm(settings: Settings) -> BaseLanguageModel:
    """根据配置创建对应的 LLM"""
    if settings.model_provider == "openai":
        return create_openai_llm(settings)
    elif settings.model_provider == "anthropic":
        return create_anthropic_llm(settings)
    elif settings.model_provider == "azure":
        return create_azure_llm(settings)
    else:
        raise ValueError(
            f"Unknown provider: {settings.model_provider}"
        )

🔧 工具系统架构

📐 Tool 接口定义

from langchain.tools import BaseTool
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    """搜索工具的输入 Schema"""
    query: str = Field(
        description="搜索查询关键词"
    )
    num_results: int = Field(
        default=5,
        description="返回结果数量"
    )

class WebSearchTool(BaseTool):
    """网页搜索工具"""
    name = "web_search"
    description = (
        "用于在互联网上搜索信息。"
        "输入搜索查询，返回相关网页结果。"
    )
    args_schema = SearchInput

    def _run(self, query: str, num_results: int = 5) -> str:
        """同步执行搜索"""
        return self._search(query, num_results)

    async def _arun(
        self, query: str, num_results: int = 5
    ) -> str:
        """异步执行搜索（Agent 使用此方法）"""
        return await self._async_search(query, num_results)

🧰 内置工具一览

🔍 网页搜索工具

class WebSearchTool(BaseTool):
    """基于 DuckDuckGo 的网页搜索"""
    name = "web_search"
    description = "搜索互联网获取最新信息"

    async def _arun(self, query: str) -> str:
        # 使用 DuckDuckGo 搜索
        async with aiohttp.ClientSession() as session:
            params = {
                "q": query,
                "format": "json",
                "no_html": 1,
            }
            async with session.get(
                "https://api.duckduckgo.com/",
                params=params
            ) as resp:
                data = await resp.json()

        # 格式化搜索结果
        results = []
        for item in data.get("RelatedTopics", [])[:5]:
            if "Text" in item:
                results.append(
                    f"- {item['FirstURL']}\n"
                    f"  {item['Text']}"
                )

        return "\n".join(results) if results \
            else "No results found."

🌐 网页浏览工具

class WebBrowserTool(BaseTool):
    """网页内容提取工具"""
    name = "web_browser"
    description = "浏览并提取网页内容"

    async def _arun(self, url: str) -> str:
        try:
            # 获取网页内容
            async with aiohttp.ClientSession() as s:
                async with s.get(url, timeout=10) as r:
                    html = await r.text()

            # 使用 BeautifulSoup 提取文本
            soup = BeautifulSoup(html, "html.parser")

            # 移除脚本和样式
            for tag in soup(
                ["script", "style", "nav", "footer"]
            ):
                tag.decompose()

            text = soup.get_text(
                separator="\n", strip=True
            )

            # 截断过长内容
            return text[:4000] if len(text) > 4000 \
                else text

        except Exception as e:
            return f"Error browsing {url}: {str(e)}"

💻 代码执行工具

class CodeExecutionTool(BaseTool):
    """安全的 Python 代码执行环境"""
    name = "code_execution"
    description = (
        "在沙箱中执行 Python 代码。"
        "可用于数学计算、数据分析。"
    )

    # 允许的模块白名单
    ALLOWED_MODULES = {
        "math", "json", "re", "datetime",
        "collections", "statistics", "itertools",
    }

    async def _arun(self, code: str) -> str:
        """在受限环境中执行代码"""
        # 限制可用模块
        restricted_globals = {
            "__builtins__": {
                k: v for k, v in __builtins__.items()
                if k in ("print", "len", "range",
                         "int", "float", "str", "list",
                         "dict", "set", "tuple", "abs",
                         "min", "max", "sum", "sorted")
            },
            **{
                m: __import__(m)
                for m in self.ALLOWED_MODULES
            },
        }
        # 执行并捕获输出
        stdout = StringIO()
        with redirect_stdout(stdout):
            exec(code, restricted_globals, {})

        return stdout.getvalue() or "Code executed."

🧠 知识图谱工具

class KnowledgeGraphTool(BaseTool):
    """基于 Pinecone 的长期记忆工具"""
    name = "knowledge_graph"
    description = (
        "存储和检索长期记忆。"
        "可用于记住重要信息和经验。"
    )

    def __init__(self, pinecone_client, embedding_model):
        super().__init__()
        self.pinecone = pinecone_client
        self.embeddings = embedding_model
        self.index_name = "agent-memory"
        self.namespace = "agent-v1"

    async def _arun(self, command: str) -> str:
        """支持 store/retrieve 命令"""
        if command.startswith("store:"):
            content = command[6:].strip()
            return await self._store(content)
        elif command.startswith("retrieve:"):
            query = command[9:].strip()
            return await self._retrieve(query)

    async def _store(self, content: str) -> str:
        vector = await self.embeddings.aembed(content)
        self.pinecone.Index(self.index_name).upsert(
            vectors=[{
                "id": str(uuid4()),
                "values": vector,
                "metadata": {"text": content}
            }],
            namespace=self.namespace,
        )
        return "Memory stored."

⚡ 工具调用流程

💾 向量记忆系统

🌲 Pinecone 集成

# services/pinecone/client.py
from pinecone import Pinecone

class MemoryStore:
    """Pinecone 向量记忆存储"""

    def __init__(self, settings: Settings):
        self.client = Pinecone(
            api_key=settings.pinecone_api_key
        )
        self.index_name = settings.pinecone_index
        self.dimension = 1536  # ada-002 维度

    async def ensure_index(self):
        """确保索引存在"""
        if self.index_name not in \
            self.client.list_indexes().names():
            self.client.create_index(
                name=self.index_name,
                dimension=self.dimension,
                metric="cosine",
                metadata_config={
                    "indexed": ["agent_id", "created_at"]
                },
            )

    async def search(
        self, query_vector: list, top_k: int = 5,
        filter_dict: dict = None
    ) -> list[MemoryRecord]:
        """语义搜索相似记忆"""
        index = self.client.Index(self.index_name)
        results = index.query(
            vector=query_vector,
            top_k=top_k,
            include_metadata=True,
            filter=filter_dict,
        )
        return [
            MemoryRecord(
                id=m["id"],
                text=m["metadata"]["text"],
                score=m["score"],
            )
            for m in results["matches"]
        ]

💾 记忆存取源码

# services/pinecone/memory.py
class AgentMemory:
    """Agent 记忆管理器"""

    def __init__(
        self,
        store: MemoryStore,
        embeddings: Embeddings,
        agent_id: str,
    ):
        self.store = store
        self.embeddings = embeddings
        self.agent_id = agent_id

    async def save(self, text: str) -> str:
        """保存记忆到向量存储"""
        vector = await self.embeddings.aembed(text)
        record_id = str(uuid4())

        self.store.upsert(
            id=record_id,
            values=vector,
            metadata={
                "text": text,
                "agent_id": self.agent_id,
                "created_at": datetime.utcnow().isoformat(),
            },
        )
        return record_id

    async def recall(
        self, query: str, top_k: int = 3
    ) -> list[str]:
        """检索相关记忆"""
        query_vector = \
            await self.embeddings.aembed(query)
        results = await self.store.search(
            query_vector,
            top_k=top_k,
            filter_dict={"agent_id": self.agent_id},
        )
        return [r.text for r in results]

🗄️ 数据库设计

📊 Prisma Schema

// prisma/schema.prisma
datasource db {
  provider = "postgresql"
  url      = env("DATABASE_URL")
}

generator client {
  provider = "prisma-client-js"
}

model User {
  id           String    @id @default(cuid())
  email        String    @unique
  name         String?
  image        String?
  emailVerified DateTime?
  accounts     Account[]
  sessions     Session[]
  agents       Agent[]
  createdAt    DateTime  @default(now())
  updatedAt    DateTime  @updatedAt
}

model Agent {
  id        String      @id @default(cuid())
  userId    String
  user      User        @relation(
    fields: [userId], references: [id]
  )
  name      String
  goal      String
  model     String      @default("gpt-4")
  settings  Json        @default("{}")
  runs      AgentRun[]
  createdAt DateTime    @default(now())
  updatedAt DateTime    @updatedAt
}

model AgentRun {
  id        String   @id @default(cuid())
  agentId   String
  agent     Agent    @relation(
    fields: [agentId], references: [id]
  )
  status    String   // running/success/failed
  steps     Json     // AgentStep[]
  result    String?
  createdAt DateTime @default(now())
}

🔐 认证与安全

# services/security.py
from passlib.context import \
    CryptContext

pwd_context = CryptContext(
    schemes=["bcrypt"],
    deprecated="auto",
)

def verify_password(
    plain: str, hashed: str
) -> bool:
    return pwd_context.verify(
        plain, hashed
    )

def hash_password(
    password: str
) -> str:
    return pwd_context.hash(password)

🌐 API 路由设计

📡 WebSocket 实时通信

# web/api/v1/agent.py
from fastapi import WebSocket, WebSocketDisconnect

@router.websocket("/ws/agent/{agent_id}")
async def agent_websocket(
    websocket: WebSocket,
    agent_id: str,
    current_user: User = Depends(
        get_current_user_ws
    ),
):
    await websocket.accept()

    try:
        # 验证 Agent 归属
        agent = await verify_agent_owner(
            agent_id, current_user.id
        )

        # 启动 Agent 并流式推送
        async for step in agent_executor.execute(
            agent.goal
        ):
            await websocket.send_json({
                "type": step.type,
                "data": {
                    "thought": step.thought,
                    "action": step.action,
                    "observation": step.observation,
                    "iteration": step.iterations,
                },
            })

    except WebSocketDisconnect:
        logger.info("Client disconnected")
    except Exception as e:
        await websocket.send_json({
            "type": "error",
            "data": {"message": str(e)},
        })

📦 前端状态管理

// stores/agentStore.ts
import { create } from 'zustand';

interface AgentState {
  agents: Agent[];
  currentAgent: Agent | null;
  isRunning: boolean;
  steps: AgentStep[];

  // Actions
  createAgent: (name: string, goal: string) 
    => Promise<void>;
  startRun: (agentId: string) => AsyncGenerator;
  stopRun: () => void;
  addStep: (step: AgentStep) => void;
}

export const useAgentStore = create<AgentState>(
  (set, get) => ({
    agents: [],
    currentAgent: null,
    isRunning: false,
    steps: [],

    createAgent: async (name, goal) => {
      const res = await fetch('/api/agents', {
        method: 'POST',
        body: JSON.stringify({ name, goal }),
      });
      const agent = await res.json();
      set((s) => ({
        agents: [...s.agents, agent]
      }));
    },

    startRun: async (agentId) => {
      set({ isRunning: true, steps: [] });
      const ws = new WebSocket(
        `/ws/agent/${agentId}`
      );
      ws.onmessage = (e) => {
        const step = JSON.parse(e.data);
        set((s) => ({
          steps: [...s.steps, step]
        }));
      };
    },
  })
);

🎨 Agent UI 组件

📝 Prompt 工程策略

🎯 System Prompt 设计

SYSTEM_PROMPT = """
You are an autonomous AI agent. Your goal is to
achieve the user's objective using available tools.

## Available Tools
{tool_descriptions}

## Instructions
1. Analyze the current context and your goal
2. Choose the most appropriate tool to use
3. Provide the tool name and input arguments
4. If you have enough information, provide
   the final answer

## Response Format
Respond ONLY with valid JSON:
{{
  "thought": "Your reasoning process",
  "tool": "tool_name or 'final'",
  "input": {{ "key": "value" }} or null,
  "output": "final answer if done"
}}

## Constraints
- You have {max_iterations} iterations
- Be concise and efficient
- Prioritize accuracy over speed
- Use tools to gather information before answering
"""

📊 Token 管理策略

# services/tokenizer/counter.py
class TokenBudget:
    """Token 预算管理器"""

    def __init__(
        self,
        model_limit: int = 8192,
        reserve_output: int = 1000,
    ):
        self.total = model_limit
        self.reserve = reserve_output
        self.available = model_limit \
            - reserve_output

    def allocate(self, parts: dict) -> dict:
        """按比例分配 Token"""
        total_input = sum(parts.values())
        return {
            k: int(
                self.available * v / total_input
            )
            for k, v in parts.items()
        }

def trim_context(
    messages: list[dict],
    max_tokens: int,
    counter: TokenCounter,
) -> list[dict]:
    """裁剪历史消息以适应 Token 限制"""
    trimmed = []
    total = 0

    # 保留 System Prompt
    trimmed.append(messages[0])
    total += counter.count(messages[0])

    # 从最新消息开始保留
    for msg in reversed(messages[1:]):
        cost = counter.count(msg)
        if total + cost > max_tokens:
            break
        trimmed.insert(1, msg)
        total += cost

    return trimmed

⚠️ 错误处理与重试

# services/retry.py
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
)

class AgentError(Exception):
    """Agent 执行错误基类"""

class LLMRateLimitError(AgentError):
    """LLM 速率限制"""

class ToolExecutionError(AgentError):
    """工具执行错误"""

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(
        min=1, max=60, multiplier=2
    ),
    retry=retry_if_exception_type(
        (LLMRateLimitError, TimeoutError)
    ),
    before_sleep=lambda _: logger.warning(
        "Retrying after rate limit..."
    ),
)
async def call_llm_with_retry(
    llm: BaseLanguageModel,
    messages: list[dict],
) -> str:
    """带重试的 LLM 调用"""
    try:
        return await llm.apredict(messages=messages)
    except RateLimitError as e:
        raise LLMRateLimitError(
            f"Rate limited: {e}"
        ) from e

🚦 并发与限流

import asyncio

class ConcurrencyLimiter:
    """Agent 并发执行限制"""

    def __init__(self, max_concurrent: int = 5):
        self.semaphore = asyncio.Semaphore(
            max_concurrent
        )
        self.active_tasks: dict[
            str, asyncio.Task
        ] = {}

    async def run_agent(
        self, agent_id: str, goal: str
    ):
        async with self.semaphore:
            task = asyncio.current_task()
            self.active_tasks[agent_id] = task
            try:
                async for step in \
                    executor.execute(goal):
                    yield step
            finally:
                del self.active_tasks[
                    agent_id
                ]

🐳 Docker 部署架构

⚙️ Docker Compose 配置

# docker-compose.yml
version: "3.9"

services:
  frontend:
    build:
      context: ./next
      dockerfile: Dockerfile
    ports:
      - "3000:3000"
    environment:
      - NEXTAUTH_SECRET=${AUTH_SECRET}
      - DATABASE_URL=${DATABASE_URL}
    depends_on:
      - postgres

  backend:
    build:
      context: ./platform
      dockerfile: Dockerfile
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=${DATABASE_URL}
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - PINECONE_API_KEY=${PINECONE_API_KEY}
      - REDIS_URL=redis://redis:6379
    depends_on:
      - postgres
      - redis

  postgres:
    image: pgvector/pgvector:pg16
    volumes:
      - pgdata:/var/lib/postgresql/data
    environment:
      POSTGRES_DB: agentgpt
      POSTGRES_PASSWORD: ${DB_PASSWORD}

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

volumes:
  pgdata:

♟️ 设计模式 - 策略模式

# 策略接口
class LLMStrategy(ABC):
    @abstractmethod
    async def generate(self, prompt: str) -> str:
        ...

# 具体策略
class OpenAIStrategy(LLMStrategy):
    async def generate(self, prompt: str) -> str:
        return await self.client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}]
        )

class AnthropicStrategy(LLMStrategy):
    async def generate(self, prompt: str) -> str:
        return await self.client.messages.create(
            model="claude-3-opus",
            messages=[{"role": "user", "content": prompt}]
        )

# 上下文
class AgentContext:
    def __init__(self, strategy: LLMStrategy):
        self._strategy = strategy

    def set_strategy(self, strategy: LLMStrategy):
        self._strategy = strategy

    async def think(self, prompt: str) -> str:
        return await self._strategy.generate(prompt)

👁️ 设计模式 - 观察者模式

class AgentEventEmitter:
    """Agent 事件发射器"""

    def __init__(self):
        self._listeners: dict[
            str, list[Callable]
        ] = {}

    def on(self, event: str, callback: Callable):
        if event not in self._listeners:
            self._listeners[event] = []
        self._listeners[event].append(callback)

    async def emit(self, event: str, data: Any):
        for callback in self._listeners.get(
            event, []
        ):
            await callback(data)

# 使用
emitter = AgentEventEmitter()

# WebSocket 订阅
emitter.on("step", websocket.send_json)
emitter.on("error", log_error)
emitter.on("complete", save_result)

# Agent 执行中发射事件
await emitter.emit("step", step_data)

🏭 设计模式 - 工厂模式

class ToolFactory:
    """工具工厂 - 根据配置创建工具"""

    _registry: dict[str, type[BaseTool]] = {
        "web_search": WebSearchTool,
        "web_browser": WebBrowserTool,
        "code_execution": CodeExecutionTool,
        "knowledge_graph": KnowledgeGraphTool,
    }

    @classmethod
    def register(
        cls, name: str, tool_class: type[BaseTool]
    ):
        """注册自定义工具"""
        cls._registry[name] = tool_class

    @classmethod
    def create_tools(
        cls, tool_names: list[str], **kwargs
    ) -> list[BaseTool]:
        """批量创建工具实例"""
        tools = []
        for name in tool_names:
            if name not in cls._registry:
                raise ValueError(
                    f"Unknown tool: {name}"
                )
            tool = cls._registry[name](**kwargs)
            tools.append(tool)
        return tools

🔄 Agent 执行流程

⚖️ 与 AutoGPT/BabyAGI 对比

⚡ 性能优化实践

🐛 常见陷阱与调试

📚 扩展阅读

📝 总结