AI Agents 全栈指南：Nick Saraev 2小时大师班深度解析

from langchain.tools import Tool
from typing import Optional

class WebSearchTool:
    """搜索工具的包装类"""
    
    def __init__(self, search_api):
        self.search_api = search_api
    
    def run(self, query: str, max_results: int = 5) -> str:
        """执行搜索并返回格式化结果"""
        try:
            results = self.search_api.search(query, num_results=max_results)
            
            # 格式化结果
            formatted = []
            for i, r in enumerate(results, 1):
                formatted.append(f"{i}. {r.title}\n   {r.snippet}\n   链接: {r.url}")
            
            return "\n\n".join(formatted)
            
        except RateLimitError:
            # 限流时使用备用方案
            return "搜索服务暂时限流，请稍后再试"
        except NetworkError:
            return "网络连接失败，请检查网络"

# 创建 LangChain Tool
web_search_tool = Tool(
    name="web_search",
    func=WebSearchTool(search_api).run,
    description="""
    搜索互联网获取最新信息。

    使用场景：
    - 需要获取实时数据时（天气、新闻、股价）
    - 需要验证不确定的事实时
    - 需要了解最新技术进展时

    输入：搜索关键词（字符串）
    输出：搜索结果列表，包含标题、摘要和链接

    示例：
    输入："2026年最新的 AI Agent 框架"
    输出：["1. AutoGPT 0.5 发布...\n   链接: ...", ...]
    """
)

from langchain.chains import RetrievalQA
from langchain.retrievers import EnsembleRetriever

class RAGAgent:
    """使用 RAG 增强的 Agent"""
    
    def __init__(self, llm, vectorstores: List):
        # 多源检索器
        retrievers = [vs.as_retriever() for vs in vectorstores]
        self.retriever = EnsembleRetriever(
            retrievers=retrievers,
            weights=[0.5, 0.3, 0.2]
        )
        
        # RAG Chain
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=llm,
            chain_type="stuff",
            retriever=self.retriever,
            return_source_documents=True
        )
    
    def query_with_context(self, query: str) -> Dict[str, Any]:
        """使用 RAG 查询，支持引用来源"""
        result = self.qa_chain({"query": query})
        
        return {
            "answer": result["result"],
            "sources": [
                {
                    "content": doc.page_content[:200],
                    "metadata": doc.metadata
                }
                for doc in result["source_documents"]
            ]
        }

from typing import List
from langchain.agents import Agent, AgentExecutor

class SupervisorAgent:
    """Supervisor 模式的主控 Agent"""
    
    def __init__(self, llm, sub_agents: List[Agent]):
        self.llm = llm
        self.sub_agents = {a.name: a for a in sub_agents}
    
    def run(self, task: str) -> str:
        # 1. 分析任务，决定调用哪些子 Agent
        plan = self._plan_subtasks(task)
        
        # 2. 依次执行子任务
        results = {}
        for subtask, agent_name in plan:
            agent = self.sub_agents[agent_name]
            result = agent.run(subtask)
            results[agent_name] = result
        
        # 3. 整合结果
        return self._synthesize(task, results)

from crewai import Agent, Task, Crew

# 定义多个 Agent
researcher = Agent(
    role="研究员",
    goal="收集相关信息",
    backstory="你是一位专业研究员，擅长信息收集和分析"
)

writer = Agent(
    role="作家",
    goal="撰写报告",
    backstory="你是一位专业作家，擅长清晰表达复杂概念"
)

reviewer = Agent(
    role="审稿人",
    goal="审核并优化内容",
    backstory="你是一位资深编辑，对内容质量有严格要求"
)

# 定义任务
tasks = [
    Task(
        description="研究 AI Agent 最新进展",
        agent=researcher
    ),
    Task(
        description="根据研究结果撰写报告",
        agent=writer
    ),
    Task(
        description="审核报告质量",
        agent=reviewer
    )
]

# 创建 Crew 并执行
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=tasks,
    verbose=True
)

result = crew.kickoff()

# 生产级 Agent 优化示例
from langchain.callbacks import get_openai_callback

class OptimizedAgent:
    def __init__(self):
        self.llm = ChatOpenAI(
            model="gpt-4o-mini",  # 简单任务用小模型
            temperature=0.7
        )
    
    def run(self, task: str) -> str:
        # 根据任务复杂度选择模型
        if self._is_complex_task(task):
            return self._run_with_model(task, "gpt-4o")
        else:
            return self._run_with_model(task, "gpt-4o-mini")

# Agent 权限控制
class SecureAgent:
    def __init__(self):
        self.allowed_tools = ["web_search", "calculator"]
        self.denied_tools = ["delete_file", "send_email"]
    
    def can_use_tool(self, tool_name: str) -> bool:
        if tool_name in self.denied_tools:
            return False
        if tool_name in self.allowed_tools:
            return True
        return False  # 默认拒绝未明确允许的工具

import logging
from langchain.callbacks import StdOutCallbackHandler

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

class MonitoredAgent:
    def run(self, task: str) -> str:
        logger = logging.getLogger("AgentMonitor")
        
        # 记录开始
        logger.info(f"任务开始: {task[:50]}...")
        
        # 执行
        with get_openai_callback() as cb:
            result = self.agent.run(task)
            
            # 记录完成
            logger.info(f"任务完成，消耗: ${cb.total_cost:.4f}")
        
        return result