Kortix开源AI Agent平台深度解析：企业级Agent开发的完整解决方案

frontend/src/
├── app/                    # Next.js应用路由
├── components/
│   ├── Thread/            # 会话管理组件
│   │   └── ThreadContent.tsx
│   ├── Agent/             # Agent配置组件
│   │   └── AgentConfiguration.tsx
│   └── Chat/              # 聊天界面组件
│       └── ChatInput.tsx

backend/
├── api.py                 # 主API应用
├── agent/
│   ├── api.py            # Agent管理路由
│   └── run_agent_background.py  # 后台执行
├── thread/
│   └── api.py            # 线程和消息路由
└── sandbox/
    └── api.py            # 沙盒管理路由

backend/agentpress/
├── thread_manager.py      # ThreadManager类
├── response_processor.py  # ResponseProcessor类
└── tool_view_registry.py  # ToolViewRegistry类

backend/
├── services/
│   ├── supabase.py       # Supabase数据库服务
│   └── redis.py          # Redis缓存和消息服务
├── sandbox/
│   └── sandbox.py        # Docker沙盒服务
└── tools/
    ├── web_tools.py      # Web搜索和抓取工具
    ├── file_tools.py     # 文件操作工具
    └── sandbox_tools.py  # 沙盒执行工具

class ThreadManager:
    """管理Agent对话线程和执行流程的核心类"""
    
    def __init__(self, db: DBConnection, tool_registry: ToolRegistry, 
                 instance_id: str = "default", trace: Optional[StatefulTraceClient] = None):
        self.db = db
        self.tool_registry = tool_registry
        self.instance_id = instance_id
        self.trace = trace or langfuse.trace(name="anonymous:thread_manager")
        self.response_processor = ResponseProcessor(
            tool_registry=tool_registry,
            add_message_callback=self.add_message,
            trace=self.trace
        )

async def run_thread(self, thread_id: str, user_message: str, 
                    agent_config: dict, stream: bool = True) -> AsyncGenerator:
    """执行Agent对话的主入口方法"""
    # 1. 添加用户消息到数据库
    await self.add_message(thread_id, "user", user_message)
    
    # 2. 获取LLM消息历史
    llm_messages = await self.get_llm_messages(thread_id, agent_config)
    
    # 3. 调用LLM获取响应
    llm_response = await self._call_llm(llm_messages, agent_config, stream)
    
    # 4. 处理响应和工具调用
    async for result in self.response_processor.process_streaming_response(
        llm_response, thread_id, llm_messages, agent_config.get('model')
    ):
        yield result

async def get_llm_messages(self, thread_id: str, agent_config: dict) -> List[Dict]:
    """获取并格式化LLM消息历史"""
    # 获取数据库中的消息
    messages = await self._fetch_thread_messages(thread_id)
    
    # 构建系统提示
    system_prompt = self._build_system_prompt(agent_config)
    
    # 格式化为LLM可理解的格式
    return self._format_messages_for_llm(messages, system_prompt)

class ResponseProcessor:
    """LLM响应处理和工具执行的核心引擎"""
    
    async def process_streaming_response(
        self, llm_response: AsyncGenerator, thread_id: str,
        prompt_messages: List[Dict], llm_model: str,
        config: ProcessorConfig = ProcessorConfig()
    ) -> AsyncGenerator[Dict[str, Any], None]:
        """处理流式LLM响应的主方法"""
        
        accumulated_content = ""
        tool_calls_buffer = {}
        
        async for chunk in llm_response:
            # 1. 提取内容块
            if hasattr(chunk, 'choices') and chunk.choices:
                delta = chunk.choices[0].delta
                if delta and hasattr(delta, 'content') and delta.content:
                    chunk_content = delta.content
                    accumulated_content += chunk_content
                    
                    # 2. 实时检测XML工具调用
                    if config.xml_tool_calling:
                        xml_chunks = self._extract_xml_chunks(accumulated_content)
                        for xml_chunk in xml_chunks:
                            tool_call = self._parse_xml_tool_call(xml_chunk)
                            if tool_call and config.execute_on_stream:
                                # 立即执行工具
                                await self._execute_tool_call(tool_call, thread_id)
                    
                    # 3. 流式输出内容
                    yield self._format_content_chunk(chunk_content, thread_id)

def _extract_xml_chunks(self, content: str) -> List[str]:
    """从内容中提取完整的XML工具调用块"""
    pattern = r'<(\w+)>.*?</\1>'
    return re.findall(pattern, content, re.DOTALL)

async def _execute_tool_call(self, tool_call: Dict, thread_id: str) -> ToolResult:
    """执行单个工具调用"""
    tool_name = tool_call.get('name')
    tool_args = tool_call.get('arguments', {})
    
    # 从工具注册表获取工具实例
    tool_info = self.tool_registry.get_tool(tool_name)
    if not tool_info:
        return ToolResult(success=False, output=f"Tool {tool_name} not found")
    
    # 执行工具方法
    tool_instance = tool_info['instance']
    tool_function = getattr(tool_instance, tool_name)
    
    try:
        result = await tool_function(**tool_args)
        return ToolResult(success=True, output=str(result))
    except Exception as e:
        return ToolResult(success=False, output=f"Tool execution failed: {str(e)}")

class ToolRegistry:
    """工具注册表，管理所有可用工具"""
    
    def __init__(self):
        self.tools = {}  # 存储工具实例和schema
    
    def register_tool(self, tool_class: Type[Tool], 
                     function_names: Optional[List[str]] = None, **kwargs):
        """注册工具类到注册表"""
        tool_instance = tool_class(**kwargs)
        schemas = tool_instance.get_schemas()
        
        # 注册OpenAPI格式的工具函数
        for func_name, schema_list in schemas.items():
            if function_names is None or func_name in function_names:
                for schema in schema_list:
                    if schema.schema_type == SchemaType.OPENAPI:
                        self.tools[func_name] = {
                            "instance": tool_instance,
                            "schema": schema
                        }
    
    def get_openapi_schemas(self) -> List[Dict[str, Any]]:
        """获取所有工具的OpenAPI schema定义"""
        return [
            tool_info['schema'].schema 
            for tool_info in self.tools.values()
            if tool_info['schema'].schema_type == SchemaType.OPENAPI
        ]

class Tool(ABC):
    """所有工具的抽象基类"""
    
    def __init__(self):
        self._schemas: Dict[str, List[ToolSchema]] = {}
        self._register_schemas()  # 自动注册装饰器定义的schema
    
    def success_response(self, data: Union[Dict, str]) -> ToolResult:
        """创建成功响应"""
        return ToolResult(success=True, output=json.dumps(data, indent=2))
    
    def fail_response(self, msg: str) -> ToolResult:
        """创建失败响应"""
        return ToolResult(success=False, output=msg)

# 工具装饰器
@openapi_schema({
    "type": "function",
    "function": {
        "name": "search_web",
        "description": "Search the web for information",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query"}
            },
            "required": ["query"]
        }
    }
})
async def search_web(self, query: str) -> ToolResult:
    """示例工具方法"""
    # 工具实现逻辑
    return self.success_response({"results": "search results"})

# 应用生命周期管理
@asynccontextmanager
async def lifespan(app: FastAPI):
    """管理应用启动和关闭流程"""
    # 启动时初始化
    await db.initialize()
    agent_api.initialize(db, instance_id)
    sandbox_api.initialize(db)
    await redis.initialize_async()
    
    yield  # 应用运行期间
    
    # 关闭时清理资源
    await agent_api.cleanup()
    await redis.close()
    await db.disconnect()

# 模块化路由设计
api_router = APIRouter()
api_router.include_router(agent_api.router)        # Agent管理
api_router.include_router(sandbox_api.router)      # 沙盒执行
api_router.include_router(mcp_api.router)          # MCP协议支持
api_router.include_router(credentials_api.router)  # 凭证管理
api_router.include_router(template_api.router)     # 模板系统

# 从LLM响应到前端显示的完整流式链路
LLM Response Stream → ResponseProcessor → ThreadManager → FastAPI → WebSocket → Frontend

@openapi_schema({...})  # 自动生成OpenAPI schema
@usage_example("...")   # 提供使用示例
async def tool_function(self, param: str) -> ToolResult:
    return self.success_response(result)

# 支持OpenAI、Anthropic、Google等多种模型
response = await litellm.completion(
    model="gpt-4o",  # 或 "claude-3-sonnet", "gemini-pro"
    messages=messages,
    stream=True
)

# backend/tools/目录结构
tools/
├── sandbox_tools.py      # shell_command, read_file
├── web_tools.py          # tavily_search, firecrawl_scrape  
├── file_tools.py         # create_file, edit_file
├── browser_tools.py      # 浏览器自动化工具
└── integration_tools.py  # MCP & Composio连接器

# backend/agentpress/response_processor.py
class ResponseProcessor:
    async def process_stream(self, llm_response_stream):
        async for chunk in llm_response_stream:
            # 解析工具调用
            tool_calls = self.extract_tool_calls(chunk)
            
            for tool_call in tool_calls:
                # 执行工具并获取结果
                result = await self.execute_tool(tool_call)
                
                # 发布实时更新
                await self.publish_tool_result(result)

agent_name: "销售数据分析师"
tools:
  - web_search: "市场趋势搜索"
  - file_management: "报告生成"
  - data_processing: "数据清洗和分析"
  - visualization: "图表生成"

agent_name: "智能客服助手"
tools:
  - knowledge_search: "知识库检索"
  - ticket_management: "工单处理"
  - user_verification: "身份验证"
  - escalation: "人工转接"

agent_name: "DevOps自动化助手"
tools:
  - shell_execution: "命令执行"
  - git_operations: "代码管理"
  - monitoring: "系统监控"
  - notification: "告警通知"

# 启动基础服务
docker-compose up -d postgres redis rabbitmq

# 验证服务状态
docker-compose ps

git clone https://github.com/kortix-ai/suna.git
cd suna

cd backend

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Linux/Mac
# 或 venv\Scripts\activate  # Windows

# 安装依赖
pip install -r requirements.txt

# 配置环境变量
cp .env.example .env
# 编辑 .env 文件，配置数据库和API密钥

# 运行数据库迁移
alembic upgrade head

# 启动后端服务
uvicorn main:app --reload --port 8000

cd ../frontend

# 安装依赖
npm install

# 配置环境变量
cp .env.example .env.local
# 编辑 .env.local 文件

# 启动开发服务器
npm run dev

cd ../backend

# 启动Agent工作进程
dramatiq suna.agent.run

# 启动工具服务
python -m suna.tools.server

# 数据库配置
DATABASE_URL=postgresql://user:password@localhost:5432/suna

# Redis配置
REDIS_URL=redis://localhost:6379

# RabbitMQ配置
RABBITMQ_URL=amqp://guest:guest@localhost:5672

# LLM API配置
OPENAI_API_KEY=your_openai_key
ANTHROPIC_API_KEY=your_anthropic_key
GOOGLE_API_KEY=your_google_key

# 安全配置
JWT_SECRET_KEY=your_jwt_secret
ENCRYPTION_KEY=your_encryption_key

# 沙盒配置
SANDBOX_ENABLED=true
DOCKER_HOST=unix:///var/run/docker.sock

http://localhost:3000

{
  "name": "数据分析助手",
  "description": "专业的数据分析和可视化Agent",
  "model": "claude-3-5-sonnet-20241022",
  "tools": [
    "web_search",
    "file_management",
    "python_execution",
    "data_visualization"
  ],
  "system_prompt": "你是一个专业的数据分析师..."
}

# 通过API测试
curl -X POST http://localhost:8000/api/v1/agents/chat \
  -H "Content-Type: application/json" \
  -d '{
    "agent_id": "your_agent_id",
    "message": "请分析这个CSV文件的销售趋势",
    "attachments": ["sales_data.csv"]
  }'

# 检查PostgreSQL状态
docker-compose logs postgres

# 重置数据库
docker-compose down
docker volume rm suna_postgres_data
docker-compose up -d postgres

# 检查RabbitMQ队列
docker-compose exec rabbitmq rabbitmqctl list_queues

# 重启Agent工作进程
pkill -f "dramatiq suna.agent.run"
dramatiq suna.agent.run

# 检查Docker权限
sudo usermod -aG docker $USER
newgrp docker

# 验证Docker访问
docker ps