模块10: LangChain与CrewAI

从零构建智能体（如我们在模块4-6中所做的）可以给您最大的控制和理解。但随着您构建更多智能体，您会发现自己在重复构建相同的模式：工具注册、记忆管理、提示模板、错误处理和对话循环。框架提供了这些模式的经过实战检验的实现，这样您可以专注于应用逻辑而非基础设施。

Python生态系统中最流行的两个框架是LangChain（通用智能体框架）和CrewAI（多智能体编排框架）。它们服务于不同的需求，甚至可以一起使用。

话虽如此，框架增加了复杂性、抽象层和依赖。并非每个项目都需要框架。关键是理解它们提供什么，以便您可以做出明智的决定。

LangChain是最广泛采用的LLM应用框架。它提供模块化架构，每个组件（模型、提示、链、智能体、记忆）可以独立使用或组合在一起。该框架已经显著演进，现在以LangChain表达式语言（LCEL）为中心，用于声明式地组合链。

pip install langchain langchain-anthropic langchain-openai

核心概念

让我们使用LangChain逐步构建越来越复杂的应用，从简单的链开始，逐步到完整的工具调用智能体。

使用LCEL的简单链

from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Create a simple chain: prompt -> model -> parse output
llm = ChatAnthropic(model="claude-sonnet-4-20250514")
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are an expert {domain} consultant."),
    ("human", "{question}")
])
parser = StrOutputParser()

# Compose with the pipe operator (LCEL)
chain = prompt | llm | parser

# Invoke
answer = chain.invoke({
    "domain": "cybersecurity",
    "question": "What are the top 3 threats for small businesses?"
})
print(answer)

使用Pydantic的结构化输出

from pydantic import BaseModel, Field
from langchain_core.output_parsers import JsonOutputParser

class MovieReview(BaseModel):
    title: str = Field(description="Movie title")
    rating: float = Field(description="Rating out of 10")
    summary: str = Field(description="One-sentence summary")
    recommend: bool = Field(description="Whether to recommend")

parser = JsonOutputParser(pydantic_object=MovieReview)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Analyse the movie and respond in JSON.\n{format_instructions}"),
    ("human", "Review the movie: {movie}")
])

chain = prompt | llm | parser
review = chain.invoke({
    "movie": "Inception",
    "format_instructions": parser.get_format_instructions()
})
print(f"{review['title']}: {review['rating']}/10 - {review['summary']}")

工具调用智能体

from langchain_anthropic import ChatAnthropic
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
import requests

@tool
def search_web(query: str) -> str:
    """Search the web for current information about a topic.
    Use this when you need up-to-date information."""
    # In production, use a real search API
    return f"Top results for '{query}': [simulated search results]"

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression. Use Python syntax.
    Example: '2 ** 10' returns '1024'."""
    try:
        result = eval(expression, {"__builtins__": {}})
        return str(result)
    except Exception as e:
        return f"Error: {e}"

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    # Simulated
    return f"Weather in {city}: 18C, partly cloudy, humidity 65%"

# Set up the agent
llm = ChatAnthropic(model="claude-sonnet-4-20250514")
tools = [search_web, calculate, get_weather]

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant with access to tools. "
               "Use tools when needed to provide accurate answers."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Run the agent
result = executor.invoke({
    "input": "What's 2^16 and what's the weather in Tokyo?"
})
print(result["output"])

使用LangChain的RAG链

from langchain_anthropic import ChatAnthropic
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# Set up vector store
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(persist_directory="./chroma_db",
                     embedding_function=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# RAG prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", """Answer based only on the following context.
If the context doesn't contain the answer, say so.

Context: {context}"""),
    ("human", "{question}")
])

# Format retrieved docs
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# Compose the RAG chain
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | ChatAnthropic(model="claude-sonnet-4-20250514")
    | StrOutputParser()
)

answer = rag_chain.invoke("What is the vacation policy?")

CrewAI采用与LangChain根本不同的方法。不是构建单独的链和智能体，而是定义一个由具有不同角色、目标和背景故事的智能体团队，然后为它们分配任务。CrewAI处理编排 — 决定哪个智能体处理哪个任务，在智能体之间传递结果，并管理整体工作流。

这自然地映射到现实世界的团队动态。就像一个产品团队有研究员、设计师和工程师一样，一个CrewAI团队可以有研究智能体、分析智能体和写作智能体，各自专注于自己的角色。

pip install crewai crewai-tools

CrewAI核心概念

让我们构建一个实用的CrewAI应用：一个分析主题并生成报告的内容研究团队。

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool  # web search tool

# --- Define Agents ---

researcher = Agent(
    role="Senior Research Analyst",
    goal="Uncover cutting-edge developments and key data points about {topic}",
    backstory="""You are an expert research analyst with 15 years of experience.
    You excel at finding reliable sources, identifying trends, and
    synthesising complex information into clear insights. You always
    verify facts from multiple sources.""",
    tools=[SerperDevTool()],
    verbose=True,
    allow_delegation=False
)

analyst = Agent(
    role="Data Analyst",
    goal="Analyse research findings and identify key patterns and implications",
    backstory="""You are a skilled data analyst who specialises in
    interpreting qualitative and quantitative information. You create
    clear comparisons and identify trends that others miss. You are
    known for your balanced, objective analysis.""",
    verbose=True,
    allow_delegation=False
)

writer = Agent(
    role="Technical Writer",
    goal="Create a compelling, well-structured report about {topic}",
    backstory="""You are an award-winning technical writer who transforms
    complex research into accessible, engaging content. You write in a
    clear, professional style with strong structure and flow.""",
    verbose=True,
    allow_delegation=False
)

# --- Define Tasks ---

research_task = Task(
    description="""Research the topic: {topic}

    Find:
    1. Current state and recent developments (last 6 months)
    2. Key players and their market positions
    3. Technical innovations and breakthroughs
    4. Challenges and limitations
    5. Market size and growth projections

    Compile your findings with sources.""",
    agent=researcher,
    expected_output="Detailed research notes with sources and key data points"
)

analysis_task = Task(
    description="""Analyse the research findings about {topic}.

    Produce:
    1. SWOT analysis (Strengths, Weaknesses, Opportunities, Threats)
    2. Trend analysis with timeline
    3. Competitive landscape comparison
    4. Risk assessment
    5. Three possible future scenarios (optimistic, realistic, pessimistic)""",
    agent=analyst,
    expected_output="Structured analysis with SWOT, trends, and scenarios",
    context=[research_task]  # depends on research results
)

report_task = Task(
    description="""Write a professional report about {topic}.

    The report should include:
    - Executive summary (200 words)
    - Introduction and background
    - Key findings (from research)
    - Analysis and implications
    - Recommendations
    - Conclusion

    Target audience: C-level executives.
    Tone: Professional but accessible. No jargon without explanation.""",
    agent=writer,
    expected_output="A polished report in markdown format, 1500-2000 words",
    context=[research_task, analysis_task]  # depends on both
)

# --- Assemble and Run the Crew ---

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, report_task],
    process=Process.sequential,  # tasks run in order
    verbose=True
)

# Kick off the crew
result = crew.kickoff(inputs={"topic": "AI agents in enterprise software"})
print(result)

CrewAI中的自定义工具

from crewai.tools import BaseTool
from pydantic import BaseModel, Field

class DatabaseQueryInput(BaseModel):
    query: str = Field(description="SQL SELECT query to execute")

class DatabaseQueryTool(BaseTool):
    name: str = "query_database"
    description: str = "Execute a read-only SQL query against the analytics database."
    args_schema: type[BaseModel] = DatabaseQueryInput

    def _run(self, query: str) -> str:
        import sqlite3
        if not query.strip().upper().startswith("SELECT"):
            return "Error: Only SELECT queries allowed"
        conn = sqlite3.connect("analytics.db")
        try:
            rows = conn.execute(query).fetchall()
            return str(rows[:50])  # limit output
        finally:
            conn.close()

# Give the tool to an agent
data_agent = Agent(
    role="Data Engineer",
    goal="Query databases to find requested information",
    backstory="Expert in SQL and data analysis.",
    tools=[DatabaseQueryTool()]
)

没有单一最好的框架。正确的选择取决于您的用例、团队经验和需求。以下是详细的比较，帮助您做出决定。

决策流程图

• 简单聊天机器人或问答？ → 直接API调用或LangChain
• 文档上的RAG？ → LangChain（优秀的检索器生态系统）
• 带工具的单智能体？ → LangChain或直接API
• 多智能体协作？ → CrewAI或LangGraph
• 复杂的有状态工作流？ → LangGraph（基于图的智能体编排）
• 需要最大性能和控制？ → 直接API调用