首页 / 文章

LangGraphAI AgentsFastAPIReactSystem ArchitecturePython

AI业务团队的架构：利用LangGraph与FastAPI协调各智能体

开源多智能体平台使用指南：如何协调共同创建者、管理者和专业智能体，实现信息共享、暂停等待审批以及汇报工作。

发布者：REACTAPP.TOP阅读约 16 分钟3791 词

来源: Building a Multi-Agent Business Automation Platform with LangGraph, FastAPI, React, Neo4j, Qdrant, and Redis

目前大多数人工智能自动化系统仍采用单智能体架构：一个模型接收指令并返回答案。这种模式适用于问答场景，但实际的业务工作需要多个角色协同完成，某些步骤可并行处理，而有些则必须等待人工审批。本指南将分析一个开源平台Multi Agent for Business Automation，该平台将小型初创团队建模为相互协作的智能体。学习结束后，您应能理解其调度机制、内存管理、审批流程以及API接口是如何相互配合的，同时也能掌握哪些设计思路可应用于自己的智能体系统中。

从聊天机器人到虚拟办公室

试想一下新项目究竟是如何规划的。创始人首先明确愿景，管理者将其转化为具体的任务，财务人员核查收入与成本预估，市场团队确定产品定位，法务部门评估风险，销售团队则准备客户拓展工作。这些环节有的按顺序进行，有的同时开展，还有些则需要等待人工批准才能继续推进。

根据其README文件，该项目是一个用于自动化业务处理并辅助决策的平台，通过LangGraph工作流、共享内存以及人工干预审批机制，协调具有不同角色、特点和工具的智能体，同时还具备报告生成功能及外部系统集成能力。该项目的目标明确不是再做一个聊天机器人，而是一个让智能体能够像初创团队那样协同工作的虚拟办公室。

五个架构层次

该系统清晰地分为五层：

  • 前端：用于基于聊天的引导、工作流可视化、审批及分析的 React 控制面板。
  • 后端 API：FastAPI REST 接口、WebSocket 流、身份验证、报告生成以及内存 API。
  • 智能体层：包括联合创始人、经理、财务、市场、法律、资金管理及销售等智能体。
  • 编排层：LangGraph 工作流，以及自定义编排器、任务分配与检查点功能。
  • 内存与基础设施层：Neo4j、Qdrant、Redis 或 Upstash、Supabase，以及报告存储系统。

根据README的说明，该技术栈包括React 18与Vite、FastAPI、LangGraph以及用于智能体工作流的LangChain和CrewAI；用于存储内存和队列的Neo4j、Qdrant和Redis；还有作为可选层的Supabase，用于实现身份验证和数据持久化。

整个生命周期从上到下依次为：用户的需求 → 领导层智能体 → 专家智能体，最终形成共享内存、审批结果及输出内容。

User Vision
   ↓
Cofounder Agent
   ↓
Manager Agent
   ↓
Specialist Agents
   ├── Finance
   ├── Marketing
   ├── Legal
   ├── Money
   └── Sales
   ↓
Shared Memory + Approval Gates
   ↓
Reports + Dashboard + Integrations

从这张图表可以看出，该设计属于工作流系统而非提示词链。状态管理、内存存储、审批流程、重试机制、可观测性以及持久化输出都是需要重点考虑的因素。

五阶段工作流

第一阶段：将模糊的想法转化为结构化的上下文

一切始于用户阐述自己的需求：产品概念、商业想法或某些运营目标。联合创始人助手会收到这些信息，这些信息可能仅为一句话而已：

Input:
"I want to build an AI tool for small businesses."

它的任务是将这些信息转化为结构化的背景资料，以便团队其他成员开展工作：

Cofounder Output:
- Vision statement
- Target users
- Problem definition
- Market opportunity
- Strategic assumptions
- Initial business direction

README中将此步骤称为“愿景收集”。在后台，根据客户是需要一次性项目、持续对话还是强化型会话，可通过多种入口触发此流程：

POST /api/start-project
POST /api/conversation/start
POST /api/enhanced/start-session

无论采用哪种路径，后台都会创建项目或会话，记录初始状态，并将消息发送给相应的助手。

第二阶段：经理进行规划与任务分配

在愿景明确之后，管理代理便接管任务，并将策略拆解为可实际执行的具体工作：

Manager Responsibilities:
- Convert vision into roadmap
- Break work into functional domains
- Create agent assignments
- Define task dependencies
- Decide which agents should execute in parallel

在自定义协调器中，这一功能通过位于 backend/flows/orchestrator.py 文件中的 AgentOrchestrator.start_project() 方法实现。该方法会生成项目ID，首先运行联合创始人模块，将联合创始人的结果传递给管理代理，随后根据管理代理的分配启动各专业模块。伪代码如下：

cofounder_result = await Cofounder.execute(vision_task)
manager_result = await Manager.execute(manager_task)
specialist_results = await execute_specialists(manager_result)
generate_outputs(all_results)

管理代理属于控制层。没有它，每个专业模块都会独立基于原始想法进行工作，产生的输出彼此之间没有关联。通过将任务分解集中到同一个代理中，才能让各专业模块的成果整合成一套连贯的计划。

第三阶段：并行运行各专业模块

一旦任务确定，各职能负责人便会接手。默认的职责分工如下：

  • 联合创始人：负责制定愿景、战略及市场机会分析。
  • 经理：负责规划路线图、分解任务并进行任务分配。
  • 财务：负责收入模式、成本模型、投资回报率分析及预测。
  • 市场：负责品牌定位、营销活动策划及内容策略制定。
  • 法律：负责合规性管理、风险控制、合同审核及政策遵守。
  • 资金运作：负责定价策略、盈利模式开发及收入管理。
  • 销售：负责客户开发流程、外部拓展及销售策略实施。

每个智能体都在 backend/agents/personalities.py 中定义，且每个定义都包含其特性、沟通风格、专业领域以及特定角色所需的工具。这一点很重要：这些智能体被建模为具有自身行为、置信度阈值和任务上下文的特定角色执行者，而不仅仅是不同的系统提示。

执行工作由 _execute_specialists() 负责，它会为每个专家（金融、营销、法律与财务，以及启用时的销售）准备一项任务。每次调用智能体时都会设置60秒的超时限制：

parallel_executions = [
    asyncio.wait_for(agent.execute(task), timeout=60.0)
    for agent_name, task in specialist_tasks
]

所有这些被封装的调用随后会一起等待完成：

specialist_results = await asyncio.gather(
    *parallel_executions,
    return_exceptions=True
)

由于 asyncio.gather() 的参数 return_exceptions=True，某个任务发生超时或崩溃时，结果列表中会以异常对象的形式返回该错误，而不会取消整个任务批次。相应的代价是调用代码必须逐一检查每个结果，决定是对失败的任务进行重试、跳过还是上报。从整体设计角度来看这一做法是合理的：当财务分析永远不需要市场分析的结果，或者反之亦然时，就没有必要让它们依次执行。

第4阶段：共享内存与上下文传播

如果每个任务只能访问自己的上下文，多智能体系统的性能会迅速下降。该平台通过统一的内存层来解决这一问题，在这个内存层中，每个存储单元都有明确的职责：

Neo4j      → graph memory and relationships
Qdrant     → vector memory and semantic retrieval
Redis      → task queue, cache, runtime coordination
Local data → fallback storage and exported outputs

README中将其概括为Neo4j图结构内存、Qdrant向量搜索、Redis或Upstash队列，以及带有优雅降级机制的本地缓存。每种存储方式都用于满足不同的需求。

图结构内存用于记录代理、任务、项目与输出之间的关联关系：

Agent → executed → Task
Task → belongs_to → Project
Agent → produced → Output
Output → depends_on → PriorContext

这种结构使得能够直观展示协作过程，并精确追踪是哪个代理在何种先前上下文下产生了什么结果。

Qdrant中的向量内存支持语义检索，因此代理可以通过含义而非ID来查找之前的工作成果：

"Find previous market assumptions."
"Retrieve prior financial analysis."
"Use the earlier legal risk summary."

基于Redis的队列内存用于实现运行时的协调功能：

- Event streaming
- Task queues
- Cache
- Agent activity updates
- WebSocket event propagation

即使缺少某些服务，该项目仍能平稳运行。在本地开发时，可以用内存适配器替代Redis，同时将大语言模型调用路由到模拟提供者。这样即便没有配置所有生产环境所需的依赖，也能在笔记本电脑上运行整个系统，不过这也意味着本地测试无法发现仅在实际服务中才会出现的问题。

第5阶段：人工审核

那些可能产生重大影响的操作不应盲目执行，因此在这里人工审核是核心工作流程的一部分，而非附加环节。这些审核机制适用于可能影响以下方面的决策：

- Customer-facing campaigns
- Pricing recommendations
- Legal-sensitive outputs
- CRM updates
- Social media publishing
- Financial projections
- Business-critical recommendations

README文档介绍了这些审核流程的Slack通知机制以及可配置的超时时间。在API层面，审核功能通过如下端点实现：

GET  /api/approvals/pending
POST /api/approvals/{approval_id}/respond
GET  /api/approvals/advanced/pending
GET  /api/approvals/stats

一次审核的基本流程如下：

Agent generates action
   ↓
System evaluates confidence/risk
   ↓
Approval request is created
   ↓
Frontend or Slack notifies human
   ↓
Human approves/rejects/provides feedback
   ↓
Workflow continues or stops

其结果是一种分层自主模型：低风险任务可自动处理，而任何面向客户、涉及财务或法律敏感性的事务则需等待人工决策。如果您正在设计类似的流程控制机制，LangGraph中的路由、分发与审批模式相关介绍能更深入地阐述图层面的实现原理。

FastAPI后端内部结构

后端的入口文件backend/main.py在应用程序启动时会实例化一组全局服务：

AgentOrchestrator
Enhanced Orchestrator
ReportGenerator
SimplePredictor
AgentCollaborator
AdvancedApprovalManager
AgentService
ReportService
LangGraphOrchestrator
Queue Manager

根据README说明，所有路由均由backend/main.py负责注册；而backend/api/main.py则是更为精简的替代入口点，仅注册部分路由。代码是按功能职责来组织的：

backend/
├── main.py                 # Primary FastAPI app
├── api/                    # Route controllers
├── agents/                 # Agent implementations and personalities
├── workflows/              # LangGraph and HITL orchestrators
├── flows/                  # PRD DAG orchestrator
├── memory/                 # Graph, vector, cache managers
├── task_queue/             # Redis/Upstash queue with fallback
├── integrations/           # HubSpot, Slack, Instagram clients
├── approvals/              # Approval managers
├── collaboration/          # Cross-agent communication
├── outputs/                # Report generation
├── analytics/              # Predictions and metrics
├── auth/                   # Supabase auth
├── services/               # LLM, agent, report services
└── tools/                  # Search and tool registry

workflows/（LangGraph和HITL调度器）与flows/（由PRD驱动的DAG调度器）之间的区分值得注意：该项目同时采用了两种调度方式，虽然灵活，但需要明确每个端点使用的是哪条路径。

按组划分的API接口

身份验证

POST /api/auth/signup
POST /api/auth/signin
GET  /api/auth/user

在生产模式下，这些端点依赖Supabase进行身份验证。而在本地或演示模式下，应用可以完全跳过Supabase，转而使用轻量级的开发逻辑。README中提到的配置选项包括DEMO_MODE、Supabase相关变量以及LLM提供商相关变量。对应的处理函数如下：

signup()          → creates user account
signin()          → authenticates user
get_current_user()→ verifies token and returns user profile

项目生命周期

GET  /api/projects
POST /api/projects
POST /api/start-project
POST /api/auto-execute
GET  /api/auto-execute/status

这些对应以下函数：

get_user_projects()       → fetch projects for authenticated user
create_project()          → create new project record
start_project()           → execute Cofounder → Manager → Specialists workflow
auto_execute_project()    → start automated coordination from raw vision
get_auto_execution_status() → return execution progress and logs

start_project()是该组的功能核心，其内部执行流程如下：

1. Generate project_id
2. Execute Cofounder Agent
3. Log Cofounder result
4. Execute Manager Agent with Cofounder context
5. Log Manager result
6. Execute specialist agents in parallel
7. Export and persist final outputs
8. Return project status and timeline

该流程与协调器模块中的AgentOrchestrator.start_project()以及_execute_specialists()直接对应。

增强型会话

POST /api/enhanced/start-session
POST /api/enhanced/continue-session
POST /api/enhanced/approve-and-execute
GET  /api/enhanced/session-status
GET  /api/enhanced/live-logs
GET  /api/enhanced/session-results
GET  /api/enhanced/system-metrics
POST /api/enhanced/cancel-session
POST /api/enhanced/cleanup-sessions

每个接口都有明确的用途：

start-session        → create enhanced automation session
continue-session     → continue existing multi-turn planning session
approve-and-execute  → approve captured vision and run workflow
session-status       → inspect session state
live-logs            → stream logs for UI monitoring
session-results      → return final session output
system-metrics       → expose high-level runtime metrics
cancel-session       → stop active session
cleanup-sessions     → remove old sessions

这是更偏向实际生产环境的处理方式，因为它将对话、审批、执行、监控和结果获取等操作拆分为独立的调用。客户端可以轮询状态或实时获取日志，而无需保持长时间开放的请求；同时也可以明确地取消或清理会话。

代理管理

GET  /api/agents/status
GET  /api/agents/list
GET  /api/agents/personalities
GET  /api/agents/configs
POST /api/agents/configs
POST /api/agents/execute
GET  /api/agents/logs/live

各路由的功能如下：

agents/status       → current status of all agents
agents/list         → available agent inventory
agents/personalities→ UI-visible personality metadata
agents/configs      → current agent runtime settings
update configs      → update temperature, approval mode, priority, enabled flag
agents/execute      → execute a single agent manually
logs/live           → recent agent execution logs

当调用 /api/agents/list 时，返回的列表会按战略、业务、运营、专业四个类别进行分类；而运行时配置接口则允许操作员调整温度参数、审批模式、优先级以及是否启用某个智能体。

对话

POST /api/conversation/start
POST /api/conversation/{conversation_id}/message
POST /api/conversation/{conversation_id}/approve

其背后的处理程序为：

start_conversation()     → starts a Cofounder-led discovery conversation
continue_conversation()  → continues the existing conversation with stored context
approve_conversation()   → approves the captured vision and starts task distribution

设置此流程是为了让系统在投入精力处理专业智能体之前先收集足够的详细信息。其内部处理流程如下：

1. User sends initial idea
2. Cofounder Agent asks clarifying questions or structures the vision
3. Conversation state is persisted
4. System detects whether vision is ready for approval
5. User approves
6. System starts agent coordination

在协调器中，start_conversation() 会分配一个新的对话 ID，保存用户输入的内容，调用联合创始人智能体，持久化存储回复内容，最后返回一个 ready_for_approval 标志，告知客户端该方案是否已完善到可以批准的程度。

LangGraph 工作流

POST /api/workflow/execute
POST /api/workflow/resume/{thread_id}

相应的函数：

execute_langgraph_workflow() → runs a LangGraph-based workflow
resume_workflow()            → resumes workflow from checkpoint

这些端点直接暴露了LangGraph层。README中将其描述为会保存进度检查点的LangGraph状态机，这类状态机能够自我修正，并包含供人工干预的中断节点。可恢复性是其核心特性：当一个长时间运行的工作流因等待审批而暂停或在中途失败时，它应该从上一个检查点继续执行，而非重新开始并再次支付每次LLM调用的费用。我们通过了解LangGraph的InMemorySaver如何存储检查点，可以明白每一步实际上会保存什么内容。

报告

GET  /api/reports/comprehensive
GET  /api/reports/{report_type}
POST /api/reports/generate-pdf
GET  /api/reports/download/{filename}
GET  /api/reports/domains
GET  /api/reports/domains/{domain}

每个报告路由生成的输出内容：

comprehensive report → generate full business report
specific report      → generate executive, marketing, financial, etc.
generate PDF         → convert report data into downloadable PDF
download report      → serve generated PDF/HTML
domain reports       → return modular reports by business domain

报告生成是该系统的一项核心功能：可生成面向管理层、市场部、财务部门的各类报告，以及以JSON格式呈现的全面报告，PDF版本则通过WeasyPrint生成。这正是该系统与聊天机器人的区别所在，因为最终输出的是业务相关文档而非聊天消息。

内存

GET    /api/memory/graph
GET    /api/memory/stats
POST   /api/memory/export
DELETE /api/memory/clear

它们返回的内容为：

memory/graph  → return graph nodes and edges for visualization
memory/stats  → return vector/graph memory statistics
memory/export → export stored memory and outputs
memory/clear  → clear memory stores

README文件将这些功能归类在/api/memory/*路径下，用于导出图表及统计信息。通过这些功能，你可以查看智能体积累的知识、追踪其输出内容，并了解每个内存后端所占用的存储空间。需要注意的是，清除数据的接口具有破坏性，因此应与其他管理操作一样设置相应的访问控制。

分析功能

GET /api/analytics/predictions
GET /api/enhanced/system-metrics
GET /api/communication/stats

它们的作用是：

predictions          → analyze generated outputs and estimate success/revenue/market timing
system metrics       → return enhanced orchestrator metrics
communication stats  → return inter-agent communication analytics

在这些功能背后，是一些简单的辅助函数：

_analyze_project_success()
_analyze_revenue_trend()
_analyze_market_timing()

这些工具会分析智能体的输出，进而得出面向业务的指标，如成功概率、收入增长趋势以及建议的操作时机。它们是基于生成文本的启发式方法，因此其输出应被视为参考信息而非精确预测。

集成方案

/api/integrations/*

该仓库列出了HubSpot、Slack和Instagram Business API，每种工具都有特定的功能角色：

HubSpot   → CRM workflows
Slack     → HITL approval notifications
Instagram → marketing automation with compliance checks

核心理念是智能体应将结果推送到真实的业务系统中，而不仅仅是生成内容；同时，任何敏感信息都需要经过审批机制才能传递。

WebSockets与流式传输

WS  /ws/agent-updates
WS  /ws/agent-events
GET /api/stream/logs

每条数据流所承载的内容：

/ws/agent-updates → periodically sends agent status updates
/ws/agent-events  → streams queue/Redis events to frontend
/api/stream/logs  → streams recent logs via server-sent response

对于多智能体用户界面而言，这种可见性至关重要。操作员需要知道哪个智能体正在运行、哪项任务处于活跃状态、是否有审批待处理，以及工作流在何处出错。

React前端

客户端基于以下技术栈构建：

React 18
Vite
React Router
Tailwind CSS
React Flow
Recharts

根据README说明，该控制面板整合了基于聊天的引导流程、智能体实时状态显示、任务流程的可视化地图，以及用于检查PRD合规性的视图。其功能职责包括：

1. Capture the user’s project vision
2. Display agent execution status
3. Visualize task flow
4. Show pending approval requests
5. Display reports and analytics
6. Show memory/monitoring data
7. Connect to WebSocket streams
8. Communicate with FastAPI through api.js

源代码的布局较为紧凑：

frontend/
└── src/
    ├── App.jsx
    ├── pages/
    ├── components/
    └── services/api.js

App.jsx作为经过身份验证的界面外壳，pages/存放工作流程、分析及监控相关视图，而components/则提供各种构建模块：仪表板面板、审批界面、PRD合规性查看功能以及集成设置。所有HTTP请求都会经过services/api.js处理，该文件统一管理重试和缓存机制，从而避免组件层面出现网络相关问题。

端到端流程

整体而言，系统中的主要处理路径如下：

1. User submits business vision from React UI
2. Frontend sends request to FastAPI
3. FastAPI creates project/session
4. Cofounder Agent structures vision
5. Manager Agent converts vision into roadmap and assignments
6. Specialist agents execute assigned tasks
7. Results are stored in memory
8. HITL approval is requested where required
9. Reports are generated
10. Frontend displays logs, status, graph, reports, and outputs

对应到具体的模块和方法后，同样的流程呈现为如下形式：

React UI
   ↓ HTTP/WebSocket
FastAPI main.py
   ↓
AgentOrchestrator / EnhancedOrchestrator
   ↓
CofounderAgent.execute()
   ↓
ManagerAgent.execute()
   ↓
_execute_specialists()
   ↓
FinanceAgent / MarketingAgent / LegalAgent / MoneyAgent / SalesAgent
   ↓
MemoryManager + GraphMemory + VectorMemory
   ↓
ReportGenerator
   ↓
Dashboard + PDF/HTML/JSON Output

每一层都有其特定职责：用户界面负责处理交互操作，协调器负责统筹管理，执行节点负责实际执行任务，内存管理器负责保存上下文信息，而报告生成器则负责整理输出结果。

为何选择图式协调器而非链式结构

多智能体工作流需要的不仅仅是系列函数调用，它们还需要：

- State transitions
- Conditional routing
- Checkpointing
- Retry behavior
- Human approval interruptions
- Recovery after failure
- Resume from previous state

README中将LangGraph的状态机称为协调层，正是基于这些功能。普通的提示词链是线性的：

A → B → C → Done

业务工作流会根据中间结果进行分支：

A → B
    ├── if finance confidence low → retry Finance
    ├── if legal risk high → ask human approval
    ├── if marketing ready → generate campaign
    └── if all complete → generate report

金融风险较低时触发重试，法律风险较高时则需人工审核，最终报告要等到所有分支处理完毕才能生成。若用硬编码的顺序调用来实现这些逻辑，很快就会变成一堆复杂的条件判断；而具有明确节点、边和检查点的图结构则能让逻辑更清晰且可恢复。

该设计的优势

  • 明确的智能体边界：每个智能体负责一个业务领域，这既能保证输出的一致性，也能便于新增角色。
  • 以工作流为驱动的协调机制：执行过程被建模为工作流，而非一系列提示词响应的流水线。
  • 并行处理的专家系统：在存在依赖关系的情况下，财务、营销、法律、资金管理及销售等功能可同时运行，从而缩短整体处理时间。
  • 共享内存机制：Neo4j、Qdrant、Redis以及本地存储共同实现关系管理、语义检索、队列处理、缓存功能及备用方案。
  • 人工干预的安全保障：审批机制确保该系统可用于需要承担实际后果的决策场景。
  • 实时可观测性：WebSockets、日志记录、时间线展示、控制面板及分析功能可实时呈现智能体的工作状态。
  • 结构化报告：输出的是可直接交给相关方的文档，而非聊天记录。
  • 集成准备就绪：对HubSpot、Slack和Instagram的支持表明该系统旨在与实际业务工具对接。

    • 为生产环境做强化

    基础已经十分扎实：模块化后端、专用代理、内存抽象层、审批流程以及可观察的前端界面。接下来的重点显然在于可靠性、安全性、监控、评估和成本控制：

    1. Add distributed tracing with OpenTelemetry
2. Add LangSmith or custom LLM evaluation dashboards
3. Add cost tracking per agent and per workflow
4. Add stronger RBAC for users, agents, and approvals
5. Add persistent workflow checkpoints in Postgres or Redis
6. Add retry policies per agent type
7. Add queue-based background execution with Celery/RQ/Arq
8. Add structured output validation with Pydantic schemas
9. Add agent-level unit tests and golden-output evaluation
10. Add Docker production profiles and deployment manifests

    其中有两点值得重点强调。按代理和流程跟踪成本十分重要，因为并行工作的专家会大幅增加大语言模型的调用次数，而成本在未被衡量之前是看不见的。使用 Pydantic 模式进行结构化输出验证也很关键，因为从内存写入到报告生成及 CRM 更新的每一个下游步骤，其可靠性都取决于代理返回的数据结构。

    核心要点

    • 将编排层视为最终产品，大语言模型只是其中的其中一个组件。
    • 安排一个管理型代理负责任务分解，以确保各专家的输出保持一致性。
    • 让多个独立代理同时运行，并为每个代理设置超时时间，同时明确处理部分失败的情况。
    • 为每个内存存储分配特定功能：用图结构记录数据来源，用向量实现检索，用队列进行协调。
  • 应将审批关卡、检查点以及实时可观测性纳入核心设计之中，而非事后添加。
    • ← 全部文章发送简报