文章提供了两个功能完整的AI Agent实例代码，分别基于LangGraph和Agno框架开发，都具备对话、多步推理、工具调用和任务规划能力。文章包含详细的环境搭建指南、完整代码实现、环境配置文件、运行说明、功能特性对比和使用建议。两个框架各有优势，LangGraph适合复杂工作流控制，Agno提供高性能和简单学习曲线，可根据具体需求选择使用。代码附带详细中文注释，便于理解和扩展。

---

概述

本文档提供两个功能完整的AI agent实例代码，分别基于LangGraph框架和Agno框架开发。两个agent都具备对话、多步推理、工具调用和任务规划能力。

1. LangGraph Agent实现

---

1.1 环境搭建

# 创建虚拟环境python -m venv langgraph_envsource langgraph_env/bin/activate  # Windows: langgraph_env\Scripts\activate# 安装依赖pip install -U langgraph langchain-openai langchain-communitypip install tavily-python duckduckgo-search yfinancepip install python-dotenv pydantic typing-extensions# 设置环境变量export OPENAI_API_KEY="sk-your-key-here"export TAVILY_API_KEY="tvly-your-key-here"

1.2 完整代码实现

"""LangGraph智能助手 - 完整实现功能：对话、多步推理、工具调用、任务规划作者：AI Assistant版本：1.0"""import osfrom typing importList, Dict, Any, Literal, Annotatedfrom typing_extensions import TypedDictfrom dataclasses import dataclassimport jsonfrom datetime import datetime# LangGraph相关导入from langgraph.graph import StateGraph, START, ENDfrom langgraph.prebuilt import ToolNode, tools_conditionfrom langgraph.checkpoint.memory import MemorySaverfrom langgraph.types import Command# LangChain相关导入from langchain_openai import ChatOpenAIfrom langchain_core.messages import HumanMessage, AIMessage, ToolMessagefrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.tools import tool# 工具导入from tavily import TavilyClientfrom duckduckgo_search import DDGSimport yfinance as yf# 环境配置from dotenv import load_dotenvload_dotenv()# ========== 状态定义 ==========classAgentState(TypedDict):"""智能体状态定义"""    messages: List[Any]  # 消息历史    current_task: str# 当前任务    task_type: str# 任务类型：simple_chat, research, analysis, planning    search_results: List[Dict]  # 搜索结果    analysis_results: Dict[str, Any]  # 分析结果    reasoning_steps: List[str]  # 推理步骤    plan: List[Dict]    # 任务计划    completed_steps: List[str]  # 已完成步骤    context: Dict[str, Any]  # 上下文信息    next_action: str# 下一步动作# ========== 工具定义 ==========@tooldefweb_search(query: str, max_results: int = 5) -> str:"""    使用Tavily搜索引擎进行网络搜索    Args:        query: 搜索查询        max_results: 最大结果数    Returns:        str: 搜索结果摘要    """try:        client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))        response = client.search(            query=query,            search_depth="advanced",            max_results=max_results,            include_answer=True        )        results = []for result in response.get("results", []):            results.append(f"标题: {result['title']}\n内容: {result['content'][:300]}...\nURL: {result['url']}")return"\n\n---\n\n".join(results)except Exception as e:returnf"搜索失败: {str(e)}"@tooldefduckduckgo_search(query: str, max_results: int = 5) -> str:"""    使用DuckDuckGo进行隐私友好的搜索    Args:        query: 搜索查询        max_results: 最大结果数    Returns:        str: 搜索结果    """try:        ddgs = DDGS()        results = list(ddgs.text(query, max_results=max_results))        formatted_results = []for result in results:            formatted_results.append(f"标题: {result['title']}\n摘要: {result['body']}\nURL: {result['href']}")return"\n\n---\n\n".join(formatted_results)except Exception as e:returnf"DuckDuckGo搜索失败: {str(e)}"@tooldefget_stock_info(symbol: str) -> str:"""    获取股票信息    Args:        symbol: 股票代码 (如: AAPL, TSLA)    Returns:        str: 股票信息    """try:        stock = yf.Ticker(symbol)        info = stock.info        hist = stock.history(period="1d")        result = f"""股票代码: {symbol}公司名称: {info.get('longName', 'N/A')}当前价格: ${info.get('currentPrice', 'N/A')}市值: ${info.get('marketCap', 'N/A'):,}52周最高: ${info.get('fiftyTwoWeekHigh', 'N/A')}52周最低: ${info.get('fiftyTwoWeekLow', 'N/A')}行业: {info.get('industry', 'N/A')}简介: {info.get('longBusinessSummary', 'N/A')[:200]}...        """return resultexcept Exception as e:returnf"获取股票信息失败: {str(e)}"@tooldefcalculator(expression: str) -> str:"""    执行数学计算    Args:        expression: 数学表达式    Returns:        str: 计算结果    """try:# 安全的数学表达式计算        allowed_chars = set('0123456789+-*/(). ')ifnotall(c in allowed_chars for c in expression):return"计算表达式包含不允许的字符"        result = eval(expression)returnf"计算结果: {expression} = {result}"except Exception as e:returnf"计算错误: {str(e)}"# 工具列表tools = [web_search, duckduckgo_search, get_stock_info, calculator]# ========== 核心Agent类 ==========classLangGraphAgent:"""基于LangGraph的智能助手"""def__init__(self, model_name: str = "gpt-4o"):"""        初始化智能助手        Args:            model_name: 使用的模型名称        """        self.llm = ChatOpenAI(model=model_name, temperature=0.1)        self.memory = MemorySaver()# 创建工具节点        self.tool_node = ToolNode(tools)# 构建图        self.app = self._build_graph()print(f"✅ LangGraph智能助手初始化完成，使用模型: {model_name}")def_build_graph(self) -> StateGraph:"""构建LangGraph工作流图"""# 创建状态图        workflow = StateGraph(AgentState)# 添加节点        workflow.add_node("classifier", self._classify_task)          # 任务分类        workflow.add_node("simple_chat", self._simple_chat)           # 简单对话        workflow.add_node("planner", self._create_plan)              # 任务规划        workflow.add_node("researcher", self._research_node)          # 研究节点        workflow.add_node("analyzer", self._analysis_node)           # 分析节点        workflow.add_node("reasoning", self._reasoning_node)         # 推理节点        workflow.add_node("tools", self.tool_node)                   # 工具调用        workflow.add_node("synthesizer", self._synthesize_results)   # 结果综合# 添加边和条件路由        workflow.add_edge(START, "classifier")# 从分类器到不同处理路径        workflow.add_conditional_edges("classifier",            self._route_after_classification,            {"simple_chat": "simple_chat","research": "planner", "analysis": "planner","planning": "planner"            }        )# 简单对话路径        workflow.add_conditional_edges("simple_chat",            tools_condition,            {"tools": "tools","end": END            }        )# 工具调用后回到简单对话        workflow.add_edge("tools", "simple_chat")# 规划后的路径        workflow.add_edge("planner", "researcher")        workflow.add_edge("researcher", "analyzer")          workflow.add_edge("analyzer", "reasoning")        workflow.add_edge("reasoning", "synthesizer")        workflow.add_edge("synthesizer", END)return workflow.compile(checkpointer=self.memory)def_classify_task(self, state: AgentState) -> Dict[str, Any]:"""任务分类节点"""        messages = state.get("messages", [])ifnot messages:return {"task_type": "simple_chat"}        last_message = messages[-1].content if messages else""# 使用LLM进行任务分类        classification_prompt = f"""        分析用户的请求，将其分类为以下类型之一：        1. simple_chat - 简单对话、问候、基本问答        2. research - 需要搜索信息的研究任务        3. analysis - 需要深度分析的任务        4. planning - 需要制定计划或策略的任务        用户请求: {last_message}        只返回类型名称，不要其他文字。        """        response = self.llm.invoke([HumanMessage(content=classification_prompt)])        task_type = response.content.strip().lower()# 验证返回值        valid_types = ["simple_chat", "research", "analysis", "planning"]if task_type notin valid_types:            task_type = "simple_chat"return {"task_type": task_type,"current_task": last_message,"reasoning_steps": [f"任务分类: {task_type}"]        }def_simple_chat(self, state: AgentState) -> Dict[str, Any]:"""简单对话处理节点"""        messages = state.get("messages", [])# 构建对话提示        system_prompt = """        你是一个友好、有用的AI助手。你可以：        1. 进行日常对话        2. 回答一般性问题        3. 在需要时调用工具获取信息        4. 进行数学计算        5. 搜索最新信息        如果用户的问题需要搜索最新信息、股票数据或数学计算，请调用相应的工具。        """        full_messages = [{"role": "system", "content": system_prompt}] + [            {"role": "human"ifisinstance(msg, HumanMessage) else"ai", "content": msg.content} for msg in messages        ]        response = self.llm.bind_tools(tools).invoke(full_messages)return {"messages": messages + [response]        }def_create_plan(self, state: AgentState) -> Dict[str, Any]:"""任务规划节点"""        current_task = state.get("current_task", "")        task_type = state.get("task_type", "")        planning_prompt = f"""        为以下{task_type}任务制定详细的执行计划：        任务: {current_task}        请将任务分解为具体的步骤，每个步骤包括：        1. 步骤名称        2. 具体行动        3. 预期结果        以JSON格式返回计划，格式如下：        {{            "steps": [                {{                    "name": "步骤名称",                    "action": "具体行动",                     "expected_result": "预期结果"                }}            ]        }}        """        response = self.llm.invoke([HumanMessage(content=planning_prompt)])try:            plan_data = json.loads(response.content)            plan = plan_data.get("steps", [])except:# 如果JSON解析失败，创建默认计划            plan = [                {"name": "信息收集","action": "搜索相关信息","expected_result": "获得背景信息"                },                {"name": "深度分析", "action": "分析收集到的信息","expected_result": "形成初步结论"                },                {"name": "推理综合","action": "进行逻辑推理","expected_result": "得出最终答案"                }            ]        reasoning_steps = state.get("reasoning_steps", [])        reasoning_steps.append(f"制定了包含{len(plan)}个步骤的执行计划")return {"plan": plan,"reasoning_steps": reasoning_steps        }def_research_node(self, state: AgentState) -> Dict[str, Any]:"""研究信息收集节点"""        current_task = state.get("current_task", "")        plan = state.get("plan", [])# 执行研究步骤        search_queries = self._generate_search_queries(current_task)        search_results = []for query in search_queries[:3]:  # 限制搜索查询数量try:# 使用Tavily搜索                result = web_search.invoke({"query": query, "max_results": 3})                search_results.append({"query": query,"results": result                })except:# 备用DuckDuckGo搜索try:                    result = duckduckgo_search.invoke({"query": query, "max_results": 3})                    search_results.append({"query": query, "results": result                    })except:pass        reasoning_steps = state.get("reasoning_steps", [])        reasoning_steps.append(f"完成信息收集，获得{len(search_results)}组搜索结果")return {"search_results": search_results,"reasoning_steps": reasoning_steps        }def_analysis_node(self, state: AgentState) -> Dict[str, Any]:"""分析处理节点"""        search_results = state.get("search_results", [])        current_task = state.get("current_task", "")# 整合搜索结果        all_info = ""for result in search_results:            all_info += f"查询: {result['query']}\n结果: {result['results']}\n\n"        analysis_prompt = f"""        基于收集到的信息，对以下任务进行深度分析：        任务: {current_task}        收集到的信息:{all_info}        请进行多角度分析：        1. 关键信息总结        2. 重要发现和趋势        3. 不同观点对比        4. 潜在影响和意义        请以结构化方式组织分析结果。        """        response = self.llm.invoke([HumanMessage(content=analysis_prompt)])        analysis_results = {"summary": "分析完成","content": response.content,"timestamp": datetime.now().isoformat()        }        reasoning_steps = state.get("reasoning_steps", [])        reasoning_steps.append("完成信息分析，形成结构化见解")return {"analysis_results": analysis_results,"reasoning_steps": reasoning_steps        }def_reasoning_node(self, state: AgentState) -> Dict[str, Any]:"""推理思考节点"""        analysis_results = state.get("analysis_results", {})        current_task = state.get("current_task", "")        reasoning_prompt = f"""        基于分析结果，请进行逐步推理来回答用户的问题：        原始问题: {current_task}        分析结果: {analysis_results.get('content', '')}        请使用以下推理框架：        1. 问题理解：重新阐述核心问题        2. 关键因素：识别影响答案的关键因素          3. 逻辑推理：           - 步骤1：[基于证据A得出结论1]           - 步骤2：[基于证据B得出结论2]           - 步骤3：[综合结论1和2得出最终结论]        4. 验证检查：检验结论的合理性        5. 最终答案：清晰明确的回答        请严格按照这个结构进行推理。        """        response = self.llm.invoke([HumanMessage(content=reasoning_prompt)])        reasoning_steps = state.get("reasoning_steps", [])        reasoning_steps.append("完成逻辑推理分析")return {"reasoning_steps": reasoning_steps,"reasoning_content": response.content        }def_synthesize_results(self, state: AgentState) -> Dict[str, Any]:"""结果综合节点"""        current_task = state.get("current_task", "")        analysis_results = state.get("analysis_results", {})        reasoning_content = state.get("reasoning_content", "")        reasoning_steps = state.get("reasoning_steps", [])        synthesis_prompt = f"""        请将所有分析和推理结果综合成一个完整、清晰的最终回答：        原始问题: {current_task}        分析结果: {analysis_results.get('content', '')}        推理过程: {reasoning_content}        请提供：        1. 直接明确的答案        2. 支持证据和理由        3. 相关的背景信息        4. 如果有的话，建议下一步行动        请以自然、易懂的方式组织回答。        """        final_response = self.llm.invoke([HumanMessage(content=synthesis_prompt)])        reasoning_steps.append("完成结果综合，生成最终回答")        messages = state.get("messages", [])        messages.append(AIMessage(content=final_response.content))return {"messages": messages,"reasoning_steps": reasoning_steps        }def_generate_search_queries(self, task: str) -> List[str]:"""生成搜索查询"""        query_prompt = f"""        为以下任务生成2-3个有效的搜索查询：        任务: {task}        请生成能够获得相关、准确信息的搜索查询。        每行一个查询，不要编号。        """        response = self.llm.invoke([HumanMessage(content=query_prompt)])        queries = [q.strip() for q in response.content.split('\n') if q.strip()]return queries[:3]  # 限制查询数量def_route_after_classification(self, state: AgentState) -> str:"""分类后的路由决策"""        task_type = state.get("task_type", "simple_chat")return task_typedefchat(self, message: str, session_id: str = "default") -> str:"""        与智能助手对话        Args:            message: 用户消息            session_id: 会话ID，用于维护对话历史        Returns:            str: 助手回复        """try:# 准备初始状态            initial_state = {"messages": [HumanMessage(content=message)],"current_task": "","task_type": "","search_results": [],"analysis_results": {},"reasoning_steps": [],"plan": [],"completed_steps": [],"context": {},"next_action": ""            }# 执行图处理            config = {"configurable": {"thread_id": session_id}}            final_state = self.app.invoke(initial_state, config=config)# 提取回复            messages = final_state.get("messages", [])if messages andisinstance(messages[-1], AIMessage):return messages[-1].contentelse:return"抱歉，处理过程中出现了问题。"except Exception as e:returnf"对话处理失败: {str(e)}"defget_reasoning_steps(self, session_id: str = "default") -> List[str]:"""获取推理步骤"""try:            config = {"configurable": {"thread_id": session_id}}            state = self.app.get_state(config)return state.values.get("reasoning_steps", [])except:return []# ========== 使用示例 ==========defdemo_langgraph_agent():"""LangGraph Agent 演示"""print("\n" + "="*60)print("🚀 LangGraph智能助手演示")print("="*60)# 创建智能助手    agent = LangGraphAgent()# 测试场景    test_cases = [        {"name": "简单对话","message": "你好，请介绍一下自己的能力"        },        {"name": "数学计算", "message": "请帮我计算 (125 + 75) * 2 - 50"        },        {"name": "信息研究","message": "请调研一下2024年人工智能的最新发展趋势"        },        {"name": "股票分析","message": "分析一下特斯拉(TSLA)股票的投资价值"        }    ]for i, test inenumerate(test_cases, 1):print(f"\n📋 测试{i}：{test['name']}")print(f"问题：{test['message']}")print("\n💭 处理中...")        response = agent.chat(test['message'], session_id=f"demo_{i}")print(f"\n🤖 回答：")print(response)# 显示推理步骤        steps = agent.get_reasoning_steps(f"demo_{i}")if steps:print(f"\n🧠 推理步骤：")for step in steps:print(f"  • {step}")print("\n" + "-"*40)print("\n✅ LangGraph Agent演示完成！")if __name__ == "__main__":    demo_langgraph_agent()

---

2. Agno Agent实现

---

2.1 环境搭建

# 创建虚拟环境python -m venv agno_envsource agno_env/bin/activate  # Windows: agno_env\Scripts\activate# 安装依赖pip install -U agno openai anthropicpip install tavily-python duckduckgo-search yfinancepip install python-dotenv# 设置环境变量export OPENAI_API_KEY="sk-your-key-here"export ANTHROPIC_API_KEY="sk-ant-your-key-here"export TAVILY_API_KEY="tvly-your-key-here"

2.2 完整代码实现

"""Agno智能助手 - 完整实现功能：对话、多步推理、工具调用、任务规划作者：AI Assistant版本：1.0"""import osfrom typing importList, Dict, Any, Optionalfrom dataclasses import dataclassimport jsonfrom datetime import datetime# Agno框架导入from agno.agent import Agentfrom agno.team import Team  from agno.workflow import Workflowfrom agno.models.openai import OpenAIChatfrom agno.models.anthropic import Claudefrom agno.tools import Toolfrom agno.tools.duckduckgo import DuckDuckGoToolsfrom agno.tools.yfinance import YFinanceToolsfrom agno.tools.reasoning import ReasoningTools# 其他导入from tavily import TavilyClientimport yfinance as yf# 环境配置from dotenv import load_dotenvload_dotenv()# ========== 自定义工具类 ==========classTavilySearchTool(Tool):"""Tavily搜索工具"""def__init__(self):super().__init__(            name="tavily_search",            description="使用Tavily搜索引擎进行高质量网络搜索，专为AI优化",        )        self.client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))defrun(self, query: str, max_results: int = 5) -> str:"""执行搜索"""try:            response = self.client.search(                query=query,                search_depth="advanced",                 max_results=max_results,                include_answer=True            )            results = []for result in response.get("results", []):                results.append(f"标题: {result['title']}\n"f"内容: {result['content'][:300]}...\n"f"URL: {result['url']}\n"                )return"\n---\n".join(results)except Exception as e:returnf"Tavily搜索失败: {str(e)}"classAdvancedCalculatorTool(Tool):"""高级计算器工具"""def__init__(self):super().__init__(            name="advanced_calculator",            description="执行数学计算、统计分析等运算",        )defrun(self, expression: str) -> str:"""执行计算"""try:# 安全计算            allowed_chars = set('0123456789+-*/().eE ')            allowed_words = {'sin', 'cos', 'tan', 'log', 'exp', 'sqrt', 'pi', 'abs'}# 基本安全检查            clean_expr = expression.replace(' ', '')ifnotall(c in allowed_chars or c.isalpha() for c in clean_expr):return"表达式包含不允许的字符"# 简单数学运算import math            safe_dict = {"__builtins__": {},"sin": math.sin, "cos": math.cos, "tan": math.tan,"log": math.log, "exp": math.exp, "sqrt": math.sqrt,"pi": math.pi, "abs": abs, "pow": pow            }            result = eval(expression, safe_dict)returnf"计算结果: {expression} = {result}"except Exception as e:returnf"计算错误: {str(e)}"classEnhancedStockTool(Tool):"""增强股票分析工具"""def__init__(self):super().__init__(            name="enhanced_stock_analysis",            description="获取详细的股票信息、财务数据和技术分析",        )defrun(self, symbol: str) -> str:"""获取股票详细信息"""try:            stock = yf.Ticker(symbol)            info = stock.info            hist = stock.history(period="1mo")# 基本信息            basic_info = f"""📊 股票代码: {symbol}🏢 公司名称: {info.get('longName', 'N/A')}💰 当前价格: ${info.get('currentPrice', 'N/A')}📈 今日涨跌: {info.get('regularMarketChangePercent', 'N/A')}%🏆 市值: ${info.get('marketCap', 0):,}📊 市盈率: {info.get('trailingPE', 'N/A')}🎯 52周最高: ${info.get('fiftyTwoWeekHigh', 'N/A')}📉 52周最低: ${info.get('fiftyTwoWeekLow', 'N/A')}🏭 行业: {info.get('industry', 'N/A')}🌍 板块: {info.get('sector', 'N/A')}            """# 技术指标ifnot hist.empty:                current_price = hist['Close'][-1]                sma_20 = hist['Close'].rolling(window=20).mean()[-1]                volatility = hist['Close'].pct_change().std() * 100                technical_info = f"""📈 技术指标:  - 20日均线: ${sma_20:.2f}  - 当前价格vs均线: {'上方'if current_price > sma_20 else'下方'}  - 波动率: {volatility:.2f}%                """else:                technical_info = "📈 技术指标: 数据不足"# 公司描述            description = info.get('longBusinessSummary', 'N/A')iflen(description) > 300:                description = description[:300] + "..."returnf"{basic_info}\n{technical_info}\n\n📝 公司简介:\n{description}"except Exception as e:returnf"获取股票信息失败: {str(e)}"# ========== 多智能体团队 ==========classResearchTeam:"""研究团队 - 多智能体协作"""def__init__(self, model_provider: str = "openai"):"""        初始化研究团队        Args:            model_provider: 模型提供商 ("openai" 或 "anthropic")        """# 选择基础模型if model_provider == "anthropic":            base_model = Claude(id="claude-3-5-sonnet-20241022")else:            base_model = OpenAIChat(id="gpt-4o")# 创建专门化的智能体        self.web_researcher = Agent(            name="网络研究专家",            role="负责搜索和收集网络信息",            model=base_model,            tools=[                TavilySearchTool(),                DuckDuckGoTools(),            ],            instructions=["专注于搜索高质量、权威的信息源","总是提供信息来源链接","优先使用最新的信息","对搜索结果进行初步筛选和整理"            ],            markdown=True,        )        self.financial_analyst = Agent(            name="金融分析师",             role="负责金融数据分析和股票研究",            model=base_model,            tools=[                EnhancedStockTool(),                YFinanceTools(                    stock_price=True,                    company_info=True,                    analyst_recommendations=True,                    company_news=True                ),                AdvancedCalculatorTool(),            ],            instructions=["提供深度的金融分析","使用表格展示关键数据","包含风险评估","给出明确的投资建议"            ],            markdown=True,        )        self.reasoning_expert = Agent(            name="推理分析专家",            role="负责逻辑推理和综合分析",            model=base_model,            tools=[ReasoningTools(add_instructions=True)],            instructions=["使用逐步推理解决复杂问题","展示完整的思考过程","从多个角度分析问题", "得出有逻辑支撑的结论"            ],            markdown=True,            show_tool_calls=True,        )# 创建协作团队        self.team = Team(            model=base_model,            members=[self.web_researcher, self.financial_analyst, self.reasoning_expert],            instructions=["团队协作完成复杂任务","每个成员发挥自己的专长","最终提供全面、准确的答案","保持逻辑清晰和结构化"            ],            show_tool_calls=True,            markdown=True,        )print(f"✅ Agno研究团队初始化完成，使用{model_provider}模型")defresearch(self, topic: str) -> str:"""执行研究任务"""return self.team.run(f"请深入研究: {topic}")# ========== 智能工作流 ==========classIntelligentWorkflow(Workflow):"""智能工作流 - 任务规划和执行"""def__init__(self, model_provider: str = "openai"):super().__init__(name="intelligent_workflow")# 选择模型if model_provider == "anthropic":            model = Claude(id="claude-3-5-sonnet-20241022")else:            model = OpenAIChat(id="gpt-4o")# 任务分类器        self.classifier = Agent(            name="任务分类器",            role="分析和分类用户请求",            model=model,            instructions=["分析用户请求的类型和复杂度","决定最适合的处理方式","提供清晰的分类结果"            ]        )# 通用对话助手        self.chat_assistant = Agent(            name="对话助手",             role="处理日常对话和简单问答",            model=model,            tools=[AdvancedCalculatorTool()],            instructions=["提供友好、有用的回答","对于简单问题直接回答","必要时使用工具辅助"            ]        )# 研究团队        self.research_team = ResearchTeam(model_provider)print(f"✅ 智能工作流初始化完成")defrun(self, user_request: str) -> str:"""执行工作流"""# Step 1: 任务分类        classification = self.classifier.run(f"""        请分析以下用户请求，并分类：        用户请求: {user_request}        分类选项:        1. simple_chat - 简单对话、问候、基本问答        2. calculation - 数学计算        3. research - 需要搜索信息的研究任务        4. financial_analysis - 金融、股票相关分析        5. complex_analysis - 需要深度分析和推理的复杂任务        请只返回分类名称，不要其他内容。        """)        task_type = classification.strip().lower()# Step 2: 根据分类执行相应处理if task_type in ["simple_chat", "calculation"]:return self.chat_assistant.run(user_request)elif task_type in ["research", "financial_analysis", "complex_analysis"]:return self.research_team.research(user_request)else:# 默认使用聊天助手return self.chat_assistant.run(user_request)# ========== 主要Agent类 ==========classAgnoAgent:"""基于Agno的智能助手"""def__init__(self, model_provider: str = "openai"):"""        初始化Agno智能助手        Args:            model_provider: 模型提供商 ("openai" 或 "anthropic")        """        self.model_provider = model_provider# 创建工作流        self.workflow = IntelligentWorkflow(model_provider)# 会话历史        self.chat_history: Dict[str, List[str]] = {}print(f"🎉 Agno智能助手初始化完成！")print(f"📍 模型提供商: {model_provider}")print(f"🔧 功能: 对话、搜索、分析、推理、金融数据")defchat(self, message: str, session_id: str = "default") -> str:"""        与智能助手对话        Args:            message: 用户消息            session_id: 会话ID        Returns:            str: 助手回复        """try:# 维护会话历史if session_id notin self.chat_history:                self.chat_history[session_id] = []# 记录用户消息            self.chat_history[session_id].append(f"用户: {message}")# 执行工作流处理            response = self.workflow.run(message)# 记录助手回复            self.chat_history[session_id].append(f"助手: {response}")return responseexcept Exception as e:            error_msg = f"处理失败: {str(e)}"print(f"❌ {error_msg}")return error_msgdefget_chat_history(self, session_id: str = "default") -> List[str]:"""获取聊天历史"""return self.chat_history.get(session_id, [])defclear_history(self, session_id: str = "default"):"""清除聊天历史"""if session_id in self.chat_history:del self.chat_history[session_id]print(f"✅ 已清除会话 {session_id} 的历史记录")defswitch_model(self, new_provider: str):"""切换模型提供商"""if new_provider in ["openai", "anthropic"]:            self.model_provider = new_provider            self.workflow = IntelligentWorkflow(new_provider)print(f"✅ 已切换到 {new_provider} 模型")else:print("❌ 不支持的模型提供商，支持: openai, anthropic")# ========== 使用示例和演示 ==========defdemo_agno_agent():"""Agno Agent 演示"""print("\n" + "="*60)print("🚀 Agno智能助手演示") print("="*60)# 创建智能助手 - 可以选择不同模型print("选择模型提供商：")print("1. OpenAI (gpt-4o)")print("2. Anthropic (claude-3-5-sonnet)")    choice = input("请输入选择 (1 或 2，默认1): ").strip()    model_provider = "anthropic"if choice == "2"else"openai"    agent = AgnoAgent(model_provider=model_provider)# 测试场景    test_cases = [        {"name": "简单对话","message": "你好！请介绍一下你的能力",        },        {"name": "数学计算","message": "计算复合增长率：初值100，年增长率15%，5年后的值是多少？"        },        {"name": "信息研究", "message": "2024年AI大模型的最新发展趋势是什么？"        },        {"name": "股票分析","message": "分析微软(MSFT)的股票投资价值，包括基本面和技术面"        },        {"name": "复杂推理","message": "如果要在2025年开始投资AI相关股票，应该考虑哪些因素？请给出详细的分析框架"        }    ]for i, test inenumerate(test_cases, 1):print(f"\n{'='*40}")print(f"📋 测试 {i}: {test['name']}")print(f"{'='*40}")print(f"❓ 问题: {test['message']}")print("\n💭 处理中...")# 执行测试        response = agent.chat(test['message'], session_id=f"demo_{i}")print(f"\n🤖 回答:")print("-" * 40)print(response)print("-" * 40)# 询问是否继续if i < len(test_cases):            continue_test = input(f"\n继续下一个测试？(y/n, 默认y): ").strip().lower()if continue_test == 'n':breakprint(f"\n✅ Agno Agent演示完成！")# 显示会话历史示例print(f"\n📚 会话历史示例 (最后一个会话):")    history = agent.get_chat_history("demo_1")for entry in history[-4:]:  # 显示最后4条记录print(f"  {entry[:100]}...")definteractive_mode():"""交互模式"""print("\n" + "="*60)print("🎯 进入交互模式")print("="*60)# 选择模型print("选择模型提供商：")print("1. OpenAI (gpt-4o)")  print("2. Anthropic (claude-3-5-sonnet)")    choice = input("请输入选择 (1 或 2，默认1): ").strip()    model_provider = "anthropic"if choice == "2"else"openai"    agent = AgnoAgent(model_provider=model_provider)print("\n✨ 智能助手已就绪！输入 'quit' 退出，'clear' 清除历史")print(f"💡 提示：我可以帮你搜索信息、分析数据、计算数学题、分析股票等")    session_id = "interactive"whileTrue:try:            user_input = input(f"\n💬 你: ").strip()if user_input.lower() == 'quit':print("👋 再见！")breakelif user_input.lower() == 'clear':                agent.clear_history(session_id)continueelif user_input.lower() == 'history':                history = agent.get_chat_history(session_id)print("\n📚 对话历史:")for entry in history:print(f"  {entry}")continueelifnot user_input:continueprint(f"\n🤖 助手: 思考中...")            response = agent.chat(user_input, session_id)print(f"\n🤖 助手: {response}")except KeyboardInterrupt:print(f"\n\n👋 再见！")breakexcept Exception as e:print(f"❌ 出错了: {str(e)}")if __name__ == "__main__":print("🎊 欢迎使用Agno智能助手！")print("\n选择运行模式:")print("1. 演示模式 (自动运行测试案例)")print("2. 交互模式 (手动对话)")    mode = input("\n请选择模式 (1 或 2，默认1): ").strip()if mode == "2":        interactive_mode()else:        demo_agno_agent()

---

3. 环境配置文件

---

3.1 .env 环境变量文件

# API密钥配置OPENAI_API_KEY=sk-your-openai-key-hereANTHROPIC_API_KEY=sk-ant-your-anthropic-key-here  TAVILY_API_KEY=tvly-your-tavily-key-here# 可选配置GROQ_API_KEY=gsk-your-groq-key-here# Agno配置AGNO_TELEMETRY=false

3.2 requirements.txt 依赖文件

LangGraph版本:

langgraph>=0.2.76langchain-openai>=0.2.0langchain-community>=0.3.0  langchain-core>=0.3.0tavily-python>=0.5.0duckduckgo-search>=6.0.0yfinance>=0.2.0python-dotenv>=1.0.0pydantic>=2.5.0typing-extensions>=4.8.0

Agno版本:

agno>=1.7.7openai>=1.50.0anthropic>=0.34.0tavily-python>=0.5.0duckduckgo-search>=6.0.0yfinance>=0.2.0python-dotenv>=1.0.0

---

4. 运行说明

---

4.1 快速开始

1. 克隆或创建项目目录

mkdir ai_agentscd ai_agents

2. 创建环境和安装依赖

# LangGraph环境python -m venv langgraph_envsource langgraph_env/bin/activatepip install -r requirements_langgraph.txt# Agno环境  python -m venv agno_envsource agno_env/bin/activatepip install -r requirements_agno.txt

3. 配置环境变量

# 创建.env文件并添加API密钥cp .env.example .env# 编辑.env文件添加你的API密钥

4. 运行示例

# 运行LangGraph示例python langgraph_agent.py# 运行Agno示例python agno_agent.py

4.2 功能特性对比

功能特性	LangGraph Agent	Agno Agent
对话功能	✅ 支持多轮对话	✅ 支持多轮对话
工具调用	✅ 灵活的工具系统	✅ 丰富的预建工具
多步推理	✅ 复杂的图状推理	✅ 内置推理工具
任务规划	✅ 动态计划生成	✅ 智能体团队协作
搜索集成	✅ 多搜索引擎	✅ 多搜索引擎
性能	🔶 中等	✅ 高性能
学习曲线	🔶 较陡峭	✅ 相对简单
模型支持	✅ 多模型	✅ 23+模型提供商

4.3 使用建议

选择LangGraph的情况：

• 需要复杂的工作流控制
• 需要精细的状态管理
• 需要自定义复杂的推理路径
• 有图结构化处理需求

选择Agno的情况：

• 需要高性能和低延迟
• 希望快速原型开发
• 需要多智能体协作
• 希望使用预建组件

---

5. 扩展和定制

---

5.1 添加新工具

LangGraph中添加工具：

@tooldefcustom_tool(param: str) -> str:"""自定义工具描述"""# 工具逻辑return"结果"# 添加到工具列表tools.append(custom_tool)

Agno中添加工具：

classCustomTool(Tool):def__init__(self):super().__init__(name="custom_tool", description="自定义工具")defrun(self, param: str) -> str:# 工具逻辑return"结果"

5.2 自定义推理模式

两个框架都支持自定义推理模式，可以根据具体需求实现特定的推理逻辑。

5.3 部署建议

• 使用Docker容器化部署
• 配置适当的资源限制
• 实现请求缓存机制
• 添加监控和日志记录

---

6. 总结

---

本文档提供了两个完整、可运行的AI agent实现，展示了LangGraph和Agno框架的强大功能。两个框架各有优势，可以根据具体需求选择使用。代码包含详细的中文注释和完整的使用示例，便于理解和扩展。

AI时代，未来的就业机会在哪里？

答案就藏在大模型的浪潮里。从ChatGPT、DeepSeek等日常工具，到自然语言处理、计算机视觉、多模态等核心领域，技术普惠化、应用垂直化与生态开源化正催生Prompt工程师、自然语言处理、计算机视觉工程师、大模型算法工程师、AI应用产品经理等AI岗位。

掌握大模型技能，就是把握高薪未来。

那么，普通人如何抓住大模型风口？

AI技术的普及对个人能力提出了新的要求，在AI时代，持续学习和适应新技术变得尤为重要。无论是企业还是个人，都需要不断更新知识体系，提升与AI协作的能力，以适应不断变化的工作环境。

因此，这里给大家整理了一份《2025最新大模型全套学习资源》，包括2025最新大模型学习路线、大模型书籍、视频教程、项目实战、最新行业报告、面试题等，带你从零基础入门到精通，快速掌握大模型技术！

由于篇幅有限，有需要的小伙伴可以扫码获取！

1. 成长路线图&学习规划

要学习一门新的技术，作为新手一定要先学习成长路线图，方向不对，努力白费。这里，我们为新手和想要进一步提升的专业人士准备了一份详细的学习成长路线图和规划。

2. 大模型经典PDF书籍

书籍和学习文档资料是学习大模型过程中必不可少的，我们精选了一系列深入探讨大模型技术的书籍和学习文档，它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。（书籍含电子版PDF）

3. 大模型视频教程

对于很多自学或者没有基础的同学来说，书籍这些纯文字类的学习教材会觉得比较晦涩难以理解，因此，我们提供了丰富的大模型视频教程，以动态、形象的方式展示技术概念，帮助你更快、更轻松地掌握核心知识。

4. 大模型项目实战

学以致用 ，当你的理论知识积累到一定程度，就需要通过项目实战，在实际操作中检验和巩固你所学到的知识，同时为你找工作和职业发展打下坚实的基础。

5. 大模型行业报告

行业分析主要包括对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。

6. 大模型面试题

面试不仅是技术的较量，更需要充分的准备。

在你已经掌握了大模型技术之后，就需要开始准备面试，我们将提供精心整理的大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。

为什么大家都在学AI大模型？

随着AI技术的发展，企业对人才的需求从“单一技术”转向 “AI+行业”双背景。企业对人才的需求从“单一技术”转向 “AI+行业”双背景。金融+AI、制造+AI、医疗+AI等跨界岗位薪资涨幅达30%-50%。

同时很多人面临优化裁员，近期科技巨头英特尔裁员2万人，传统岗位不断缩减，因此转行AI势在必行！

这些资料有用吗？

这份资料由我们和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理，现任上海殷泊信息科技CEO，其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证，服务航天科工、国家电网等1000+企业，以第一作者在IEEE Transactions发表论文50+篇，获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的技术人员，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

大模型全套学习资料已整理打包，有需要的小伙伴可以微信扫描下方CSDN官方认证二维码，免费领取【保证100%免费】