传统大语言模型(LLM)是被动的——你问,它答,然后等你的下一个问题。而AI Agent(智能体)是主动的——它接收目标,自主规划步骤,调用工具,迭代执行,直到任务完成。
| 特性 | 普通LLM对话 | AI Agent |
|---|---|---|
| 工作模式 | 单次请求→单次回复 | 目标→自主多步执行 |
| 工具使用 | ❌ 无 | ✅ 搜索/代码/API/文件 |
| 自我反思 | ❌ 无 | ✅ 评估结果、修正错误 |
| 记忆 | 仅当前对话 | 短期+长期持久记忆 |
| 并行执行 | ❌ | ✅ 多Agent并行 |
| 适合任务 | 简单问答、生成 | 复杂多步骤自动化 |
ReAct(Reasoning + Acting)是目前最主流的单Agent架构:模型交替进行推理(Thought)和行动(Action),每次行动后观察结果,再继续推理,直到任务完成。
先让一个规划LLM将复杂任务拆解为子任务列表,再由执行Agent逐一完成。适合长期、复杂的多步骤任务。
多个专业Agent协作,每个Agent负责特定领域(研究员、写作者、审查员、执行者),通过编排层(Orchestrator)协调工作,实现复杂任务并行执行。
LangGraph将Agent工作流建模为有向图(DAG + 循环),每个节点是一个处理步骤,边代表条件转移。这种方式让复杂Agent逻辑变得清晰可调试,是目前企业级AI Agent开发的主流选择。
pip install langgraph langchain-anthropic langchain-community
from typing import TypedDict, Annotated, Sequence
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import BaseMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
import operator
# ── 1. 定义工具 ──────────────────────────────────
@tool
def search_web(query: str) -> str:
"""搜索网络获取最新信息"""
# 实际使用时替换为真实搜索API
return f"搜索结果:关于'{query}'的相关信息..."
@tool
def run_python_code(code: str) -> str:
"""执行Python代码并返回结果"""
import io, contextlib
output = io.StringIO()
with contextlib.redirect_stdout(output):
exec(code, {})
return output.getvalue()
@tool
def read_file(filepath: str) -> str:
"""读取本地文件内容"""
with open(filepath, 'r', encoding='utf-8') as f:
return f.read()
tools = [search_web, run_python_code, read_file]
# ── 2. 定义Agent状态 ─────────────────────────────
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
# ── 3. 创建模型(绑定工具)────────────────────────
model = ChatAnthropic(
model="claude-3-7-sonnet-20250219",
temperature=0
).bind_tools(tools)
# ── 4. 定义节点函数 ──────────────────────────────
def call_model(state: AgentState):
"""LLM推理节点"""
messages = state["messages"]
response = model.invoke(messages)
return {"messages": [response]}
def should_continue(state: AgentState):
"""判断是否继续调用工具"""
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools" # 有工具调用,继续
return END # 无工具调用,结束
# ── 5. 构建图 ────────────────────────────────────
tool_node = ToolNode(tools) # 自动处理工具执行
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
"agent",
should_continue,
{"tools": "tools", END: END}
)
workflow.add_edge("tools", "agent") # 工具执行完回到agent
app = workflow.compile()
# ── 6. 运行Agent ─────────────────────────────────
result = app.invoke({
"messages": [HumanMessage(content="帮我搜索2025年最新的AI Agent框架排名,然后用Python统计下有多少个")]
})
print(result["messages"][-1].content)
CrewAI让你像管理团队一样管理AI Agent:定义角色(Role)、分配任务(Task)、设定目标(Goal),多个Agent自动协作完成复杂项目。非常适合内容创作、研究、数据分析等场景。
pip install crewai crewai-tools
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, FileReadTool
import os
os.environ["OPENAI_API_KEY"] = "your-key" # 或使用Anthropic
search_tool = SerperDevTool()
# ── 定义AI团队成员 ────────────────────────────────
researcher = Agent(
role='AI研究分析师',
goal='收集和分析最新的AI技术趋势,提供深度见解',
backstory="""你是一位经验丰富的AI研究员,擅长从海量信息中
提炼关键洞察,对技术发展有独到的判断力。""",
tools=[search_tool],
verbose=True,
max_iter=5 # 最多迭代5次
)
writer = Agent(
role='技术内容写作专家',
goal='将研究报告转化为清晰易懂、引人入胜的技术文章',
backstory="""你是一位资深技术写作者,能将复杂的技术概念
以通俗易懂的方式呈现给非专业读者,文章结构清晰,逻辑严谨。""",
verbose=True,
llm="claude-3-7-sonnet-20250219"
)
reviewer = Agent(
role='内容质量审核员',
goal='确保文章准确性、完整性和可读性,提出具体改进建议',
backstory="""你是严格的内容审核专家,专注于事实核查和
文章质量提升,不放过任何错误或不清晰之处。""",
verbose=True
)
# ── 定义任务 ──────────────────────────────────────
research_task = Task(
description="""搜索并分析2025年AI Agent的最新发展趋势:
1. 主流框架及其优缺点
2. 实际商业应用案例
3. 技术挑战和未来方向
输出:500字结构化研究报告""",
agent=researcher,
expected_output="包含框架对比、应用案例和技术趋势的研究报告"
)
writing_task = Task(
description="""基于研究报告,撰写一篇面向开发者的技术文章:
- 标题吸引眼球
- 包含具体代码示例
- 适合中级Python开发者阅读
- 长度:800-1000字""",
agent=writer,
expected_output="完整的技术博客文章,包含代码示例"
)
review_task = Task(
description="""审核文章并提供改进意见:
1. 技术准确性检查
2. 代码示例验证
3. 结构和可读性评估
4. 提出3-5条具体改进建议""",
agent=reviewer,
expected_output="审核报告和最终优化版文章"
)
# ── 创建并运行Crew ────────────────────────────────
crew = Crew(
agents=[researcher, writer, reviewer],
tasks=[research_task, writing_task, review_task],
process=Process.sequential, # 顺序执行(也可parallel)
verbose=2
)
result = crew.kickoff()
print(result)
from openai import OpenAI
client = OpenAI()
# ── 创建Assistant ─────────────────────────────────
assistant = client.beta.assistants.create(
name="数据分析助手",
instructions="""你是专业的数据分析师。
当用户上传数据文件时,自动分析数据并生成可视化图表。
代码必须有注释,结论要有数据支撑。""",
model="gpt-4o",
tools=[
{"type": "code_interpreter"}, # 执行Python代码
{"type": "file_search"} # 搜索上传的文件
]
)
# ── 创建Thread(对话线程)────────────────────────
thread = client.beta.threads.create()
# ── 发送消息 ──────────────────────────────────────
client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content="分析这份销售数据,找出增长最快的产品类别并生成折线图"
)
# ── 运行Assistant ─────────────────────────────────
run = client.beta.threads.runs.create_and_poll(
thread_id=thread.id,
assistant_id=assistant.id,
timeout=120
)
if run.status == "completed":
messages = client.beta.threads.messages.list(thread_id=thread.id)
for msg in messages.data:
if msg.role == "assistant":
for content in msg.content:
if content.type == "text":
print(content.text.value)
pip install pyautogen
import autogen
# 配置LLM
config_list = [{
"model": "gpt-4o",
"api_key": "your-openai-key"
}]
llm_config = {"config_list": config_list, "temperature": 0.1}
# ── 定义多个Agent ─────────────────────────────────
user_proxy = autogen.UserProxyAgent(
name="用户代理",
human_input_mode="NEVER", # 自动运行,不需要人工输入
max_consecutive_auto_reply=10,
is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
code_execution_config={
"work_dir": "workspace",
"use_docker": False
}
)
python_expert = autogen.AssistantAgent(
name="Python专家",
system_message="""你是Python专家。编写高质量、有注释的Python代码。
代码写完后说 TERMINATE""",
llm_config=llm_config
)
code_reviewer = autogen.AssistantAgent(
name="代码审核",
system_message="""你是代码审核专家。审查代码质量、安全性和效率。
给出具体改进建议后说 TERMINATE""",
llm_config=llm_config
)
# ── 启动多Agent对话 ────────────────────────────────
# GroupChat让多个Agent协作
groupchat = autogen.GroupChat(
agents=[user_proxy, python_expert, code_reviewer],
messages=[],
max_round=12
)
manager = autogen.GroupChatManager(groupchat=groupchat, llm_config=llm_config)
user_proxy.initiate_chat(
manager,
message="写一个爬取GitHub Trending的Python脚本,然后让代码审核专家审查"
)
| 框架 | 上手难度 | 适合场景 | 生产就绪 | 推荐指数 |
|---|---|---|---|---|
| LangGraph | 中等 | 复杂状态管理、企业级 | ⭐⭐⭐⭐⭐ | 9.5/10 |
| CrewAI | 简单 | 多Agent协作、内容创作 | ⭐⭐⭐⭐ | 9.0/10 |
| OpenAI Assistants | 最简单 | 快速原型、内置工具 | ⭐⭐⭐⭐ | 8.5/10 |
| AutoGen | 中等 | 多Agent对话、研究 | ⭐⭐⭐ | 8.0/10 |
多Agent协作:搜索Agent收集信息 → 分析Agent提炼洞察 → 写作Agent生成报告 → 审核Agent校对优化。一键生成专业级行业研究报告,耗时从2天降至30分钟。
RAG + Agent结合:将代码库向量化后,Agent读取代码结构,定位Bug位置,生成修复补丁,运行测试验证,全程无需人工干预。
# 代码库分析Agent核心逻辑
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import GitLoader
# 加载代码库
loader = GitLoader(repo_path="./my_project", branch="main")
documents = loader.load()
# 向量化存储
vectorstore = Chroma.from_documents(documents, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 10})
# Agent能搜索代码库来回答问题
@tool
def search_codebase(query: str) -> str:
"""搜索代码库中的相关代码"""
docs = retriever.invoke(query)
return "\n\n".join([d.page_content for d in docs])
Agent连接数据库,自动生成SQL查询,运行分析,生成可视化图表,撰写数据洞察报告,每日定时推送给业务团队。
集成CRM、知识库、工单系统的客服Agent:理解用户问题 → 查询历史订单 → 搜索解决方案 → 自动处理退款/换货 → 未解决时升级人工客服。
能访问日历、邮件、任务管理、浏览器的个人Agent:自动安排会议、整理邮件摘要、跟踪任务进度、搜索信息并总结。
from langchain_community.chat_message_histories import SQLChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_anthropic import ChatAnthropic
# 持久化对话历史(SQLite)
def get_session_history(session_id: str):
return SQLChatMessageHistory(session_id, "sqlite:///agent_memory.db")
llm = ChatAnthropic(model="claude-3-7-sonnet-20250219")
# 带持久记忆的Agent
agent_with_memory = RunnableWithMessageHistory(
llm,
get_session_history,
input_messages_key="input",
history_messages_key="history"
)
# 第一次对话
response1 = agent_with_memory.invoke(
{"input": "我叫张伟,是一名Python开发者"},
config={"configurable": {"session_id": "user_001"}}
)
# 第二次对话(记住了上次信息)
response2 = agent_with_memory.invoke(
{"input": "我刚才说我叫什么名字?"},
config={"configurable": {"session_id": "user_001"}}
)
# 输出:您之前告诉我您叫张伟,是一名Python开发者。
| 问题 | 解决方案 |
|---|---|
| Agent无限循环 | 设置max_iterations(通常5-15次)和超时限制 |
| 工具调用失败 | 实现retry机制 + 降级策略 |
| 幻觉导致错误操作 | 关键操作前加入人工确认节点(HITL) |
| 上下文窗口溢出 | 实现消息摘要压缩或分块处理 |
| API费用过高 | 轻量任务用GPT-4o mini/Claude Haiku,重任务才用旗舰 |