#实战案例
三个由浅入深的例子：顺序双角色 Crew、Flow 包住 Crew、自定义工具（检索思路对接 RAG）。运行前设置 OPENAI_API_KEY；搜索工具另需 SERPER_API_KEY。
#案例 1：研究员 + 写作者（顺序 Crew）
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool
search = SerperDevTool()
researcher = Agent(
role="Senior Researcher",
goal="Gather accurate, dated facts about {topic}",
backstory="You verify claims and prefer primary sources.",
tools=[search],
verbose=True,
)
writer = Agent(
role="Technical Writer",
goal="Turn research notes into a briefing engineers will finish",
backstory="You write short paragraphs and never invent citations.",
verbose=True,
)
research_task = Task(
description="Research {topic}. Use web search. Note dates and sources.",
expected_output="Markdown bullets: fact, source, why it matters.",
agent=researcher,
)
write_task = Task(
description="Write a 400-word briefing from the research notes.",
expected_output="Markdown with Summary, Findings, Risks. No wrapping fence.",
agent=writer,
context=[research_task],
markdown=True,
output_file="output/briefing.md",
)
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, write_task],
process=Process.sequential,
verbose=True,
memory=True,
)
result = crew.kickoff(inputs={"topic": "open-weight LLM serving"})
print(result.raw)无 Serper 时去掉 tools=[search]，改为纯模型知识（时效差，仅作 API 练习）。
#案例 2：Flow 设主题，步骤内 kickoff Crew
from pydantic import BaseModel
from crewai import Agent, Task, Crew, Process
from crewai.flow.flow import Flow, listen, start
class PipelineState(BaseModel):
topic: str = ""
briefing: str = ""
def make_crew() -> Crew:
agent = Agent(
role="Briefing researcher",
goal="Produce a sourced outline of {topic}",
backstory="You are skeptical of marketing copy.",
)
task = Task(
description="Outline {topic} in 8 bullets with one risk each.",
expected_output="Eight markdown bullets.",
agent=agent,
)
return Crew(agents=[agent], tasks=[task], process=Process.sequential)
class BriefingPipeline(Flow[PipelineState]):
@start()
def set_topic(self):
self.state.topic = "CrewAI vs LangGraph"
@listen(set_topic)
def research(self):
out = make_crew().kickoff(inputs={"topic": self.state.topic})
self.state.briefing = out.raw
return out.raw
@listen(research)
def done(self):
print("chars:", len(self.state.briefing))
return self.state.briefing
if __name__ == "__main__":
print(BriefingPipeline().kickoff())扩展：在 research 后加 @router，按长度或关键词走「通过 / 重写」；或 @persist 以便失败后续跑。这就是官方说的 Flow 骨架 + Crew 智能。
#案例 3：自定义工具做「假检索」
真实项目应接向量库（RAG、LlamaIndex 的 LlamaIndexTool）。下面用内存字典演示工具契约：
from typing import Type
from pydantic import BaseModel, Field
from crewai import Agent, Task, Crew, Process
from crewai.tools import BaseTool
FAQ = {
"flows": "Flows own state and control; use @start and @listen.",
"crews": "Crews are role-playing teams; Process is sequential or hierarchical.",
}
class SearchInput(BaseModel):
query: str = Field(..., description="Keyword: flows or crews")
class FaqSearchTool(BaseTool):
name: str = "faq_search"
description: str = "Search the internal FAQ. Use for CrewAI concept questions."
args_schema: Type[BaseModel] = SearchInput
def _run(self, query: str) -> str:
q = query.lower()
hits = [v for k, v in FAQ.items() if k in q or q in k]
return "\n".join(hits) if hits else "No FAQ hit. Say you do not know."
agent = Agent(
role="Academy tutor",
goal="Answer only from faq_search",
backstory="You refuse to guess when the tool is empty.",
tools=[FaqSearchTool()],
)
task = Task(
description="Explain how Flows relate to Crews for a new student.",
expected_output="Four sentences max, grounded in tool output.",
agent=agent,
)
print(Crew(agents=[agent], tasks=[task], process=Process.sequential).kickoff().raw)上线时把 _run 换成 Chroma / pgvector / Qdrant 查询，评估方法见 RAG 课，不要在 Crew 里「再加一个只会复述的 Agent」冒充检索质量。
#反模式
| 反模式 | 改进 |
|---|---|
| 整个产品只有 Crew、没有 Flow | 可预测步骤放到 Flow，协作步骤再嵌 Crew |
| 每个 Agent 挂全部工具 | 按角色裁剪；任务级再收紧 |
| 层级 Process 却不设 manager | 必须 manager_llm 或 manager_agent |
用 Memory 代替 state.topic | 编排变量用 Flow state / Task inputs |