基于 CrewAI 和 LangGraph 构建自动化研报分析流水线的实战踩坑指南

直接把 CrewAI 当成简单的“多 Agent 聊天室”是很多人的误区，在构建研报分析流水线时，如果你依赖它的默认顺序执行，很容易在处理长文本分析时陷入死循环或者逻辑漂移。我尝试将 LangGraph 作为底层状态机来调度 CrewAI 的 Agent，解决了状态丢失和无法精准回溯的问题。

最核心的坑在于：CrewAI 的 Process.sequential 无法处理复杂的条件跳转。比如，当分析 Agent 发现研报数据缺失时，应该跳回搜索 Agent 补齐，而不是强行进入总结阶段。

我的解决方案是利用 LangGraph 构建一个有向图，将 CrewAI 的 Agent 封装成 Node。

配置核心逻辑：

from langgraph.graph import StateGraph, END
from crewai import Agent, Task, Crew

# 定义状态，用于在 Node 之间传递分析进度和中间结论
class AnalysisState(TypedDict):
    report_content: str
    extracted_data: list
    validation_status: bool

# 封装 CrewAI Agent 为 LangGraph 节点
def research_node(state: AnalysisState):
    researcher = Agent(role='资深分析师', goal='提取核心财务指标', backlight='...')
    task = Task(description=f'分析以下内容: {state["report_content"]}', agent=researcher)
    crew = Crew(agents=[researcher], tasks=[task])
    result = crew.kickoff()
    return {"extracted_data": result}

def validate_node(state: AnalysisState):
    # 这里写校验逻辑，如果数据不全则返回 False
    is_valid = len(state["extracted_data"]) > 5 
    return {"validation_status": is_valid}

# 构建工作流
workflow = StateGraph(AnalysisState)
workflow.add_node("research", research_node)
workflow.add_node("validate", validate_node)

workflow.set_entry_point("research")
workflow.add_edge("research", "validate")
workflow.add_conditional_edges(
    "validate",
    lambda x: "research" if not x["validation_status"] else END
)
app = workflow.compile()

几个实战避坑点：

内存溢出与上下文截断：研报动辄几十页，直接塞给 Agent 会导致 Token 爆炸且丢失细节。建议在 research_node 前加一层 RAG 预处理，用 RecursiveCharacterTextSplitter 切分，并只将相关片段传给 CrewAI。

Prompt 污染：CrewAI 的内置角色描述有时会干扰具体任务。我发现把 goal 写得极简，而把具体要求写在 Task 的 description 里，Agent 的执行稳定性会提升 30% 以上。

依赖冲突：安装 crewai 和 langgraph 时，经常会出现 pydantic 版本冲突。建议强制统一使用 pydantic>=2.0，并在虚拟环境中先装 langgraph 再装 crewai。

效率提升技巧：为了加速调试，不要每次都跑整个图。利用 LangGraph 的 checkpoint 功能把中间状态存入 SQLite，出错时直接从 validate 节点重启，不用重新跑耗时的搜索过程。