项目实训实验报告2
·
目录
一、项目概述与目标
项目简介
AI Social Scheduler 是一个基于 LangGraph 构建的智能社交媒体运营调度系统，旨在实现内容创作、发布、互动和数据分析的全流程智能化。系统通过统一的调度核心自动化管理多个社交媒体平台，支持小红书等平台的运营需求。
技术架构
┌─────────────────────────────────────────────────────────────┐
│ 用户交互层 │
│ (FastAPI/Web/CLI) │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ API 层 │
│ (流式 API、RESTful API) │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ 图执行层 │
│ (LangGraph - 路由、节点编排、状态管理) │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Router │ │ Graph │ │ State │ │
│ │ System │→ │ Builder │→ │ Manager │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ 智能体层 │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ XHS Agent │ │ Content │ │ Publish │ │
│ │ │ │ Agent │ │ Agent │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ MCP 服务层 │
│ (小红书浏览器自动化、图视频生成、内容生成服务) │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ 平台适配层 │
│ (浏览器自动化/API调用) │
└─────────────────────────────────────────────────────────────┘
本周开发目
- 设计核心架构：基于 LangGraph 的分层模块化架构设计
- 实现基础组件：工作流引擎、路由系统、状态管理
- 开发小红书工作流：内容生成 → 图片生成 → 发布流程
- 实现流式 API：支持实时进度展示
- 编写基础测试：确保核心功能的稳定性
二、开发环境搭建
环境要求
- Python >= 3.11
- uv (Python包管理器)
- LangGraph >= 0.3.0
- FastAPI >= 0.100.0
- LangChain >= 0.2.0
项目初始化
### 依赖安装
```bash
# 使用 uv 安装依赖
uv sync
# 验证安装
uv run python -c "import langgraph; print('LangGraph version:', langgraph.__version__)"
pyproject.toml 配置示例：
[tool.poetry]
name = "ai-social-scheduler"
version = "0.1.0"
description = "AI-driven social media scheduling system"
authors = ["Developer"]
[tool.poetry.dependencies]
python = "^3.11"
langgraph = "^0.3.0"
langchain = "^0.2.0"
langchain-core = "^0.2.0"
langchain-community = "^0.2.0"
fastapi = "^0.115.0"
uvicorn = { extras = ["standard"], version = "^0.32.0" }
pydantic = "^2.9.0"
pydantic-settings = "^2.6.0"
python-dotenv = "^1.0.0"
httpx = "^0.27.0"
sse-starlette = "^2.1.0"
[tool.poetry.group.dev.dependencies]
pytest = "^8.3.0"
pytest-asyncio = "^0.24.0"
三、核心架构设计
架构层次
系统采用分层模块化架构，各层职责明确：
- 用户交互层：提供 FastAPI Web 接口和 CLI 接口
- API 层：处理请求验证、响应格式化、流式处理
- 图执行层：基于 LangGraph 的工作流编排和状态管理
- 智能体层：各平台特定的 AI Agent 实现
- MCP 服务层：统一的外部服务接口
- 平台适配层：具体的平台集成实现
核心组件
| 组件 | 职责 | 关键类 |
|---|---|---|
| Graph Builder | 动态构建 LangGraph | GraphBuilder |
| Router System | 意图分析和路由决策 | RouterSystem |
| State Manager | 会话状态和任务状态管理 | StateManager |
| Node Registry | 节点注册和发现 | NodeRegistry |
| Orchestrator | 任务调度和执行 | Orchestrator |
| Streaming Executor | 流式执行和进度推送 | StreamingGraphExecutor |
数据流设计
用户输入
↓
[RouterSystem] → 意图分析 → 路由决策
↓
[Orchestrator] → 任务创建 → 任务入队
↓
[GraphBuilder] → 图构建 → 节点编排
↓
[各Agent节点] → 任务执行 → 结果输出
↓
[StreamingExecutor] → SSE 事件 → 前端展示
四、核心功能实现
1. 基于 LangGraph 的工作流引擎
状态管理设计
LangGraph 需要一个 TypedDict 来定义图状态，这是整个工作流的基础。状态定义如下：
from typing import TypedDict, Optional, Any, List
from langchain_core.messages import BaseMessage
class GraphState(TypedDict, total=False):
"""图状态定义
LangGraph 需要一个 TypedDict 来定义状态，包含：
- messages: 消息历史（LangChain格式）
- task: 当前任务
- thread_id: 会话 ID
- user_id: 用户 ID
- iteration_count: 迭代计数
- metadata: 元数据
"""
messages: List[BaseMessage]
task: Optional[Task]
thread_id: str
user_id: str
iteration_count: int
metadata: Dict[str, Any]
设计要点：
total=False表示所有字段都是可选的messages存储完整的对话历史，支持多轮交互task引用当前执行的任务对象iteration_count用于防止无限循环metadata提供扩展性
图构建器实现
图构建器是 LangGraph 工作流的核心，负责动态构建图结构：
class GraphBuilder:
"""图构建器 - 动态构建 LangGraph
核心职责：
1. 基于节点配置构建图
2. 添加节点和边
3. 编译图
4. 集成路由和调度
"""
def __init__(
self,
orchestrator: Orchestrator,
node_registry: Optional[NodeRegistry] = None,
node_factory: Optional[NodeFactory] = None,
checkpointer: Optional[MemorySaver] = None,
):
self.orchestrator = orchestrator
self.node_registry = node_registry or NodeRegistry()
self.node_factory = node_factory or NodeFactory(registry=self.node_registry)
self.checkpointer = checkpointer or MemorySaver()
async def build(self) -> StateGraph:
"""构建状态图（异步版本）
图结构：
START
↓
[router] ← 入口节点，负责意图分析和路由决策
↓
{条件路由} ← 根据任务目标节点路由
↓
┌─────┴─────┬─────────┐
↓ ↓ ↓
[xhs_workflow] [wait] [END]
↓ ↑
└───────────┘ ← 完成后回到 router
"""
workflow = StateGraph(GraphState)
# 1. 添加路由节点
workflow.add_node("router", self._create_router_node())
# 2. 添加子图作为节点
try:
from ..subgraphs import XHSWorkflowSubgraph
xhs_subgraph = XHSWorkflowSubgraph()
await xhs_subgraph.initialize()
# 直接添加编译后的子图
workflow.add_node("xhs_workflow", xhs_subgraph.graph)
except Exception as e:
logger.error(f"Failed to add xhs_workflow subgraph: {e}")
# 3. 添加等待节点
workflow.add_node("wait", self._create_wait_node())
# 4. 定义边关系
workflow.add_edge(START, "router")
# 条件路由
workflow.add_conditional_edges(
"router",
self._route_from_router,
route_map
)
# 所有节点完成后回到 router
for node_id in successfully_added_nodes:
workflow.add_edge(node_id, "router")
return workflow
路由节点的核心逻辑：
def _create_router_node(self):
async def router_node(state: GraphState) -> dict[str, Any]:
"""路由节点：分析意图并决定下一步"""
messages = state.get("messages", [])
current_task = state.get("task")
# 获取最后的用户消息
user_input = self._extract_user_input(messages)
# 检查是否是新的用户输入
is_new_input = self._check_new_input(messages, current_task)
# 如果已有任务且未完成，并且没有新输入，继续执行
if current_task and not current_task.is_terminal() and not is_new_input:
return {"task": current_task}
# 提交新任务到调度器
task = await self.orchestrator.submit(
user_input=user_input,
context={
"thread_id": state.get("thread_id", ""),
"user_id": state.get("user_id", ""),
},
messages=messages,
)
# 添加响应消息
response = task.metadata.get("response", "")
if response:
messages.append(AIMessage(content=response))
return {
"task": task,
"messages": messages,
"iteration_count": state.get("iteration_count", 0) + 1,
}
return router_node
2. 智能路由系统
路由系统架构
路由系统采用混合策略，结合规则引擎和 LLM 分析：
class RouterSystem:
"""路由系统 - 混合策略的智能路由
支持的路由策略：
1. RULE_FIRST: 先尝试规则匹配，失败则使用LLM（推荐）
2. LLM_FIRST: 先使用LLM分析，失败则尝试规则
3. HYBRID: 同时使用规则和LLM，结合结果
4. LLM_ONLY: 只使用LLM
5. RULE_ONLY: 只使用规则
"""
def __init__(
self,
routes: Optional[list[RouteConfig]] = None,
strategy: str = RouterStrategy.RULE_FIRST,
llm_model: str = "qwen-plus",
llm_temperature: float = 0.3,
enable_llm: bool = True,
available_nodes: Optional[list[str]] = None,
):
self.strategy = strategy
self.enable_llm = enable_llm
self.available_nodes = available_nodes or []
# 初始化规则引擎
self.rule_engine = RuleEngine(routes=routes)
# 初始化意图分析器
if enable_llm:
self.intent_analyzer = IntentAnalyzer(
llm_model=llm_model,
temperature=llm_temperature,
available_nodes=self.available_nodes,
)
规则优先策略的实现：
async def _route_rule_first(
self,
user_input: str,
context: Optional[dict[str, Any]],
messages: Optional[list[BaseMessage]],
) -> RouteDecision:
"""规则优先策略"""
# 1. 先尝试规则匹配
decision = self.rule_engine.match(user_input, context)
if decision and decision.target_nodes:
# 规则匹配成功，使用 LLM 生成回复
if self.intent_analyzer:
llm_decision = await self.intent_analyzer.analyze(
user_input, context, messages
)
# 合并决策：使用规则的路由 + LLM 的回复
decision.response = llm_decision.response
decision.extracted_params.update(llm_decision.extracted_params)
return decision
# 2. 规则未匹配，使用 LLM
if self.intent_analyzer:
return await self.intent_analyzer.analyze(user_input, context, messages)
# 3. 都不可用，返回降级决策
return self._create_fallback_decision("No routing method available")
规则引擎
规则引擎基于关键词匹配和模式识别：
class RuleEngine:
"""规则引擎 - 基于关键词和模式的快速路由"""
def __init__(self, routes: Optional[list[RouteConfig]] = None):
self.routes = routes or []
self._build_matcher()
def match(self, user_input: str, context: Optional[dict]) -> Optional[RouteDecision]:
"""匹配用户输入"""
user_input_lower = user_input.lower()
for route in self.routes:
# 检查关键词
if route.match(user_input_lower, context):
return self._create_decision(route)
return None
def _build_matcher(self):
"""构建匹配器"""
self.matchers = []
for route in self.routes:
keywords = route.keywords or []
patterns = route.patterns or []
matcher = {
"route": route,
"keywords": [k.lower() for k in keywords],
"patterns": patterns,
}
self.matchers.append(matcher)
意图分析器
意图分析器使用 LLM 进行复杂意图识别：
class IntentAnalyzer:
"""意图分析器 - LLM 驱动的意图识别
使用 Pydantic 模型约束 LLM 输出：
"""
def __init__(
self,
llm_model: str = "qwen-plus",
temperature: float = 0.3,
system_prompt: Optional[str] = None,
available_nodes: Optional[list[str]] = None,
):
self.llm_model = llm_model
self.temperature = temperature
self.available_nodes = available_nodes or []
# 初始化 LLM 客户端
self._llm = None
self._structured_llm = None
# 系统提示词
self.system_prompt = system_prompt or self._build_default_prompt()
@property
def structured_llm(self):
"""获取结构化输出的 LLM"""
if self._structured_llm is None:
self._structured_llm = self.llm.with_structured_output(IntentAnalysisOutput)
return self._structured_llm
async def analyze(
self,
user_input: str,
context: Optional[dict[str, Any]] = None,
messages: Optional[list[BaseMessage]] = None,
) -> RouteDecision:
"""分析用户意图"""
# 构建消息列表
msg_list = [SystemMessage(content=self.system_prompt)]
# 添加历史消息
if messages:
recent_messages = messages[-5:]
msg_list.extend(recent_messages)
# 添加当前用户输入
msg_list.append(HumanMessage(content=user_input))
# 调用 LLM
output: IntentAnalysisOutput = await self.structured_llm.ainvoke(msg_list)
# 转换为 RouteDecision
return RouteDecision(
intent=output.intent,
confidence=output.confidence,
reasoning=output.reasoning,
response=output.response,
extracted_params=output.extracted_params,
target_nodes=output.suggested_nodes,
should_wait=output.should_wait,
)
LLM 输出模型定义：
class IntentAnalysisOutput(BaseModel):
"""LLM 意图分析输出"""
intent: str = Field(
description="识别的意图类型"
)
confidence: float = Field(
default=0.0,
ge=0.0,
le=1.0,
description="置信度"
)
reasoning: str = Field(
default="",
description="分析理由"
)
response: str = Field(
default="",
description="给用户的回复"
)
extracted_params: dict[str, Any] = Field(
default_factory=dict,
description="提取的参数"
)
suggested_nodes: list[str] = Field(
default_factory=list,
description="建议的目标节点"
)
should_wait: bool = Field(
default=False,
description="是否需要等待用户输入"
)
3. 流式 API 支持
SSE 协议实现
使用 Server-Sent Events 实现实时数据流：
from fastapi import APIRouter, Query
from fastapi.responses import StreamingResponse
def create_streaming_router(executor: StreamingGraphExecutor) -> APIRouter:
"""创建流式 API 路由器"""
router = APIRouter(prefix="/api/v1", tags=["streaming"])
@router.get("/chat/stream")
async def chat_stream(
message: str = Query(..., description="用户消息"),
thread_id: Optional[str] = Query(None, description="会话ID"),
user_id: Optional[str] = Query(None, description="用户ID"),
):
"""流式聊天接口（SSE）
事件类型：
- started: 开始执行
- node_start: 节点开始执行
- node_output: 节点输出数据
- node_end: 节点执行结束
- subgraph_start: 子图开始执行
- subgraph_node_output: 子图节点输出
- completed: 执行完成
- error: 错误
"""
# 生成 thread_id（如果未提供）
if not thread_id:
thread_id = str(uuid.uuid4())
return StreamingResponse(
stream_graph_sse(
executor=executor,
user_input=message,
thread_id=thread_id,
user_id=user_id,
),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
}
)
return router
SSE 响应格式示例：
event: node_start
data: {"node": "router", "timestamp": "2025-12-16T10:30:00"}
event: node_output
data: {"node": "router", "task": {"task_id": "abc123"}}
event: subgraph_start
data: {"parent_node": "xhs_agent", "subgraph_name": "xhs_workflow"}
event: subgraph_node_start
data: {"parent_node": "xhs_agent", "subgraph_node": "generate_content"}
event: subgraph_node_output
data: {"parent_node": "xhs_agent", "subgraph_node": "generate_content", "message": {...}}
event: completed
data: {"thread_id": "user_123", "timestamp": "2025-12-16T10:30:05"}
流式执行器
流式执行器封装了图执行并生成 SSE 事件：
async def stream_graph_sse(
executor: StreamingGraphExecutor,
user_input: str,
thread_id: str,
user_id: Optional[str] = None,
):
"""生成 SSE 事件流"""
async def event_generator():
# 1. 开始事件
yield f"event: started\ndata: {json.dumps({'thread_id': thread_id})}\n\n"
try:
# 2. 流式执行图
async for event in executor.stream_async(
user_input=user_input,
thread_id=thread_id,
user_id=user_id,
):
event_type = event.get("type")
event_data = event.get("data", {})
# 根据事件类型发送不同的 SSE 事件
if event_type == "node_start":
yield f"event: node_start\ndata: {json.dumps(event_data)}\n\n"
elif event_type == "node_output":
yield f"event: node_output\ndata: {json.dumps(event_data)}\n\n"
elif event_type == "subgraph_event":
# 子图事件需要特殊处理
yield f"event: subgraph_{event_data.get('subgraph_event_type')}\n"
yield f"data: {json.dumps(event_data)}\n\n"
except Exception as e:
yield f"event: error\ndata: {json.dumps({'error': str(e)})}\n\n"
# 3. 完成事件
yield f"event: completed\ndata: {json.dumps({'thread_id': thread_id})}\n\n"
return event_generator()
4. 小红书工作流子图
工作流状态定义
小红书工作流子图定义了完整的内容生成到发布流程的状态：
class XHSWorkflowState(TypedDict, total=False):
"""小红书工作流状态（扩展 GraphState 以支持直接集成）
从 GraphState 继承的字段：
messages: 消息历史（LangChain格式）
task: 当前任务
thread_id: 会话 ID
user_id: 用户 ID
iteration_count: 迭代计数
metadata: 元数据
工作流专有字段：
description: 用户描述的主题或需求
image_count: 需要生成的图片数量
should_publish: 是否发布到小红书
中间状态：
content_result: 内容生成Agent的结果
image_result: 图片生成Agent的结果
publish_result: 发布Agent的结果
输出字段：
success: 整个流程是否成功
error: 错误信息（如果失败）
final_output: 最终输出（整合所有结果）
"""
# 从 GraphState 继承的字段
messages: List[BaseMessage]
task: Optional[Task]
thread_id: str
user_id: str
iteration_count: int
metadata: Dict[str, Any]
# 工作流输入
description: str
image_count: int
should_publish: bool
# 中间状态
content_result: dict
image_result: dict
publish_result: dict
# 输出
success: bool
error: str
final_output: dict
节点函数实现
内容生成节点：
async def _generate_content_node(self, state: XHSWorkflowState) -> XHSWorkflowState:
"""内容生成节点
调用 XHSContentAgent 生成小红书内容大纲
"""
# 如果 description 未设置，从 task 中提取
if not state.get("description"):
task = state.get("task")
if task:
extracted_params = task.input_data.get("extracted_params", {})
route_decision = task.metadata.get("route_decision", {})
route_params = route_decision.get("extracted_params", {})
state["description"] = (
extracted_params.get("description")
or route_params.get("description")
or task.input_data.get("user_input", "")
)
state["image_count"] = int(
extracted_params.get("image_count")
or route_params.get("image_count")
or 3
)
state["should_publish"] = bool(task.input_data.get("publish", True))
description = state.get("description", "")
if not description:
state["success"] = False
state["error"] = "未提供内容描述"
return state
try:
# 确保Agent已初始化
await self.content_agent.initialize()
# 构建提示
prompt = f"""你是小红书内容生成助手，现在有一个创作需求：
主题：{description}
请你根据用户意图，在可用工具中选择最合适的一个来完成创作：
- 如果是普通主题类、小红书图文笔记，优先考虑使用 generate_xhs_note(topic=...)
- 如果更偏向人物设定、生活化吐槽、带强烈情绪的内容，优先考虑使用 generate_lifestyle_content(...)
要求：
1. 必须通过工具完成内容生成，不要自己写正文
2. 工具返回什么，就原样作为最终结果返回给用户，不要修改"""
# 调用内容生成Agent
result = await self.content_agent.invoke(prompt)
# 更新状态
state["content_result"] = result
state["messages"] = result.get("messages", [])
except Exception as e:
state["success"] = False
state["error"] = f"内容生成失败: {str(e)}"
return state
图片生成节点：
async def _generate_images_node(self, state: XHSWorkflowState) -> XHSWorkflowState:
"""图片生成节点
调用 XHSImageAgent 生成配图
"""
if state.get("error"):
return state
try:
await self.image_agent.initialize()
# 从内容结果中提取信息
content_result = state.get("content_result", {})
messages = content_result.get("messages", [])
# 尝试从消息中提取结果
outline_result = self._extract_outline_result(messages)
if outline_result and isinstance(outline_result, dict):
title = outline_result.get("title", "")
content_text = outline_result.get("content", "")
tags = outline_result.get("tags", [])
tags_str = " ".join([f"#{tag}" for tag in tags]) if tags else ""
prompt = f"""请调用 generate_images_batch 工具生成图片。
从上游内容生成Agent接收到的数据：
- 标题：{title}
- 正文：{content_text[:100]}...
- 标签：{tags_str}
操作：
调用 generate_images_batch(
full_content=\"\"\"标题：{title}\n\n正文：{content_text}\n\n标签：{tags_str}\"\"\",
style="",
)"""
else:
prompt = f"""请从以下内容结果中提取title、content、tags，然后调用 generate_images_batch 工具生成图片。
内容结果：
{content_result}"""
result = await self.image_agent.invoke(prompt)
state["image_result"] = result
# 如果不需要发布，标记为成功
if not state.get("should_publish"):
state["success"] = True
state["final_output"] = {
"content": state.get("content_result"),
"images": result,
}
task = state.get("task")
if task:
task.mark_completed(state["final_output"])
except Exception as e:
state["success"] = False
state["error"] = f"图片生成失败: {str(e)}"
task = state.get("task")
if task:
task.mark_failed(str(e))
return state
发布节点：
async def _publish_node(self, state: XHSWorkflowState) -> XHSWorkflowState:
"""发布节点
调用 XHSPublishAgent 发布到小红书
"""
if state.get("error"):
return state
try:
await self.publish_agent.initialize()
# 提取内容信息
content_result = state.get("content_result", {})
image_result = state.get("image_result", {})
outline_result = self._extract_outline_result(content_result.get("messages", []))
image_urls = self._extract_image_urls(image_result.get("messages", []))
if outline_result:
title = outline_result.get("title", "")
content_text = outline_result.get("content", "")
tags = outline_result.get("tags", [])
prompt = f"""请按以下步骤发布内容到小红书：
【内容信息】
- 标题：{title}
- 正文：{content_text[:200]}...
- 标签：{tags}
【图片信息】
- 图片URLs：{image_urls}
操作步骤：
1. 调用 xiaohongshu_check_login_session() 检查登录状态
2. 如果未登录，提示用户需要登录
3. 如果已登录，调用 xiaohongshu_publish_content(
title="{title}",
content="{content_text}",
images={image_urls},
tags={tags}
)"""
result = await self.publish_agent.invoke(prompt)
state["publish_result"] = result
state["success"] = True
state["final_output"] = {
"content": state.get("content_result"),
"images": state.get("image_result"),
"publish": result,
}
task = state.get("task")
if task:
task.mark_completed(state["final_output"])
except Exception as e:
state["success"] = False
state["error"] = f"发布失败: {str(e)}"
task = state.get("task")
if task:
task.mark_failed(str(e))
return state
条件路由
def _should_publish_router(self, state: XHSWorkflowState) -> Literal["publish", "end"]:
"""判断是否需要发布
条件：
1. should_publish为True
2. 前面步骤没有错误
Returns:
"publish": 进入发布节点
"end": 结束流程
"""
should_publish = state.get("should_publish", False)
has_error = bool(state.get("error"))
if should_publish and not has_error:
self.logger.info("Routing to publish node")
return "publish"
else:
if has_error:
self.logger.info("Skipping publish due to error")
else:
self.logger.info("Skipping publish (should_publish=False)")
return "end"
子图构建：
async def _build_graph(self) -> StateGraph:
"""构建工作流图
图结构：
START
↓
[内容生成节点]
↓
[图片生成节点]
↓
{条件路由}
↙ ↘
[发布节点] END
↓
END
"""
workflow = StateGraph(XHSWorkflowState)
# 添加节点
workflow.add_node("generate_content", self._generate_content_node)
workflow.add_node("generate_images", self._generate_images_node)
workflow.add_node("publish", self._publish_node)
# 定义边
workflow.set_entry_point("generate_content")
workflow.add_edge("generate_content", "generate_images")
# 条件路由
workflow.add_conditional_edges(
"generate_images",
self._should_publish_router,
{
"publish": "publish",
"end": END,
}
)
workflow.add_edge("publish", END)
return workflow
五、遇到的问题与解决方案
问题1：LangGraph 版本兼容性
现象：安装依赖时出现版本冲突，LangGraph 与其他依赖包版本不兼容
原因分析：
- LangGraph 作为新兴框架，版本更新频繁，API 变化较大
- 与 LangChain 等依赖包存在严格的版本依赖关系
- 不同版本的 API 接口存在显著差异，例如异步接口的变化
排查过程：
- 检查
pyproject.toml中的版本约束 - 查看 LangGraph 官方文档的版本要求
- 尝试不同版本组合
解决方案：
- 固定 LangGraph 版本为 0.3.0
- 确保 LangChain 版本与 LangGraph 兼容（0.2.x 系列）
- 使用 uv 包管理器管理依赖版本，解决冲突
- 在
pyproject.toml中明确指定所有依赖版本
最终配置：
[tool.poetry.dependencies]
python = "^3.11"
langgraph = "0.3.0"
langchain = "0.2.16"
langchain-core = "0.2.38"
langchain-community = "0.2.16"
langchain-qwen = "0.2.1"
问题2：流式输出实现
现象：实现流式输出时，前端无法正确接收事件数据，连接经常中断
原因分析：
- SSE 协议要求特定的响应格式和编码
- FastAPI 的流式响应配置不正确
- Nginx 代理会缓冲响应，导致实时性降低
- 缺少心跳机制，连接可能超时
排查过程：
- 使用 Chrome 开发者工具检查 Network 面板
- 查看服务器日志中的 SSE 事件发送记录
- 测试不同浏览器对 SSE 的支持情况
解决方案：
- 正确配置 FastAPI 的 StreamingResponse
- 添加 Nginx 配置禁用缓冲
- 实现心跳机制保持连接活跃
- 添加错误处理和前端重连机制
关键代码实现：
# API 层配置
return StreamingResponse(
stream_graph_sse(...),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no", # 关键：禁用 Nginx 缓冲
}
)
# 心跳机制
async def keep_alive():
while True:
yield ": keepalive\n\n"
await asyncio.sleep(15)
# 前端重连机制
const eventSource = new EventSource(url);
eventSource.onerror = () => {
console.log('Connection lost, reconnecting...');
setTimeout(() => {
eventSource = new EventSource(url);
}, 1000);
};
问题3：状态管理与持久化
现象：工作流执行过程中状态丢失，无法恢复执行，多轮对话时上下文丢失
原因分析：
- LangGraph 使用内存存储状态，服务重启后丢失
- 状态检查点机制配置不正确
- 多用户并发时状态混淆
排查过程：
- 添加日志追踪状态变化
- 检查 MemorySaver 配置
- 测试服务重启后的状态恢复
解决方案：
- 实现基于文件的检查点存储
- 使用 MemorySaver 作为内存缓存
- 添加会话隔离机制
代码实现：
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
class StateManager:
"""状态管理器 - 支持多种存储后端"""
def __init__(
self,
storage_type: str = "memory",
storage_path: str = "./state",
):
if storage_type == "memory":
self.checkpointer = MemorySaver()
elif storage_type == "sqlite":
self.checkpointer = SqliteSaver.from_conn_string(
f"{storage_path}/state.db"
)
else:
raise ValueError(f"Unknown storage type: {storage_type}")
self._task_store: dict[str, Task] = {}
self._session_store: dict[str, dict] = {}
async def save_state(self, thread_id: str, state: dict):
"""保存状态到检查点"""
config = {"configurable": {"thread_id": thread_id}}
# LangGraph 自动处理状态持久化
async def load_state(self, thread_id: str) -> Optional[dict]:
"""从检查点加载状态"""
config = {"configurable": {"thread_id": thread_id}}
# 通过图的 get_state 方法获取
问题4：MCP 服务集成
现象：无法正确连接和调用 MCP 服务，超时或返回错误结果
原因分析：
- MCP 服务配置格式不正确
- 网络连接问题或认证失败
- 服务调用参数格式不匹配
- 服务响应超时
排查过程：
- 检查 MCP 服务配置文件
- 测试服务连接性
- 查看服务日志定位问题
解决方案：
- 实现服务健康检查机制
- 添加服务调用重试逻辑
- 实现超时控制
- 统一错误处理
代码实现：
class MCPServiceClient:
"""MCP 服务客户端"""
def __init__(self, service_config: dict):
self.config = service_config
self.timeout = service_config.get("timeout", 30)
self.max_retries = service_config.get("max_retries", 3)
async def call_tool(
self,
tool_name: str,
parameters: dict,
) -> dict:
"""调用 MCP 工具，支持重试"""
for attempt in range(self.max_retries):
try:
async with asyncio.timeout(self.timeout):
result = await self._execute_tool(tool_name, parameters)
return result
except asyncio.TimeoutError:
logger.warning(
f"MCP call timeout",
tool=tool_name,
attempt=attempt + 1,
max_retries=self.max_retries
)
except Exception as e:
logger.error(
f"MCP call failed",
tool=tool_name,
error=str(e),
attempt=attempt + 1
)
if attempt < self.max_retries - 1:
await asyncio.sleep(2 ** attempt) # 指数退避
raise MCPServiceError(f"Failed to call {tool_name} after {self.max_retries} attempts")
问题5：子图与主图集成
现象：子图无法正确嵌入主图，流式输出事件无法传递到前端
原因分析：
- 子图的编译方式与主图不兼容
- 事件传递机制未正确实现
- 状态映射不正确
解决方案：
- 直接添加编译后的子图到主图
- 使用特殊的边类型连接子图节点
- 在流式执行器中正确处理子图事件
关键代码：
async def build(self) -> StateGraph:
"""构建包含子图的主图"""
workflow = StateGraph(GraphState)
# 添加小红书工作流子图
try:
from ..subgraphs import XHSWorkflowSubgraph
xhs_subgraph = XHSWorkflowSubgraph()
# 关键：初始化子图
await xhs_subgraph.initialize()
# 关键：直接添加编译后的子图
# LangGraph 会自动处理子图的流式输出
workflow.add_node("xhs_workflow", xhs_subgraph.graph)
except Exception as e:
logger.error(f"Failed to add xhs_workflow subgraph: {e}")
# 条件路由到子图
workflow.add_conditional_edges(
"router",
self._route_from_router,
{
"xhs_workflow": "xhs_workflow",
"wait": "wait",
"end": END,
}
)
六、测试与验证
单元测试
测试框架配置：
# tests/conftest.py
import pytest
import asyncio
@pytest.fixture(scope="session")
def event_loop():
"""创建事件循环"""
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest.fixture
async def router_system():
"""路由系统测试夹具"""
from ai_social_scheduler.router import RouterSystem
routes = [
RouteConfig(
route_id="xhs_content",
keywords=["小红书", "笔记", "内容"],
target_nodes=["xhs_agent"],
),
]
router = RouterSystem(
routes=routes,
strategy=RouterStrategy.RULE_FIRST,
enable_llm=False, # 测试时不启用 LLM
)
yield router
测试用例示例：
# tests/test_router.py
@pytest.mark.asyncio
async def test_rule_matching(router_system):
"""测试规则匹配"""
decision = await router_system.route(
user_input="帮我写一篇小红书笔记",
context={},
messages=[],
)
assert decision is not None
assert "xhs_agent" in decision.target_nodes
assert decision.confidence > 0
@pytest.mark.asyncio
async def test_llm_routing():
"""测试 LLM 路由"""
router = RouterSystem(
routes=[],
strategy=RouterStrategy.LLM_FIRST,
enable_llm=True,
available_nodes=["xhs_agent", "image_agent"],
)
decision = await router.route(
user_input="生成一些关于旅行的内容",
context={},
messages=[],
)
assert decision is not None
assert decision.intent is not None
集成测试
# tests/test_workflow.py
@pytest.mark.asyncio
async def test_xhs_workflow_subgraph():
"""测试小红书工作流子图"""
from ai_social_scheduler.subgraphs import XHSWorkflowSubgraph
workflow = XHSWorkflowSubgraph()
await workflow.initialize()
result = await workflow.invoke({
"description": "咖啡制作教程",
"image_count": 3,
"should_publish": False,
})
assert result.get("success") == True
assert "content_result" in result
assert "image_result" in result
性能测试
# tests/performance/test_streaming.py
import time
import asyncio
@pytest.mark.asyncio
async def test_streaming_latency():
"""测试流式输出延迟"""
start_time = time.time()
async for event in executor.stream_async(...):
latency = time.time() - start_time
assert latency < 0.5 # 延迟应小于 500ms
start_time = time.time() # 重置计时器
@pytest.mark.asyncio
async def test_concurrent_requests():
"""测试并发请求"""
tasks = [
executor.stream_async(
user_input=f"测试请求 {i}",
thread_id=f"thread_{i}",
)
for i in range(10)
]
results = await asyncio.gather(*tasks)
assert len(results) == 10
七、开发进度与下一步计划
下一步计划
-
完善工作流定义
- 扩展更多平台的工作流
- 优化工作流执行效率
- 支持自定义工作流配置
-
增强错误处理和重试机制
- 实现更完善的错误处理
- 添加智能重试策略
- 提供详细的错误反馈
-
性能优化和监控
- 优化工作流执行速度
- 实现系统监控
- 提供性能指标分析
-
扩展功能
- 定时任务支持
- 数据分析功能
- 自动回复功能
- 接入更多平台（抖音、快手）
-
完善文档和示例
- 编写详细的开发文档
- 提供更多使用示例
- 完善 API 文档
总结
本周完成了 AI Social Scheduler 项目的基础架构搭建和核心功能实现。通过深入学习和实践 LangGraph 框架，我们成功构建了基于状态图的工作流引擎，实现了智能路由系统和流式 API 支持。
在开发过程中，我们遇到了多个技术挑战，包括版本兼容性、流式输出实现、状态管理与持久化、MCP 服务集成以及子图与主图集成等问题。通过系统性的分析和解决，我们不仅完成了功能开发，还积累了宝贵的经验。
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
10
0- 0
已为社区贡献4条内容
所有评论(0)