零基础到高手：可落地Agent全栈教程（全案例+可运行代码）

本教程全程大白话讲解+逐行可运行代码+从易到难的落地案例，帮你从0基础小白，快速掌握Agent核心逻辑，最终能开发生产级商用Agent。

前言：先搞懂最核心的2个问题

1. 大白话定义Agent

Agent（智能体）= 有自主思考、规划、工具使用、记忆、反思能力的AI。
和普通ChatGPT对话的核心区别：

  • 普通LLM：你问一句，它答一句，只会被动输出，不会主动做事
  • Agent：你给一个目标，它自己拆解步骤、调用工具、执行纠错、循环迭代，直到完成目标，全程不用你插手
    在这里插入图片描述

2. 学习路线图（精准可控的成长路径）

阶段 核心目标 学习周期 通关标准
入门篇 吃透Agent底层逻辑，跑通第一个可运行Agent 2天 能自主开发单/多工具Agent，理解Function Call核心
进阶篇 掌握Agent标准范式，开发能解决真实问题的实用Agent 7天 能实现带规划、记忆、RAG能力的商用级单Agent
高手篇 掌握多智能体协作，开发生产级可部署的Agent系统 14天 能搭建Multi-Agent系统，完成复杂任务闭环，上线可用产品

3. 前置准备（零门槛，5分钟搞定）

  1. 环境安装：Python 3.10+（官网下载安装，勾选Add to PATH）
  2. 依赖库安装，cmd执行：
pip install python-dotenv requests dashscope chromadb langchain crewai streamlit
  1. API Key申请（国内用户优先选通义千问，免费额度足够学习使用）
    • 通义千问API Key：阿里云百炼 注册获取
    • 备选：OpenAI API Key、文心一言API Key，教程代码可无缝适配

第一部分：入门篇（0基础→跑通第一个Agent）

核心目标：不依赖任何框架，手写原生Agent，彻底搞懂Agent底层运行逻辑

核心知识点：Agent的四大核心组件（缺一不可）

  1. 大脑：大语言模型（LLM）：负责思考、决策、生成内容，是Agent的核心
  2. 手脚：工具调用（Tool Use/Function Call）：让LLM能调用外部能力（查天气、搜网页、读文件、跑代码等），突破LLM的知识边界和能力限制
  3. 思考逻辑：规划（Planning）：把大目标拆解成小步骤，按顺序执行，遇到问题调整方案
  4. 记事本：记忆（Memory）：记住对话历史、用户偏好、执行过的操作，保证交互连贯

核心原理：Agent的一次完整工作流

用户下达目标 → LLM思考拆解任务 → 决定调用工具 → 本地执行工具获取结果 → 结果返回给LLM → LLM基于结果继续思考/输出最终答案

这是所有Agent的底层逻辑，再复杂的Agent也只是这个流程的循环和扩展。

入门案例1：手写极简天气查询Agent（零框架，原生Python）

需求说明

用户问天气相关问题，Agent能自主调用天气API，获取实时天气，生成自然语言回答；非天气问题，直接正常回答。

完整可运行代码（逐行注释，复制就能跑）
  1. 先创建.env文件，填入你的API Key：
DASHSCOPE_API_KEY=你的通义千问API Key
WEATHER_API_KEY=你的和风天气API Key（免费注册：https://dev.qweather.com/）
  1. 主代码weather_agent.py：
import os
import requests
from dotenv import load_dotenv
from http import HTTPStatus
import dashscope

# 加载环境变量
load_dotenv()
dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")
WEATHER_API_KEY = os.getenv("WEATHER_API_KEY")

# ---------------------- 步骤1：定义工具（函数） ----------------------
def get_weather(city: str) -> str:
    """
    获取指定城市的实时天气
    :param city: 城市名称，比如"成都"
    :return: 天气信息字符串
    """
    # 1. 先获取城市的Location ID
    geo_url = f"https://geoapi.qweather.com/v2/city/lookup?location={city}&key={WEATHER_API_KEY}"
    geo_response = requests.get(geo_url).json()
    if geo_response["code"] != "200":
        return f"获取城市{city}信息失败"
    
    location_id = geo_response["location"][0]["id"]
    # 2. 获取实时天气
    weather_url = f"https://devapi.qweather.com/v7/weather/now?location={location_id}&key={WEATHER_API_KEY}"
    weather_response = requests.get(weather_url).json()
    if weather_response["code"] != "200":
        return f"获取{city}天气失败"
    
    now = weather_response["now"]
    return f"{city}当前天气：{now['text']}，温度{now['temp']}℃，湿度{now['humidity']}%，风向{now['windDir']}{now['windScale']}级"

# ---------------------- 步骤2：定义工具的元信息（给LLM看的工具说明） ----------------------
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的实时天气信息，只能用于天气查询相关问题",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，比如成都、北京、上海",
                    }
                },
                "required": ["city"],
            },
        },
    }
]

# ---------------------- 步骤3：定义Agent核心执行逻辑 ----------------------
def agent_run(user_query: str):
    # 第一轮：调用LLM，判断是否需要调用工具
    messages = [
        {"role": "system", "content": "你是一个智能助手，能调用工具获取信息，回答用户问题。如果需要调用工具，严格按照工具格式返回；不需要调用工具，直接回答用户问题。"},
        {"role": "user", "content": user_query}
    ]

    # 调用通义千问大模型，开启工具调用能力
    response = dashscope.Generation.call(
        model=dashscope.Generation.Models.qwen_turbo,
        messages=messages,
        tools=tools,
        result_format="message",
    )

    if response.status_code != HTTPStatus.OK:
        return f"模型调用失败：{response.message}"
    
    output_message = response.output.choices[0].message

    # 判断：LLM是否要求调用工具
    if "tool_calls" in output_message:
        # 提取工具调用信息
        tool_call = output_message.tool_calls[0]
        function_name = tool_call["function"]["name"]
        function_args = eval(tool_call["function"]["arguments"])

        # 执行对应的工具函数
        if function_name == "get_weather":
            tool_result = get_weather(**function_args)
        
        # 把工具结果返回给LLM，生成最终回答
        messages.append(output_message)
        messages.append({
            "role": "tool",
            "content": tool_result,
            "tool_call_id": tool_call["id"]
        })

        # 第二轮调用LLM，生成最终答案
        final_response = dashscope.Generation.call(
            model=dashscope.Generation.Models.qwen_turbo,
            messages=messages,
            result_format="message",
        )
        return final_response.output.choices[0].message.content
    
    # 不需要调用工具，直接返回LLM的回答
    else:
        return output_message.content

# ---------------------- 测试运行 ----------------------
if __name__ == "__main__":
    # 测试1：天气查询问题
    print("测试1：", agent_run("成都今天天气怎么样？"))
    print("-"*50)
    # 测试2：非天气问题
    print("测试2：", agent_run("给我讲一个冷笑话"))
核心提炼

这个极简Agent，已经包含了所有Agent的核心要素：

  1. 工具定义：给LLM明确的工具说明，告诉它什么时候用、怎么用
  2. 决策逻辑：LLM自主判断是否需要调用工具
  3. 工具执行：本地执行函数，获取真实数据
  4. 结果整合：LLM基于工具返回的结果，生成最终回答

入门案例2：多工具扩展版Agent

需求说明

给Agent增加3个工具：天气查询、计算器、中英文翻译，让LLM自主选择调用对应的工具解决问题。

核心代码修改

只需要在上面的代码基础上，新增工具函数和工具元信息即可：

# 新增工具函数1：计算器
def calculate(expression: str) -> str:
    """
    执行数学计算，只支持加减乘除四则运算，比如"100*5+30"
    :param expression: 数学表达式字符串
    :return: 计算结果
    """
    try:
        # 安全限制：只允许数字和加减乘除
        allowed_chars = "0123456789+-*/(). "
        for c in expression:
            if c not in allowed_chars:
                return "不支持的计算表达式，仅支持加减乘除四则运算"
        result = eval(expression)
        return f"计算结果：{expression} = {result}"
    except Exception as e:
        return f"计算失败：{str(e)}"

# 新增工具函数2：翻译
def translate(text: str, target_language: str) -> str:
    """
    中英文翻译
    :param text: 要翻译的文本
    :param target_language: 目标语言，只能是"中文"或"英文"
    :return: 翻译结果
    """
    prompt = f"把以下文本翻译成{target_language}，只返回翻译结果：{text}"
    response = dashscope.Generation.call(
        model=dashscope.Generation.Models.qwen_turbo,
        prompt=prompt,
    )
    return f"翻译结果：{response.output.text}"

# 扩展tools元信息，新增两个工具的定义
tools = [
    # 原来的天气工具
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的实时天气信息，只能用于天气查询相关问题",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称，比如成都、北京、上海"}
                },
                "required": ["city"],
            },
        },
    },
    # 新增计算器工具
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "执行数学四则运算，只能用于数学计算相关问题",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "数学表达式，比如100*5+30"}
                },
                "required": ["expression"],
            },
        },
    },
    # 新增翻译工具
    {
        "type": "function",
        "function": {
            "name": "translate",
            "description": "中英文翻译，只能用于文本翻译相关问题",
            "parameters": {
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "要翻译的文本"},
                    "target_language": {"type": "string", "description": "目标语言，只能是中文或英文"}
                },
                "required": ["text", "target_language"],
            },
        },
    }
]

# 同时修改agent_run函数里的工具执行逻辑，支持多个工具
if function_name == "get_weather":
    tool_result = get_weather(**function_args)
elif function_name == "calculate":
    tool_result = calculate(**function_args)
elif function_name == "translate":
    tool_result = translate(**function_args)
测试运行
if __name__ == "__main__":
    print(agent_run("帮我算一下1234*5678等于多少"))
    print("-"*50)
    print(agent_run("把'我要成为Agent开发高手'翻译成英文"))
    print("-"*50)
    print(agent_run("北京今天天气怎么样？适合出门吗？"))
入门篇通关考核

自己动手实现一个「快递查询Agent」：

  1. 对接免费的快递查询API
  2. 实现用户输入快递单号，Agent自主查询物流信息，返回自然语言结果
  3. 扩展功能：支持快递公司名称识别

第二部分：进阶篇（实用Agent开发→解决真实问题）

核心目标：掌握Agent的工业级标准范式，开发能解决真实业务问题的Agent

核心知识点1：ReAct框架——Agent的标准思考-行动范式

ReAct是目前工业界最主流的Agent范式，大白话就是**「思考→行动→观察→循环」**，直到完成目标。

  • Thought（思考）：当前要解决什么问题？下一步该做什么？为什么要这么做？
  • Action（行动）：调用哪个工具，传入什么参数？
  • Observation（观察）：工具执行的结果是什么？拿到了什么信息？
  • 重复以上步骤，直到判断任务完成，输出Final Answer（最终答案）

ReAct的核心价值：让LLM的思考过程可解释、可管控，大幅减少幻觉，提升任务完成率。

进阶案例1：ReAct实现全网信息检索Agent

需求说明

用户提出一个需要实时信息/全网资料的问题，Agent能自主搜索网页、提取信息、整理成结构化报告，解决LLM知识过时、信息不全的问题。

完整代码实现
  1. 先在.env新增Serper搜索API Key（免费注册：https://serper.dev/，国内可用）
SERPER_API_KEY=你的Serper API Key
  1. 主代码react_search_agent.py：
import os
import requests
from dotenv import load_dotenv
import dashscope

load_dotenv()
dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")
SERPER_API_KEY = os.getenv("SERPER_API_KEY")

# ---------------------- 1. 定义工具：全网搜索 ----------------------
def web_search(query: str, num_results: int = 5) -> str:
    """
    全网搜索，获取相关网页信息
    :param query: 搜索关键词
    :param num_results: 返回结果数量，默认5条
    :return: 搜索结果字符串
    """
    url = "https://google.serper.dev/search"
    payload = {"q": query, "num": num_results}
    headers = {
        "X-API-KEY": SERPER_API_KEY,
        "Content-Type": "application/json"
    }
    response = requests.post(url, json=payload, headers=headers)
    if response.status_code != 200:
        return "搜索失败，请重试"
    
    results = response.json().get("organic", [])
    if not results:
        return "没有找到相关信息"
    
    # 格式化搜索结果
    formatted_results = []
    for idx, res in enumerate(results, 1):
        formatted_results.append(
            f"【结果{idx}】标题：{res['title']}\n摘要：{res['snippet']}\n链接：{res['link']}\n"
        )
    return "\n".join(formatted_results)

# ---------------------- 2. ReAct核心Prompt模板 ----------------------
REACT_PROMPT = """
你是一个具备自主思考和全网搜索能力的ReAct智能助手，严格按照以下格式执行任务，绝对不能偏离格式。

你的执行流程：
1. Thought：先思考用户的需求是什么，要完成这个需求，我需要做什么，下一步应该调用什么工具
2. Action：只输出要调用的工具名称和参数，格式：Action: 工具名(参数名=参数值)
3. Observation：工具执行后返回的结果，我会给你填充
4. 重复1-3步，直到你获取了足够的信息，完成用户的需求
5. 最终输出：Final Answer: 整理后的完整答案

注意事项：
- 只能使用你拥有的工具，禁止编造不存在的工具
- 所有事实性内容必须通过搜索验证，绝对不能编造信息
- 每次只执行一个Action，不要一次性执行多个操作
- 必须严格按照格式输出，不能添加额外内容

你拥有的工具：
- web_search(query: str, num_results: int = 5)：全网搜索，获取网页信息，query是搜索关键词
"""

# ---------------------- 3. ReAct Agent核心执行逻辑 ----------------------
def react_agent_run(user_query: str, max_steps: int = 5):
    # 初始化对话上下文
    messages = [
        {"role": "system", "content": REACT_PROMPT},
        {"role": "user", "content": f"用户需求：{user_query}"}
    ]

    # 限制最大循环次数，防止无限循环
    for step in range(max_steps):
        print(f"\n===== 执行步骤 {step+1}/{max_steps} =====")
        # 调用LLM，生成思考和行动
        response = dashscope.Generation.call(
            model=dashscope.Generation.Models.qwen_plus,
            messages=messages,
            temperature=0.1,  # 降低随机性，提升稳定性
        )
        output = response.output.text
        print(output)

        # 判断是否已经输出最终答案
        if "Final Answer:" in output:
            return output.split("Final Answer:")[-1].strip()
        
        # 解析Action，提取工具名和参数
        if "Action:" in output:
            action_str = output.split("Action:")[-1].strip()
            try:
                # 解析工具名和参数，比如 web_search(query="2026年春节档票房")
                func_name = action_str.split("(")[0]
                args_str = action_str.split("(")[1].rstrip(")")
                # 解析参数字典
                args_dict = {}
                for arg in args_str.split(","):
                    k, v = arg.strip().split("=")
                    args_dict[k.strip()] = v.strip().strip('"').strip("'")
                
                # 执行工具
                if func_name == "web_search":
                    observation = web_search(**args_dict)
                else:
                    observation = f"错误：不存在工具{func_name}"
                
                print(f"\nObservation: {observation}")
                # 把思考、行动、观察结果加入上下文
                messages.append({"role": "assistant", "content": output})
                messages.append({"role": "user", "content": f"Observation: {observation}"})
            
            except Exception as e:
                observation = f"工具执行失败：{str(e)}，请检查Action格式是否正确"
                print(f"\nObservation: {observation}")
                messages.append({"role": "assistant", "content": output})
                messages.append({"role": "user", "content": f"Observation: {observation}"})
    
    # 超过最大步数，返回结果
    return "执行步数超过上限，未能完成任务，请简化需求重试"

# ---------------------- 测试运行 ----------------------
if __name__ == "__main__":
    result = react_agent_run("2026年春节档票房排名，以及票房最高的3部电影的核心信息")
    print("\n===== 最终答案 =====")
    print(result)
核心优化技巧
  1. 降低temperature到0.1-0.3，大幅提升LLM输出格式的稳定性
  2. 限制最大循环步数，防止Agent无限循环
  3. 严格的格式约束，让LLM必须按照ReAct模板输出，避免格式混乱
  4. 异常捕获，工具执行失败时，告诉LLM错误原因，让它重新规划

核心知识点2：记忆模块——让Agent记住你，实现连贯交互

Agent的记忆分为2类，缺一不可：

  1. 短期记忆：当前对话的上下文，保存在内存里，对话结束就消失
  2. 长期记忆：用户的偏好、历史行为、核心信息，保存在向量数据库里，永久留存，每次对话自动检索相关记忆

进阶案例2：带长期记忆的个人专属助理Agent

需求说明

Agent能记住你的饮食偏好、日程安排、工作习惯、个人信息，不管多少次对话，都能基于你的专属信息提供个性化回答。

完整代码实现
import os
import dashscope
import chromadb
from dotenv import load_dotenv
from chromadb.utils import embedding_functions

load_dotenv()
dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")

# ---------------------- 1. 初始化向量数据库（长期记忆存储） ----------------------
# 用通义千问的embedding模型
embedding_func = embedding_functions.DashScopeEmbeddingFunction(
    api_key=dashscope.api_key,
    model_name="text-embedding-v2"
)

# 初始化chroma客户端，创建记忆集合
chroma_client = chromadb.PersistentClient(path="./memory_db")
memory_collection = chroma_client.get_or_create_collection(
    name="user_long_term_memory",
    embedding_function=embedding_func,
    metadata={"description": "用户的长期记忆"}
)

# ---------------------- 2. 记忆管理函数 ----------------------
def add_memory(memory_content: str, user_id: str = "default_user"):
    """添加长期记忆"""
    memory_collection.add(
        documents=[memory_content],
        metadatas=[{"user_id": user_id}],
        ids=[f"memory_{user_id}_{int(os.times()[4])}"]
    )
    print(f"已添加记忆：{memory_content}")

def retrieve_memory(query: str, user_id: str = "default_user", top_k: int = 3) -> str:
    """检索相关的长期记忆"""
    results = memory_collection.query(
        query_texts=[query],
        where={"user_id": user_id},
        n_results=top_k
    )
    if not results["documents"][0]:
        return "暂无相关记忆"
    return "\n".join([f"记忆{i+1}：{doc}" for i, doc in enumerate(results["documents"][0])])

# ---------------------- 3. 带记忆的Agent核心逻辑 ----------------------
MEMORY_AGENT_PROMPT = """
你是用户的专属个人助理，你需要先参考用户的长期记忆，再回答用户的问题，回答要贴合用户的个人情况。
如果用户提供了新的个人信息，你需要调用add_memory工具，把信息存入长期记忆。

你拥有的工具：
1. add_memory(memory_content: str)：把用户的新信息添加到长期记忆
2. retrieve_memory(query: str)：检索用户的相关长期记忆

执行规则：
- 回答用户问题前，必须先检索相关的长期记忆
- 用户提到的个人偏好、习惯、信息，必须存入长期记忆
- 禁止编造记忆，所有记忆都必须来自用户的对话
"""

def memory_agent_run(user_query: str, user_id: str = "default_user"):
    # 第一步：检索相关记忆
    related_memory = retrieve_memory(user_query, user_id)
    print(f"检索到的相关记忆：\n{related_memory}\n")

    # 初始化上下文
    messages = [
        {"role": "system", "content": MEMORY_AGENT_PROMPT},
        {"role": "user", "content": f"用户的相关长期记忆：{related_memory}\n用户当前问题：{user_query}"}
    ]

    # 调用LLM
    response = dashscope.Generation.call(
        model=dashscope.Generation.Models.qwen_turbo,
        messages=messages,
        tools=[
            {
                "type": "function",
                "function": {
                    "name": "add_memory",
                    "description": "把用户的个人信息添加到长期记忆",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "memory_content": {"type": "string", "description": "要存储的记忆内容"}
                        },
                        "required": ["memory_content"],
                    },
                },
            }
        ],
        result_format="message",
    )

    output_message = response.output.choices[0].message

    # 判断是否要添加记忆
    if "tool_calls" in output_message:
        tool_call = output_message.tool_calls[0]
        function_name = tool_call["function"]["name"]
        function_args = eval(tool_call["function"]["arguments"])
        if function_name == "add_memory":
            add_memory(**function_args, user_id=user_id)
        # 生成最终回答
        messages.append(output_message)
        messages.append({"role": "tool", "content": "记忆添加成功", "tool_call_id": tool_call["id"]})
        final_response = dashscope.Generation.call(
            model=dashscope.Generation.Models.qwen_turbo,
            messages=messages,
        )
        return final_response.output.text
    else:
        return output_message.content

# ---------------------- 测试运行 ----------------------
if __name__ == "__main__":
    # 第一次对话：告诉Agent你的信息
    print("第一次对话：", memory_agent_run("我叫张三，对芒果过敏，喜欢喝无糖冰美式，每周三下午要开周会"))
    print("-"*50)
    # 第二次对话：测试记忆
    print("第二次对话：", memory_agent_run("下午给我推荐一杯喝的"))
    print("-"*50)
    # 第三次对话：测试日程记忆
    print("第三次对话：", memory_agent_run("我周三下午3点有个会，帮我记一下"))
    print("-"*50)
    print("第四次对话：", memory_agent_run("周三下午我有什么安排？"))

进阶篇核心扩展：RAG+Agent

RAG（检索增强生成）+ Agent，是企业级Agent最常用的组合，核心是让Agent能对接你的私有数据（PDF、Word、Excel、企业文档、代码库等），解决私有数据问答、内部知识库查询等核心业务需求。

因篇幅限制，这里给出核心实现思路，完整案例可参考：

  1. 用LlamaIndex/LangChain加载私有文档，做分块、向量化，存入向量数据库
  2. 给Agent定义retrieve_knowledge工具，让LLM自主决定什么时候检索知识库
  3. Agent拿到检索结果后，整合信息，生成精准回答，彻底解决幻觉和私有数据访问问题

进阶篇通关考核

实现一个「个人理财助手Agent」，具备以下能力：

  1. 实时汇率查询工具
  2. 理财收益计算工具
  3. 账单整理与分析能力
  4. 长期记忆：记住你的风险偏好、理财习惯

第三部分：高手篇（生产级Agent→从demo到商用）

核心目标：掌握多智能体协作、反思迭代、工程化部署，开发能商用的Agent系统

核心知识点1：Multi-Agent多智能体协作

单Agent的能力有上限，复杂任务需要多个Agent分工协作，就像一个公司，有不同的岗位，各司其职，共同完成目标。

主流的Multi-Agent框架：

  • CrewAI：上手最简单，专门为多智能体协作设计，角色定义清晰，适合快速开发
  • LangGraph：基于LangChain，支持复杂的状态流转、分支逻辑，适合生产级复杂系统
  • AutoGen：微软开源，支持多Agent对话、代码执行，适合科研和复杂场景

高手案例1：Multi-Agent自动化内容生产工作室（CrewAI实现）

需求说明

搭建一个5人内容生产团队，从选题到成稿全自动化，角色分工：

  1. 策划Agent：负责选题策划、内容大纲设计
  2. 写作Agent：根据大纲，完成正文写作
  3. 校对Agent：负责错别字、语病、逻辑校验
  4. SEO优化Agent：负责关键词优化、标题优化，提升传播性
  5. 排版Agent：负责markdown格式排版，适配公众号/小红书等平台
完整代码实现
import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from langchain_community.llms import Tongyi

load_dotenv()
os.environ["DASHSCOPE_API_KEY"] = os.getenv("DASHSCOPE_API_KEY")

# 初始化LLM
llm = Tongyi(model_name="qwen-plus", temperature=0.7)

# ---------------------- 1. 定义Agent角色 ----------------------
# 策划Agent
planner = Agent(
    role="内容策划专家",
    goal="根据用户的需求，策划出有传播性、逻辑清晰的内容选题和详细大纲",
    backstory="你有10年新媒体内容策划经验，擅长爆款内容的选题和结构设计，能精准把握用户痛点",
    llm=llm,
    verbose=True,
    allow_delegation=False
)

# 写作Agent
writer = Agent(
    role="资深内容作家",
    goal="根据策划大纲，写出内容详实、语言流畅、有感染力的原创文章",
    backstory="你是资深专栏作家，擅长多种文体写作，文字功底扎实，能把复杂的内容讲得通俗易懂",
    llm=llm,
    verbose=True,
    allow_delegation=False
)

# 校对Agent
proofreader = Agent(
    role="专业校对编辑",
    goal="校对文章的错别字、语病、逻辑漏洞，保证内容的准确性和可读性",
    backstory="你有8年出版社校对经验，对文字错误零容忍，能精准发现文章中的所有问题",
    llm=llm,
    verbose=True,
    allow_delegation=False
)

# SEO优化Agent
seo_optimizer = Agent(
    role="SEO优化专家",
    goal="优化文章的标题、关键词、结构，提升搜索引擎排名和传播性",
    backstory="你有10年SEO优化经验，熟悉各大平台的流量规则，能让文章获得更多曝光",
    llm=llm,
    verbose=True,
    allow_delegation=False
)

# 排版Agent
typesetter = Agent(
    role="排版设计师",
    goal="把文章排版成美观、易读的markdown格式，适配公众号、小红书等平台",
    backstory="你是资深新媒体排版设计师，擅长用markdown做出美观的排版，提升用户阅读体验",
    llm=llm,
    verbose=True,
    allow_delegation=False
)

# ---------------------- 2. 定义任务 ----------------------
# 任务1：内容策划
task_plan = Task(
    description="用户需求：写一篇关于「零基础学习Agent开发」的爆款文章，目标受众是编程初学者和AI爱好者。请完成选题策划和详细的文章大纲，大纲要包含每个部分的核心内容。",
    agent=planner,
    expected_output="详细的文章选题和大纲，包含每个章节的核心内容和写作要点"
)

# 任务2：正文写作
task_write = Task(
    description="根据策划的文章大纲，完成全文写作，字数不少于2000字，内容详实，通俗易懂，适合初学者阅读",
    agent=writer,
    expected_output="完整的文章正文，逻辑清晰，语言流畅，符合大纲要求",
    context=[task_plan]
)

# 任务3：校对
task_proofread = Task(
    description="校对文章，修正所有错别字、语病、逻辑漏洞，优化语句的流畅度",
    agent=proofreader,
    expected_output="校对后的完整文章，标注出所有修改的地方",
    context=[task_write]
)

# 任务4：SEO优化
task_seo = Task(
    description="优化文章的标题、小标题、关键词，提升文章的搜索排名和传播性，适配新媒体平台",
    agent=seo_optimizer,
    expected_output="SEO优化后的完整文章，包含优化后的标题、关键词布局",
    context=[task_proofread]
)

# 任务5：排版
task_typeset = Task(
    description="把文章排版成美观的markdown格式，适配公众号和小红书，添加合适的标题层级、重点标注、分段",
    agent=typesetter,
    expected_output="排版完成的markdown格式文章，美观易读，适配新媒体平台",
    context=[task_seo]
)

# ---------------------- 3. 组建Crew，启动执行 ----------------------
content_crew = Crew(
    agents=[planner, writer, proofreader, seo_optimizer, typesetter],
    tasks=[task_plan, task_write, task_proofread, task_seo, task_typeset],
    process=Process.sequential,  # 顺序执行，前一个任务完成，后一个才开始
    verbose=True
)

# 运行
if __name__ == "__main__":
    result = content_crew.kickoff()
    print("\n===== 最终成品 =====")
    print(result)

核心知识点2：Agent的反思与自我迭代

让Agent具备反思能力，是从“能用”到“好用”的关键。核心逻辑是：Agent完成任务后，自己复盘执行过程，找出问题，优化方案，下次执行时做得更好。

高手案例2：带反思能力的代码开发Agent

核心实现思路：

  1. 需求理解：Agent拆解用户的代码需求
  2. 代码编写：生成初始代码
  3. 代码执行：运行代码，获取报错信息
  4. 反思复盘：分析报错原因，找出代码的问题
  5. 代码优化：修改代码，重新运行
  6. 循环以上步骤，直到代码正常运行，同时生成优化说明

核心知识点3：Agent的工程化与部署上线

一个生产级Agent，必须解决以下工程化问题：

  1. 安全可控：权限控制，禁止Agent执行危险操作，比如删除文件、执行系统命令
  2. 日志监控：全流程日志记录，可追溯、可排查问题
  3. 成本优化：控制Token消耗，比如上下文精简、模型分级调用
  4. 高可用：异常重试、降级处理，保证服务稳定
  5. 部署上线：封装成API，开发前端界面，让用户能直接使用
高手案例3：Agent部署上线（Streamlit+FastAPI）
  1. 用FastAPI把Agent封装成HTTP接口，供前端调用
  2. 用Streamlit开发可视化前端界面，用户能直接在网页上使用Agent
  3. 部署到服务器，实现公网访问

第四部分：避坑指南与进阶提升

初学者最容易踩的10个坑

  1. 一上来就用复杂框架，不理解底层逻辑，遇到问题就懵
  2. Prompt写得太模糊，工具描述不清晰，导致LLM乱调用工具、格式混乱
  3. 不限制循环次数，导致Agent无限循环，Token疯狂消耗
  4. 不做参数校验和异常处理，LLM返回的参数不对，直接导致程序崩溃
  5. 不做事实校验，完全相信LLM的输出，导致严重的幻觉问题
  6. 上下文不做精简，对话越长，Token消耗越大，最终溢出报错
  7. 用高参数模型做简单任务，导致成本极高，性价比极低
  8. 多Agent角色定义模糊，分工重叠，导致任务执行混乱
  9. 不做权限控制，让Agent能执行任意代码/系统命令，带来严重的安全风险
  10. 只做demo，不做工程化，无法落地到真实业务场景

Agent幻觉的终极解决方案

  1. 工具优先：所有事实性内容，必须通过工具检索/校验，绝对不能让LLM凭空输出
  2. 溯源机制：所有输出的信息，必须标注来源（比如搜索结果链接、知识库文档）
  3. 反思校验：让Agent自己检查输出的内容，找出可能的错误，重新验证
  4. 降低随机性：事实性任务，把temperature调到0.1以下，减少幻觉
  5. 多轮交叉验证：关键信息，用多个工具/多次搜索交叉验证

主流Agent框架选型指南

框架 优势 劣势 适用场景
LangChain 生态最全、功能最多、社区最活跃，支持几乎所有LLM和工具 上手有难度，版本迭代快，容易有兼容性问题 复杂单Agent、企业级RAG+Agent系统
CrewAI 专为Multi-Agent设计，上手极快，角色定义清晰，代码简洁 复杂流程定制能力弱于LangGraph 多智能体协作场景、自动化工作流
LangGraph 基于LangChain，支持复杂的状态流转、分支、循环逻辑 有一定的学习门槛，需要理解状态机 生产级复杂Agent系统、多轮循环任务
AutoGen 微软开源，支持多Agent对话、代码执行、人机交互 定制化难度高，适合科研，不适合商用 科研、复杂代码开发、多Agent对话场景

最终通关指南

想要从初学者快速成为Agent高手，核心只有3点：

  1. 吃透底层：先手写原生Agent，理解每一步的运行逻辑，不要一上来就依赖框架
  2. 多练多改：每个案例都自己敲一遍，然后修改扩展，比如给Agent加新的工具、新的能力，举一反三
  3. 落地实战：找一个真实的需求（比如个人助理、自动化办公、企业知识库），从头到尾做一个完整的Agent项目，解决真实问题

开箱即用Agent全栈代码包（完整版）

以下是全量可运行、零修改开箱、带完整部署文档+排障手册的代码包，所有代码已修复安全隐患、补充健壮性处理、注释全覆盖，复制到本地即可直接运行。

一、代码包整体目录结构

Agent_Full_Stack_Code/
├── 00_快速启动指南.md
├── 01_环境配置全流程.md
├── 02_常见问题排障手册.md
├── 03_生产级部署上线指南.md
├── requirements.txt          # 统一依赖文件
├── .env.example              # 环境变量模板（复制改名为.env即可）
├── 入门篇_零基础Agent/
│   ├── 案例1_极简天气查询Agent.py
│   ├── 案例2_多工具全能Agent.py
│   └── 入门通关考核_快递查询Agent.py
├── 进阶篇_实用商用Agent/
│   ├── 案例1_ReAct全网检索Agent.py
│   ├── 案例2_带长期记忆的专属助理Agent.py
│   └── 进阶通关考核_个人理财助手Agent.py
└── 高手篇_生产级多智能体/
    ├── 案例1_多智能体内容生产工作室.py
    ├── 案例2_带反思能力的代码开发Agent.py
    └── 部署套件/
        ├── agent_api.py      # FastAPI接口封装
        └── agent_webui.py    # Streamlit可视化前端

二、核心基础文件（先配置这2个）

1. requirements.txt（全量依赖，版本固定无兼容问题）

python-dotenv==1.0.1
requests==2.32.3
dashscope==1.20.1
chromadb==0.5.11
langchain==0.2.16
langchain-community==0.2.16
crewai==0.51.1
streamlit==1.38.0
fastapi==0.115.0
uvicorn==0.30.6
pydantic==2.9.2

2. .env.example（复制改名为.env，填入你的API Key即可）

# 核心LLM：通义千问API Key（免费注册：https://www.aliyun.com/product/ailabs/tongyi）
DASHSCOPE_API_KEY=你的通义千问API_KEY

# 天气查询API（免费注册：https://dev.qweather.com/）
WEATHER_API_KEY=你的和风天气API_KEY

# 全网搜索API（免费注册：https://serper.dev/，国内可用）
SERPER_API_KEY=你的Serper API_KEY

# 快递查询API（免费注册：https://www.kuaidi100.com/openapi/）
KUAIDI100_API_KEY=你的快递100API_KEY
KUAIDI100_CUSTOMER=你的快递100商户编号

三、入门篇_零基础Agent 完整代码

案例1_极简天气查询Agent.py

import os
import json
import requests
from dotenv import load_dotenv
from http import HTTPStatus
import dashscope

# 加载环境变量
load_dotenv()
dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")
WEATHER_API_KEY = os.getenv("WEATHER_API_KEY")

# ====================== 1. 定义工具函数（Agent的手脚） ======================
def get_weather(city: str) -> str:
    """
    获取指定城市的实时天气
    :param city: 城市名称，比如"成都"
    :return: 格式化后的天气信息字符串
    """
    try:
        # 第一步：获取城市Location ID
        geo_url = f"https://geoapi.qweather.com/v2/city/lookup?location={city}&key={WEATHER_API_KEY}"
        geo_response = requests.get(geo_url, timeout=10).json()
        
        if geo_response["code"] != "200":
            return f"错误：未找到城市【{city}】，请检查城市名称是否正确"
        
        location_id = geo_response["location"][0]["id"]
        city_name = geo_response["location"][0]["name"]

        # 第二步：获取实时天气
        weather_url = f"https://devapi.qweather.com/v7/weather/now?location={location_id}&key={WEATHER_API_KEY}"
        weather_response = requests.get(weather_url, timeout=10).json()
        
        if weather_response["code"] != "200":
            return f"错误：获取【{city_name}】天气失败，请稍后重试"
        
        now_data = weather_response["now"]
        return (
            f"【{city_name}实时天气】\n"
            f"天气状况：{now_data['text']}\n"
            f"当前温度：{now_data['temp']}℃\n"
            f"相对湿度：{now_data['humidity']}%\n"
            f"风向风力：{now_data['windDir']} {now_data['windScale']}级"
        )
    except Exception as e:
        return f"天气查询异常：{str(e)}"

# ====================== 2. 工具元信息（给LLM看的工具说明书） ======================
TOOLS_CONFIG = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "仅用于查询指定城市的实时天气信息，非天气类问题禁止调用",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "中文城市名称，例如：成都、北京、上海、广州"
                    }
                },
                "required": ["city"]
            }
        }
    }
]

# ====================== 3. Agent核心执行逻辑 ======================
def weather_agent_run(user_query: str) -> str:
    # 初始化对话上下文
    messages = [
        {
            "role": "system",
            "content": "你是一个智能天气助手，仅能回答天气相关问题，非天气问题请礼貌告知用户你仅支持天气查询。需要查询天气时，严格按照要求调用工具，禁止编造天气数据。"
        },
        {"role": "user", "content": user_query}
    ]

    # 第一轮调用LLM：判断是否需要调用工具
    try:
        response = dashscope.Generation.call(
            model=dashscope.Generation.Models.qwen_turbo,
            messages=messages,
            tools=TOOLS_CONFIG,
            result_format="message",
            timeout=10
        )
    except Exception as e:
        return f"模型调用失败：{str(e)}"

    if response.status_code != HTTPStatus.OK:
        return f"请求异常：{response.message}"

    output_msg = response.output.choices[0].message

    # 判断：LLM是否要求调用工具
    if "tool_calls" in output_msg:
        tool_call = output_msg.tool_calls[0]
        func_name = tool_call["function"]["name"]
        # 安全解析参数（替换eval，杜绝代码注入风险）
        try:
            func_args = json.loads(tool_call["function"]["arguments"])
        except json.JSONDecodeError:
            return "参数解析失败，请重新提问"

        # 执行工具函数
        if func_name == "get_weather":
            tool_result = get_weather(**func_args)
        else:
            tool_result = f"错误：不支持的工具【{func_name}】"

        # 把工具结果返回给LLM，生成最终回答
        messages.append(output_msg)
        messages.append({
            "role": "tool",
            "content": tool_result,
            "tool_call_id": tool_call["id"]
        })

        # 第二轮调用LLM：生成最终自然语言回答
        try:
            final_response = dashscope.Generation.call(
                model=dashscope.Generation.Models.qwen_turbo,
                messages=messages,
                result_format="message",
                timeout=10
            )
        except Exception as e:
            return f"结果生成失败：{str(e)}"
        
        return final_response.output.choices[0].message.content

    # 不需要调用工具，直接返回回答
    else:
        return output_msg.content

# ====================== 测试运行 ======================
if __name__ == "__main__":
    print("="*30 + " 测试1：天气查询 " + "="*30)
    print(weather_agent_run("成都今天天气怎么样？适合出门吗？"))
    print("\n" + "="*30 + " 测试2：非天气问题 " + "="*30)
    print(weather_agent_run("给我讲一个冷笑话"))

案例2_多工具全能Agent.py

import os
import json
import requests
from dotenv import load_dotenv
from http import HTTPStatus
import dashscope

# 加载环境变量
load_dotenv()
dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")
WEATHER_API_KEY = os.getenv("WEATHER_API_KEY")

# ====================== 1. 工具函数库 ======================
def get_weather(city: str) -> str:
    """获取指定城市的实时天气"""
    try:
        geo_url = f"https://geoapi.qweather.com/v2/city/lookup?location={city}&key={WEATHER_API_KEY}"
        geo_response = requests.get(geo_url, timeout=10).json()
        if geo_response["code"] != "200":
            return f"错误：未找到城市【{city}】"
        location_id = geo_response["location"][0]["id"]
        city_name = geo_response["location"][0]["name"]

        weather_url = f"https://devapi.qweather.com/v7/weather/now?location={location_id}&key={WEATHER_API_KEY}"
        weather_response = requests.get(weather_url, timeout=10).json()
        if weather_response["code"] != "200":
            return f"错误：获取【{city_name}】天气失败"
        
        now_data = weather_response["now"]
        return f"{city_name}当前天气：{now_data['text']}，温度{now_data['temp']}℃，湿度{now_data['humidity']}%，{now_data['windDir']}{now_data['windScale']}级"
    except Exception as e:
        return f"天气查询异常：{str(e)}"

def calculate(expression: str) -> str:
    """执行数学四则运算，仅支持加减乘除和括号"""
    try:
        # 安全校验：仅允许数字和合法运算符
        allowed_chars = "0123456789+-*/(). "
        for char in expression:
            if char not in allowed_chars:
                return "错误：仅支持加减乘除四则运算，禁止其他字符"
        # 执行计算
        result = eval(expression, {"__builtins__": None}, {})
        return f"计算结果：{expression} = {result}"
    except ZeroDivisionError:
        return "错误：除数不能为0"
    except Exception as e:
        return f"计算失败：{str(e)}"

def translate(text: str, target_language: str) -> str:
    """中英文互译"""
    try:
        prompt = f"将以下文本精准翻译成{target_language}，仅返回翻译结果，不要额外内容：{text}"
        response = dashscope.Generation.call(
            model=dashscope.Generation.Models.qwen_turbo,
            prompt=prompt,
            timeout=10
        )
        return f"翻译结果：{response.output.text}"
    except Exception as e:
        return f"翻译失败：{str(e)}"

# ====================== 2. 工具配置 ======================
TOOLS_CONFIG = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的实时天气，仅用于天气相关问题",
            "parameters": {
                "type": "object",
                "properties": {"city": {"type": "string", "description": "中文城市名称"}},
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "执行数学四则运算，仅用于数学计算相关问题",
            "parameters": {
                "type": "object",
                "properties": {"expression": {"type": "string", "description": "数学表达式，例如：100*5+30"}},
                "required": ["expression"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "translate",
            "description": "中英文互译，仅用于文本翻译相关问题",
            "parameters": {
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "需要翻译的文本"},
                    "target_language": {"type": "string", "description": "目标语言，仅支持「中文」或「英文」"}
                },
                "required": ["text", "target_language"]
            }
        }
    }
]

# ====================== 3. Agent核心逻辑 ======================
def multi_tool_agent_run(user_query: str) -> str:
    messages = [
        {
            "role": "system",
            "content": "你是一个全能智能助手，拥有天气查询、数学计算、中英文翻译三个工具。根据用户问题，自主选择最合适的工具解决问题，禁止编造信息。不需要工具时，直接正常回答用户问题。"
        },
        {"role": "user", "content": user_query}
    ]

    # 调用LLM
    try:
        response = dashscope.Generation.call(
            model=dashscope.Generation.Models.qwen_turbo,
            messages=messages,
            tools=TOOLS_CONFIG,
            result_format="message",
            timeout=10
        )
    except Exception as e:
        return f"模型调用失败：{str(e)}"

    if response.status_code != HTTPStatus.OK:
        return f"请求异常：{response.message}"

    output_msg = response.output.choices[0].message

    # 处理工具调用
    if "tool_calls" in output_msg:
        tool_call = output_msg.tool_calls[0]
        func_name = tool_call["function"]["name"]
        try:
            func_args = json.loads(tool_call["function"]["arguments"])
        except json.JSONDecodeError:
            return "参数解析失败，请重新提问"

        # 匹配工具执行
        tool_map = {
            "get_weather": get_weather,
            "calculate": calculate,
            "translate": translate
        }
        if func_name in tool_map:
            tool_result = tool_map[func_name](**func_args)
        else:
            tool_result = f"错误：不支持的工具【{func_name}】"

        # 生成最终回答
        messages.append(output_msg)
        messages.append({
            "role": "tool",
            "content": tool_result,
            "tool_call_id": tool_call["id"]
        })

        try:
            final_response = dashscope.Generation.call(
                model=dashscope.Generation.Models.qwen_turbo,
                messages=messages,
                result_format="message",
                timeout=10
            )
        except Exception as e:
            return f"结果生成失败：{str(e)}"
        
        return final_response.output.choices[0].message.content

    # 无工具调用，直接返回
    else:
        return output_msg.content

# ====================== 测试运行 ======================
if __name__ == "__main__":
    print("="*30 + " 测试1：计算 " + "="*30)
    print(multi_tool_agent_run("帮我算一下1234*5678+9876等于多少"))
    print("\n" + "="*30 + " 测试2：翻译 " + "="*30)
    print(multi_tool_agent_run("把'零基础入门Agent开发，快速成为高手'翻译成英文"))
    print("\n" + "="*30 + " 测试3：天气 " + "="*30)
    print(multi_tool_agent_run("上海今天天气怎么样？"))

入门通关考核_快递查询Agent.py

import os
import json
import requests
from dotenv import load_dotenv
from http import HTTPStatus
import dashscope

# 加载环境变量
load_dotenv()
dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")
KUAIDI100_API_KEY = os.getenv("KUAIDI100_API_KEY")
KUAIDI100_CUSTOMER = os.getenv("KUAIDI100_CUSTOMER")

# ====================== 工具函数 ======================
def query_express(express_no: str, company_name: str = "") -> str:
    """
    查询快递物流信息
    :param express_no: 快递单号
    :param company_name: 快递公司名称（可选，自动识别可不填）
    :return: 物流信息
    """
    try:
        # 第一步：自动识别快递公司（如果没填）
        if not company_name:
            auto_url = "https://www.kuaidi100.com/autonumber/autoComNum"
            auto_response = requests.post(auto_url, data={"text": express_no}, timeout=10).json()
            if not auto_response["auto"]:
                return "错误：无法识别快递公司，请手动输入快递公司名称"
            company_code = auto_response["auto"][0]["comCode"]
            company_name = auto_response["auto"][0]["name"]
        else:
            # 快递公司名称转编码
            company_map = {
                "顺丰": "shunfeng", "京东": "jd", "中通": "zhongtong",
                "圆通": "yuantong", "申通": "shentong", "韵达": "yunda",
                "极兔": "jtexpress", "邮政": "youzhengguonei", "EMS": "ems"
            }
            company_code = company_map.get(company_name)
            if not company_code:
                return "错误：暂不支持该快递公司，请换一个快递公司查询"

        # 第二步：查询物流信息
        query_url = "https://poll.kuaidi100.com/poll/query.do"
        params = {
            "customer": KUAIDI100_CUSTOMER,
            "param": json.dumps({"com": company_code, "num": express_no})
        }
        # 生成签名（快递100要求）
        import hashlib
        sign_str = params["param"] + KUAIDI100_API_KEY + params["customer"]
        params["sign"] = hashlib.md5(sign_str.encode()).hexdigest().upper()

        query_response = requests.post(query_url, data=params, timeout=10).json()
        if not query_response["status"]:
            return f"查询失败：{query_response['message']}"
        
        # 格式化物流信息
        data = query_response["data"]
        if not data:
            return f"【{company_name} 单号{express_no}】暂无物流信息，请检查单号是否正确"
        
        result = [f"【{company_name} 快递单号：{express_no}】\n最新状态：{query_response['state']}\n物流轨迹："]
        for item in data:
            result.append(f"[{item['time']}] {item['context']}")
        return "\n".join(result)
    
    except Exception as e:
        return f"快递查询异常：{str(e)}"

# ====================== 工具配置 ======================
TOOLS_CONFIG = [
    {
        "type": "function",
        "function": {
            "name": "query_express",
            "description": "查询快递物流信息，仅用于快递查询相关问题",
            "parameters": {
                "type": "object",
                "properties": {
                    "express_no": {"type": "string", "description": "快递单号"},
                    "company_name": {"type": "string", "description": "快递公司名称，可选，不填则自动识别"}
                },
                "required": ["express_no"]
            }
        }
    }
]

# ====================== Agent核心逻辑 ======================
def express_agent_run(user_query: str) -> str:
    messages = [
        {
            "role": "system",
            "content": "你是一个快递查询智能助手，仅能回答快递物流相关问题。需要查询时，调用工具获取真实物流信息，禁止编造数据。如果用户只给了单号，直接调用工具自动识别快递公司。"
        },
        {"role": "user", "content": user_query}
    ]

    try:
        response = dashscope.Generation.call(
            model=dashscope.Generation.Models.qwen_turbo,
            messages=messages,
            tools=TOOLS_CONFIG,
            result_format="message",
            timeout=10
        )
    except Exception as e:
        return f"模型调用失败：{str(e)}"

    if response.status_code != HTTPStatus.OK:
        return f"请求异常：{response.message}"

    output_msg = response.output.choices[0].message

    if "tool_calls" in output_msg:
        tool_call = output_msg.tool_calls[0]
        func_name = tool_call["function"]["name"]
        try:
            func_args = json.loads(tool_call["function"]["arguments"])
        except json.JSONDecodeError:
            return "参数解析失败，请重新提供快递单号"

        if func_name == "query_express":
            tool_result = query_express(**func_args)
        else:
            tool_result = f"错误：不支持的工具【{func_name}】"

        messages.append(output_msg)
        messages.append({
            "role": "tool",
            "content": tool_result,
            "tool_call_id": tool_call["id"]
        })

        try:
            final_response = dashscope.Generation.call(
                model=dashscope.Generation.Models.qwen_turbo,
                messages=messages,
                result_format="message",
                timeout=10
            )
        except Exception as e:
            return f"结果生成失败：{str(e)}"
        
        return final_response.output.choices[0].message.content
    else:
        return output_msg.content

# ====================== 测试运行 ======================
if __name__ == "__main__":
    # 替换成真实的快递单号测试
    print(express_agent_run("帮我查一下这个快递：SF1234567890"))

四、进阶篇_实用商用Agent 完整代码

案例1_ReAct全网检索Agent.py

import os
import json
import requests
from dotenv import load_dotenv
import dashscope

# 加载环境变量
load_dotenv()
dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")
SERPER_API_KEY = os.getenv("SERPER_API_KEY")

# ====================== 1. 工具函数 ======================
def web_search(query: str, num_results: int = 5) -> str:
    """全网搜索，获取网页信息"""
    try:
        url = "https://google.serper.dev/search"
        payload = {"q": query, "num": num_results}
        headers = {
            "X-API-KEY": SERPER_API_KEY,
            "Content-Type": "application/json"
        }
        response = requests.post(url, json=payload, headers=headers, timeout=10)
        if response.status_code != 200:
            return f"搜索失败，状态码：{response.status_code}"
        
        results = response.json().get("organic", [])
        if not results:
            return "未找到相关搜索结果"
        
        # 格式化结果
        formatted = []
        for idx, res in enumerate(results, 1):
            formatted.append(
                f"【结果{idx}】\n标题：{res['title']}\n摘要：{res['snippet']}\n来源链接：{res['link']}\n"
            )
        return "\n".join(formatted)
    except Exception as e:
        return f"搜索异常：{str(e)}"

# ====================== 2. ReAct核心Prompt（严格格式约束） ======================
REACT_SYSTEM_PROMPT = """
你是一个严格遵循ReAct框架的智能检索助手，必须按照以下固定格式执行任务，禁止任何格式偏离。

【执行流程】
1. Thought：先思考用户的核心需求，要完成需求需要获取什么信息，下一步应该调用什么工具
2. Action：仅输出要调用的工具和参数，固定格式：Action: 工具名(参数名=参数值)
3. Observation：工具执行返回的结果（由系统填充）
4. 重复1-3步，直到获取足够的信息，完成用户需求
5. 最终输出：Final Answer: 整理后的完整、准确的答案

【核心规则】
- 仅可使用web_search工具，禁止编造任何不存在的工具
- 所有事实性内容必须通过搜索验证，绝对禁止编造数据、时间、事件
- 每次仅执行一个Action，禁止一次性调用多个工具
- 必须严格按照格式输出，Thought、Action、Final Answer必须单独成行
- 最大执行步数为5步，超过步数必须输出Final Answer

【工具说明】
web_search(query: str, num_results: int = 5)：全网搜索，query为搜索关键词，num_results为返回结果数量
"""

# ====================== 3. ReAct Agent核心逻辑 ======================
def react_search_agent_run(user_query: str, max_steps: int = 5) -> str:
    # 初始化对话上下文
    messages = [
        {"role": "system", "content": REACT_SYSTEM_PROMPT},
        {"role": "user", "content": f"用户需求：{user_query}"}
    ]

    # 循环执行，限制最大步数防止无限循环
    for step in range(max_steps):
        print(f"\n===== 执行步骤 {step+1}/{max_steps} =====")
        # 调用LLM生成思考和行动
        try:
            response = dashscope.Generation.call(
                model=dashscope.Generation.Models.qwen_plus,
                messages=messages,
                temperature=0.1,  # 极低随机性，保证格式稳定
                timeout=15
            )
        except Exception as e:
            return f"模型调用失败：{str(e)}"
        
        output = response.output.text
        print(output)

        # 判断是否输出最终答案
        if "Final Answer:" in output:
            return output.split("Final Answer:")[-1].strip()
        
        # 解析Action，执行工具
        if "Action:" in output:
            try:
                # 提取工具名和参数
                action_str = output.split("Action:")[-1].strip()
                func_name = action_str.split("(")[0].strip()
                args_part = action_str.split("(")[1].rstrip(")").strip()
                
                # 解析参数字典
                args_dict = {}
                for arg in args_part.split(","):
                    if "=" not in arg:
                        continue
                    k, v = arg.split("=", 1)
                    args_dict[k.strip()] = v.strip().strip('"').strip("'")
                
                # 执行工具
                if func_name == "web_search":
                    observation = web_search(**args_dict)
                else:
                    observation = f"错误：不存在工具【{func_name}】，仅支持web_search工具"
                
                print(f"\nObservation: {observation}")
                # 把思考、行动、观察结果加入上下文
                messages.append({"role": "assistant", "content": output})
                messages.append({"role": "user", "content": f"Observation: {observation}"})
            
            except Exception as e:
                observation = f"工具执行失败：{str(e)}，请检查Action格式是否正确"
                print(f"\nObservation: {observation}")
                messages.append({"role": "assistant", "content": output})
                messages.append({"role": "user", "content": f"Observation: {observation}"})
    
    # 超过最大步数，返回最终结果
    return "执行步数已达上限，未能完成完整需求，请简化问题重试"

# ====================== 测试运行 ======================
if __name__ == "__main__":
    result = react_search_agent_run("2026年春节档票房排名，票房最高的3部电影的票房数据、导演和主演信息")
    print("\n" + "="*50 + "\n===== 最终答案 =====\n" + "="*50)
    print(result)

案例2_带长期记忆的专属助理Agent.py

import os
import json
import dashscope
import chromadb
from dotenv import load_dotenv
from chromadb.utils import embedding_functions

# 加载环境变量
load_dotenv()
dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")

# ====================== 1. 初始化向量数据库（长期记忆存储） ======================
# 通义千问embedding模型
embedding_func = embedding_functions.DashScopeEmbeddingFunction(
    api_key=dashscope.api_key,
    model_name="text-embedding-v2"
)

# 持久化存储（本地生成memory_db文件夹，永久保存记忆）
chroma_client = chromadb.PersistentClient(path="./memory_db")
memory_collection = chroma_client.get_or_create_collection(
    name="user_long_term_memory",
    embedding_function=embedding_func,
    metadata={"description": "用户长期记忆库"}
)

# ====================== 2. 记忆管理工具 ======================
def add_memory(memory_content: str, user_id: str = "default_user") -> str:
    """添加长期记忆到数据库"""
    try:
        memory_id = f"memory_{user_id}_{int(os.times()[4])}"
        memory_collection.add(
            documents=[memory_content],
            metadatas=[{"user_id": user_id}],
            ids=[memory_id]
        )
        return f"记忆添加成功：{memory_content}"
    except Exception as e:
        return f"记忆添加失败：{str(e)}"

def retrieve_memory(query: str, user_id: str = "default_user", top_k: int = 3) -> str:
    """从数据库检索相关记忆"""
    try:
        results = memory_collection.query(
            query_texts=[query],
            where={"user_id": user_id},
            n_results=top_k
        )
        if not results["documents"][0]:
            return "暂无相关长期记忆"
        # 格式化记忆
        formatted = []
        for idx, doc in enumerate(results["documents"][0], 1):
            formatted.append(f"记忆{idx}：{doc}")
        return "\n".join(formatted)
    except Exception as e:
        return f"记忆检索失败：{str(e)}"

# ====================== 3. 工具配置 ======================
TOOLS_CONFIG = [
    {
        "type": "function",
        "function": {
            "name": "add_memory",
            "description": "把用户提供的个人信息、偏好、日程等内容添加到长期记忆库",
            "parameters": {
                "type": "object",
                "properties": {
                    "memory_content": {"type": "string", "description": "要存储的记忆内容，例如：用户叫张三，对芒果过敏，每周三下午2点开周会"}
                },
                "required": ["memory_content"]
            }
        }
    }
]

# ====================== 4. 记忆Agent核心逻辑 ======================
def memory_agent_run(user_query: str, user_id: str = "default_user") -> str:
    # 第一步：先检索相关记忆，注入上下文
    related_memory = retrieve_memory(user_query, user_id)
    print(f"【检索到的相关记忆】\n{related_memory}\n")

    # 初始化对话
    messages = [
        {
            "role": "system",
            "content": "你是用户的专属个人助理，必须先参考用户的长期记忆回答问题，保证回答贴合用户的个人情况。如果用户提供了新的个人信息、偏好、日程、习惯等内容，必须调用add_memory工具存入长期记忆库。禁止编造记忆，所有记忆必须来自用户的对话内容。"
        },
        {"role": "user", "content": f"用户相关长期记忆：\n{related_memory}\n\n用户当前问题：{user_query}"}
    ]

    # 调用LLM
    try:
        response = dashscope.Generation.call(
            model=dashscope.Generation.Models.qwen_turbo,
            messages=messages,
            tools=TOOLS_CONFIG,
            result_format="message",
            timeout=10
        )
    except Exception as e:
        return f"模型调用失败：{str(e)}"

    output_msg = response.output.choices[0].message

    # 处理记忆添加
    if "tool_calls" in output_msg:
        tool_call = output_msg.tool_calls[0]
        func_name = tool_call["function"]["name"]
        try:
            func_args = json.loads(tool_call["function"]["arguments"])
        except json.JSONDecodeError:
            return "参数解析失败，请重新输入"

        if func_name == "add_memory":
            tool_result = add_memory(**func_args, user_id=user_id)
        else:
            tool_result = f"错误：不支持的工具【{func_name}】"

        # 生成最终回答
        messages.append(output_msg)
        messages.append({
            "role": "tool",
            "content": tool_result,
            "tool_call_id": tool_call["id"]
        })

        try:
            final_response = dashscope.Generation.call(
                model=dashscope.Generation.Models.qwen_turbo,
                messages=messages,
                result_format="message",
                timeout=10
            )
        except Exception as e:
            return f"结果生成失败：{str(e)}"
        
        return final_response.output.choices[0].message.content

    # 无工具调用，直接返回
    else:
        return output_msg.content

# ====================== 测试运行 ======================
if __name__ == "__main__":
    print("="*30 + " 第一次对话 " + "="*30)
    print(memory_agent_run("我叫李四，对芒果和海鲜过敏，喜欢喝无糖冰美式，每周三下午2点要开部门周会"))
    print("\n" + "="*30 + " 第二次对话 " + "="*30)
    print(memory_agent_run("下午给我推荐一杯喝的"))
    print("\n" + "="*30 + " 第三次对话 " + "="*30)
    print(memory_agent_run("朋友约我周三下午2点吃饭，我能去吗？"))
    print("\n" + "="*30 + " 第四次对话 " + "="*30)
    print(memory_agent_run("生日聚会给我推荐几个菜品"))

进阶通关考核_个人理财助手Agent.py

import os
import json
import requests
from dotenv import load_dotenv
from http import HTTPStatus
import dashscope
import chromadb
from chromadb.utils import embedding_functions

# 加载环境变量
load_dotenv()
dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")

# ====================== 向量数据库初始化（记忆存储） ======================
embedding_func = embedding_functions.DashScopeEmbeddingFunction(
    api_key=dashscope.api_key,
    model_name="text-embedding-v2"
)
chroma_client = chromadb.PersistentClient(path="./finance_memory_db")
finance_memory_collection = chroma_client.get_or_create_collection(
    name="user_finance_memory",
    embedding_function=embedding_func
)

# ====================== 工具函数库 ======================
# 1. 实时汇率查询
def get_exchange_rate(from_currency: str, to_currency: str, amount: float = 1.0) -> str:
    """查询实时货币汇率"""
    try:
        url = f"https://api.exchangerate.host/convert?from={from_currency}&to={to_currency}&amount={amount}"
        response = requests.get(url, timeout=10).json()
        if not response["success"]:
            return f"汇率查询失败：{response['error']['info']}"
        result = response["result"]
        rate = response["info"]["rate"]
        return f"汇率查询结果：{amount} {from_currency.upper()} = {result:.2f} {to_currency.upper()}，当前汇率：1 {from_currency.upper()} = {rate:.4f} {to_currency.upper()}"
    except Exception as e:
        return f"汇率查询异常：{str(e)}"

# 2. 理财收益计算
def calculate_finance_income(principal: float, annual_rate: float, months: int, invest_type: str = "一次性") -> str:
    """
    计算理财收益
    :param principal: 本金
    :param annual_rate: 年化利率（%），例如4.5代表年化4.5%
    :param months: 投资期限（月）
    :param invest_type: 投资类型，一次性/定投
    """
    try:
        rate = annual_rate / 100
        if invest_type == "一次性":
            total_income = principal * rate * (months / 12)
            total_amount = principal + total_income
            return (
                f"【一次性理财收益计算】\n"
                f"本金：{principal}元\n"
                f"年化利率：{annual_rate}%\n"
                f"投资期限：{months}个月\n"
                f"预期收益：{total_income:.2f}元\n"
                f"到期总金额：{total_amount:.2f}元"
            )
        elif invest_type == "定投":
            monthly_principal = principal
            total_principal = monthly_principal * months
            monthly_rate = rate / 12
            total_amount = monthly_principal * (((1 + monthly_rate) ** months - 1) / monthly_rate) * (1 + monthly_rate)
            total_income = total_amount - total_principal
            return (
                f"【基金定投收益计算】\n"
                f"每月定投金额：{monthly_principal}元\n"
                f"年化利率：{annual_rate}%\n"
                f"投资期限：{months}个月\n"
                f"投入总本金：{total_principal:.2f}元\n"
                f"预期收益：{total_income:.2f}元\n"
                f"到期总金额：{total_amount:.2f}元"
            )
        else:
            return "错误：投资类型仅支持「一次性」或「定投」"
    except Exception as e:
        return f"收益计算失败：{str(e)}"

# 3. 记忆管理工具
def add_finance_memory(memory_content: str, user_id: str = "default_user") -> str:
    """添加用户理财相关记忆"""
    try:
        memory_id = f"finance_memory_{user_id}_{int(os.times()[4])}"
        finance_memory_collection.add(
            documents=[memory_content],
            metadatas=[{"user_id": user_id}],
            ids=[memory_id]
        )
        return f"记忆添加成功：{memory_content}"
    except Exception as e:
        return f"记忆添加失败：{str(e)}"

def retrieve_finance_memory(query: str, user_id: str = "default_user", top_k: int = 3) -> str:
    """检索用户理财相关记忆"""
    try:
        results = finance_memory_collection.query(
            query_texts=[query],
            where={"user_id": user_id},
            n_results=top_k
        )
        if not results["documents"][0]:
            return "暂无相关理财记忆"
        formatted = [f"记忆{idx+1}：{doc}" for idx, doc in enumerate(results["documents"][0])]
        return "\n".join(formatted)
    except Exception as e:
        return f"记忆检索失败：{str(e)}"

# ====================== 工具配置 ======================
TOOLS_CONFIG = [
    {
        "type": "function",
        "function": {
            "name": "get_exchange_rate",
            "description": "查询实时货币汇率，仅用于汇率兑换相关问题",
            "parameters": {
                "type": "object",
                "properties": {
                    "from_currency": {"type": "string", "description": "原货币代码，例如：CNY、USD、EUR、JPY"},
                    "to_currency": {"type": "string", "description": "目标货币代码，例如：CNY、USD、EUR、JPY"},
                    "amount": {"type": "number", "description": "兑换金额，默认1.0"}
                },
                "required": ["from_currency", "to_currency"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate_finance_income",
            "description": "计算理财收益，支持一次性理财和基金定投，仅用于理财收益计算相关问题",
            "parameters": {
                "type": "object",
                "properties": {
                    "principal": {"type": "number", "description": "本金/每月定投金额"},
                    "annual_rate": {"type": "number", "description": "年化利率，单位%，例如4.5代表年化4.5%"},
                    "months": {"type": "integer", "description": "投资期限，单位月"},
                    "invest_type": {"type": "string", "description": "投资类型，可选值：一次性、定投，默认一次性"}
                },
                "required": ["principal", "annual_rate", "months"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "add_finance_memory",
            "description": "添加用户的理财偏好、风险承受能力、投资习惯等信息到长期记忆",
            "parameters": {
                "type": "object",
                "properties": {
                    "memory_content": {"type": "string", "description": "要存储的理财相关记忆内容"}
                },
                "required": ["memory_content"]
            }
        }
    }
]

# ====================== Agent核心逻辑 ======================
def finance_agent_run(user_query: str, user_id: str = "default_user") -> str:
    # 检索相关记忆
    related_memory = retrieve_finance_memory(user_query, user_id)
    print(f"【检索到的理财记忆】\n{related_memory}\n")

    messages = [
        {
            "role": "system",
            "content": "你是一个专业的个人理财助手，拥有汇率查询、理财收益计算、用户记忆管理能力。回答必须参考用户的长期记忆，贴合用户的风险偏好和投资习惯。用户提供的理财相关信息必须存入长期记忆。禁止编造金融数据，所有计算必须通过工具完成。"
        },
        {"role": "user", "content": f"用户理财相关记忆：\n{related_memory}\n\n用户当前问题：{user_query}"}
    ]

    try:
        response = dashscope.Generation.call(
            model=dashscope.Generation.Models.qwen_turbo,
            messages=messages,
            tools=TOOLS_CONFIG,
            result_format="message",
            timeout=10
        )
    except Exception as e:
        return f"模型调用失败：{str(e)}"

    output_msg = response.output.choices[0].message

    if "tool_calls" in output_msg:
        tool_call = output_msg.tool_calls[0]
        func_name = tool_call["function"]["name"]
        try:
            func_args = json.loads(tool_call["function"]["arguments"])
        except json.JSONDecodeError:
            return "参数解析失败，请重新输入"

        tool_map = {
            "get_exchange_rate": get_exchange_rate,
            "calculate_finance_income": calculate_finance_income,
            "add_finance_memory": add_finance_memory
        }
        if func_name in tool_map:
            tool_result = tool_map[func_name](**func_args, user_id=user_id) if func_name == "add_finance_memory" else tool_map[func_name](**func_args)
        else:
            tool_result = f"错误：不支持的工具【{func_name}】"

        messages.append(output_msg)
        messages.append({
            "role": "tool",
            "content": tool_result,
            "tool_call_id": tool_call["id"]
        })

        try:
            final_response = dashscope.Generation.call(
                model=dashscope.Generation.Models.qwen_turbo,
                messages=messages,
                result_format="message",
                timeout=10
            )
        except Exception as e:
            return f"结果生成失败：{str(e)}"
        
        return final_response.output.choices[0].message.content
    else:
        return output_msg.content

# ====================== 测试运行 ======================
if __name__ == "__main__":
    print("="*30 + " 测试1：收益计算 " + "="*30)
    print(finance_agent_run("我每个月定投1000元，年化4.5%，投3年，最终能拿到多少钱？"))
    print("\n" + "="*30 + " 测试2：汇率查询 " + "="*30)
    print(finance_agent_run("1000美元能兑换多少人民币？"))
    print("\n" + "="*30 + " 测试3：记忆存储 " + "="*30)
    print(finance_agent_run("我是保守型投资者，风险承受能力低，只买保本型理财，年化收益接受3%-4%"))
    print("\n" + "="*30 + " 测试4：记忆调用 " + "="*30)
    print(finance_agent_run("给我推荐适合我的理财方式"))

五、高手篇_生产级多智能体 完整代码

案例1_多智能体内容生产工作室.py

import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from langchain_community.llms import Tongyi

# 加载环境变量
load_dotenv()
os.environ["DASHSCOPE_API_KEY"] = os.getenv("DASHSCOPE_API_KEY")

# 初始化LLM
llm = Tongyi(model_name="qwen-plus", temperature=0.7)

# ====================== 1. 定义智能体角色 ======================
# 内容策划专家
planner = Agent(
    role="内容策划专家",
    goal="根据用户需求，策划出有传播性、逻辑清晰的爆款内容选题和详细大纲",
    backstory="你拥有10年新媒体内容策划经验，擅长打造全网爆款内容，能精准把握用户痛点和平台流量规则，输出的大纲逻辑严谨、可落地性强",
    llm=llm,
    verbose=True,
    allow_delegation=False,
    max_iter=5
)

# 资深内容作家
writer = Agent(
    role="资深内容作家",
    goal="根据策划大纲，写出内容详实、语言流畅、有感染力的原创深度文章",
    backstory="你是国内头部科技/职场媒体的专栏作家，擅长把复杂的技术内容讲得通俗易懂，文字风格接地气有干货，拒绝水文和套话",
    llm=llm,
    verbose=True,
    allow_delegation=False,
    max_iter=5
)

# 专业校对编辑
proofreader = Agent(
    role="专业校对编辑",
    goal="校对文章的错别字、语病、逻辑漏洞，优化语句流畅度，保证内容零错误",
    backstory="你有8年国家级出版社校对经验，对文字错误零容忍，能精准发现文章中的错别字、标点错误、语病、逻辑矛盾，同时优化语句的可读性",
    llm=llm,
    verbose=True,
    allow_delegation=False,
    max_iter=5
)

# SEO优化专家
seo_optimizer = Agent(
    role="SEO优化专家",
    goal="优化文章的标题、关键词、结构，提升搜索引擎排名和平台传播性",
    backstory="你有10年SEO和新媒体流量运营经验，熟悉百度、微信、小红书等平台的流量规则，能通过关键词优化让文章获得更多自然流量，同时不破坏文章的可读性",
    llm=llm,
    verbose=True,
    allow_delegation=False,
    max_iter=5
)

# 排版设计师
typesetter = Agent(
    role="新媒体排版设计师",
    goal="把文章排版成美观、易读的markdown格式，适配公众号、小红书、知乎等主流平台",
    backstory="你是资深新媒体排版设计师，擅长用markdown打造高阅读体验的排版，精通标题层级、重点标注、分段、列表等格式，让文章结构清晰、重点突出、阅读无压力",
    llm=llm,
    verbose=True,
    allow_delegation=False,
    max_iter=5
)

# ====================== 2. 定义任务流 ======================
# 任务1：内容策划
task_plan = Task(
    description="用户需求：写一篇面向零基础初学者的「Agent开发从入门到精通」爆款干货文章，目标受众是编程小白、AI爱好者、想转行AI的职场人。请完成：1. 3个爆款标题备选；2. 详细的文章大纲，包含每个章节的核心内容、写作要点、字数分配；3. 文章的核心卖点和用户痛点。",
    agent=planner,
    expected_output="完整的内容策划方案，包含爆款标题、详细文章大纲、核心卖点和用户痛点分析"
)

# 任务2：正文写作
task_write = Task(
    description="根据策划专家输出的文章大纲，完成全文写作，字数不少于3000字，要求：1. 内容通俗易懂，零基础能看懂；2. 干货满满，有实操步骤，拒绝空泛理论；3. 结构清晰，逻辑连贯；4. 语言接地气，有感染力，适合新媒体传播。",
    agent=writer,
    expected_output="完整的原创文章正文，符合大纲要求，字数不少于3000字",
    context=[task_plan]
)

# 任务3：校对优化
task_proofread = Task(
    description="对作家输出的全文进行专业校对，完成：1. 修正所有错别字、标点错误、语病；2. 修正逻辑漏洞和前后矛盾的内容；3. 优化不通顺的语句，提升可读性；4. 标注出所有修改的地方和修改原因。",
    agent=proofreader,
    expected_output="校对优化后的完整文章，附带修改说明",
    context=[task_write]
)

# 任务4：SEO优化
task_seo = Task(
    description="对校对后的文章进行SEO和流量优化，完成：1. 最终确定1个主标题+3个副标题，适配平台流量规则；2. 布局核心关键词，提升搜索排名；3. 优化小标题，提升吸引力；4. 优化开头和结尾，提升完读率和互动率。",
    agent=seo_optimizer,
    expected_output="SEO优化后的完整文章，包含优化后的标题、关键词布局",
    context=[task_proofread]
)

# 任务5：排版输出
task_typeset = Task(
    description="把SEO优化后的文章，排版成适配公众号、知乎、小红书的markdown格式，要求：1. 清晰的标题层级；2. 重点内容加粗标注；3. 合理分段，避免大段文字；4. 合适的列表和引用格式；5. 适配多平台的排版规范。",
    agent=typesetter,
    expected_output="排版完成的markdown格式完整文章，适配主流新媒体平台",
    context=[task_seo]
)

# ====================== 3. 组建团队，启动执行 ======================
content_studio_crew = Crew(
    agents=[planner, writer, proofreader, seo_optimizer, typesetter],
    tasks=[task_plan, task_write, task_proofread, task_seo, task_typeset],
    process=Process.sequential,  # 顺序执行，前一个任务完成后一个才启动
    verbose=True,
    full_output=True
)

# ====================== 运行 ======================
if __name__ == "__main__":
    print("="*50)
    print("🚀 多智能体内容生产工作室启动")
    print("="*50)
    final_result = content_studio_crew.kickoff()
    print("\n" + "="*50)
    print("✅ 内容生产完成，最终成品：")
    print("="*50)
    print(final_result)

案例2_带反思能力的代码开发Agent.py

import os
import json
import subprocess
import tempfile
from dotenv import load_dotenv
import dashscope

# 加载环境变量
load_dotenv()
dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")

# ====================== 1. 工具函数 ======================
def run_python_code(code: str) -> str:
    """运行Python代码，返回执行结果或报错信息"""
    try:
        # 创建临时文件
        with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False, encoding="utf-8") as f:
            f.write(code)
            temp_file_path = f.name
        
        # 执行代码，捕获输出和报错
        result = subprocess.run(
            ["python", temp_file_path],
            capture_output=True,
            text=True,
            timeout=10
        )

        # 删除临时文件
        os.unlink(temp_file_path)

        # 返回结果
        if result.returncode == 0:
            return f"✅ 代码执行成功\n输出结果：\n{result.stdout}"
        else:
            return f"❌ 代码执行失败\n错误信息：\n{result.stderr}"
    except subprocess.TimeoutExpired:
        return "❌ 代码执行超时，限制10秒"
    except Exception as e:
        return f"❌ 代码运行异常：{str(e)}"

# ====================== 2. 核心Prompt ======================
CODE_AGENT_PROMPT = """
你是一个具备代码编写、执行、反思、优化能力的Python开发专家，严格按照以下流程执行任务，禁止任何步骤跳过。

【执行流程】
1. 需求理解：拆解用户的代码开发需求，明确核心功能、输入输出、边界条件
2. 代码编写：生成完整、可运行的Python代码，代码必须有详细注释，符合PEP8规范
3. 代码执行：调用run_python_code工具运行代码，获取执行结果或报错信息
4. 反思复盘：如果代码执行失败，分析报错原因，找出代码中的问题；如果执行成功，分析代码的优化空间
5. 代码优化：根据反思结果，修改优化代码，重新执行
6. 循环3-5步，直到代码完美运行，且满足用户的所有需求
7. 最终输出：Final Code: 最终的完整代码 + Final Result: 代码执行结果

【核心规则】
- 所有代码必须通过run_python_code工具执行验证，禁止声称代码可运行但不执行
- 每次仅生成完整的代码，禁止生成代码片段
- 禁止生成危险代码，包括但不限于：删除文件、修改系统配置、访问非法网站、无限循环
- 最大执行步数为5步，超过步数必须输出最终结果
- 必须严格按照格式输出，禁止偏离格式

【工具说明】
run_python_code(code: str)：运行Python代码，返回执行结果或报错信息，code为完整的Python代码字符串
"""

# ====================== 3. Agent核心逻辑 ======================
def code_agent_run(user_requirement: str, max_steps: int = 5) -> str:
    messages = [
        {"role": "system", "content": CODE_AGENT_PROMPT},
        {"role": "user", "content": f"用户代码开发需求：{user_requirement}"}
    ]

    for step in range(max_steps):
        print(f"\n===== 执行步骤 {step+1}/{max_steps} =====")
        # 调用LLM
        try:
            response = dashscope.Generation.call(
                model=dashscope.Generation.Models.qwen_plus,
                messages=messages,
                tools=[
                    {
                        "type": "function",
                        "function": {
                            "name": "run_python_code",
                            "description": "运行Python代码，返回执行结果或报错信息",
                            "parameters": {
                                "type": "object",
                                "properties": {"code": {"type": "string", "description": "完整的Python代码"}},
                                "required": ["code"]
                            }
                        }
                    }
                ],
                result_format="message",
                temperature=0.2,
                timeout=15
            )
        except Exception as e:
            return f"模型调用失败：{str(e)}"

        output_msg = response.output.choices[0].message
        print(output_msg.content)

        # 判断是否输出最终结果
        if "Final Code:" in output_msg.content and "Final Result:" in output_msg.content:
            return output_msg.content

        # 处理代码执行
        if "tool_calls" in output_msg:
            tool_call = output_msg.tool_calls[0]
            func_name = tool_call["function"]["name"]
            try:
                func_args = json.loads(tool_call["function"]["arguments"])
            except json.JSONDecodeError:
                observation = "代码解析失败，请重新生成完整代码"
                print(f"\nObservation: {observation}")
                messages.append(output_msg)
                messages.append({"role": "user", "content": f"Observation: {observation}"})
                continue

            if func_name == "run_python_code":
                observation = run_python_code(**func_args)
            else:
                observation = f"错误：不支持的工具【{func_name}】"

            print(f"\nObservation: {observation}")
            messages.append(output_msg)
            messages.append({"role": "user", "content": f"Observation: {observation}"})

    # 超过最大步数
    return "执行步数已达上限，最终代码和结果如下：\n" + output_msg.content

# ====================== 测试运行 ======================
if __name__ == "__main__":
    # 测试需求：生成一个斐波那契数列生成器，带用户输入和异常处理
    result = code_agent_run("写一个Python斐波那契数列生成器，用户输入要生成的项数，输出对应的斐波那契数列，要有完善的异常处理，比如用户输入非数字、负数的情况")
    print("\n" + "="*50 + "\n===== 最终结果 =====\n" + "="*50)
    print(result)

部署套件

agent_api.py（FastAPI接口封装）
import os
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn

# 加载环境变量
load_dotenv()

# 导入之前开发的Agent函数
from 入门篇_零基础Agent.案例2_多工具全能Agent import multi_tool_agent_run
from 进阶篇_实用商用Agent.案例1_ReAct全网检索Agent import react_search_agent_run
from 进阶篇_实用商用Agent.案例2_带长期记忆的专属助理Agent import memory_agent_run

# 初始化FastAPI
app = FastAPI(title="Agent全栈API服务", version="1.0.0", description="开箱即用的Agent API接口")

# 定义请求体模型
class AgentRequest(BaseModel):
    query: str
    user_id: str = "default_user"
    max_steps: int = 5

# 健康检查接口
@app.get("/health", summary="健康检查")
async def health_check():
    return {"status": "ok", "message": "Agent API服务运行正常"}

# 多工具Agent接口
@app.post("/api/multi-tool-agent", summary="多工具全能Agent")
async def multi_tool_agent(request: AgentRequest):
    try:
        result = multi_tool_agent_run(request.query)
        return {"code": 200, "result": result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# 全网检索Agent接口
@app.post("/api/react-search-agent", summary="ReAct全网检索Agent")
async def react_search_agent(request: AgentRequest):
    try:
        result = react_search_agent_run(request.query, request.max_steps)
        return {"code": 200, "result": result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# 记忆助理Agent接口
@app.post("/api/memory-agent", summary="带长期记忆的专属助理Agent")
async def memory_agent(request: AgentRequest):
    try:
        result = memory_agent_run(request.query, request.user_id)
        return {"code": 200, "result": result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# 启动服务
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)
agent_webui.py（Streamlit可视化前端）
import streamlit as st
import requests

# 页面配置
st.set_page_config(
    page_title="Agent全栈开发平台",
    page_icon="🤖",
    layout="wide"
)

# 页面标题
st.title("🤖 零基础Agent开发平台")
st.divider()

# 侧边栏：Agent选择
with st.sidebar:
    st.header("Agent选择")
    agent_type = st.radio(
        "选择要使用的Agent",
        options=["多工具全能Agent", "ReAct全网检索Agent", "带长期记忆的专属助理Agent"],
        index=0
    )
    st.divider()
    user_id = st.text_input("用户ID", value="default_user", help="用于区分不同用户的记忆数据")
    max_steps = st.slider("最大执行步数", min_value=1, max_value=10, value=5, help="Agent最大循环执行次数")
    st.divider()
    st.caption("API服务地址")
    api_base_url = st.text_input("API地址", value="http://127.0.0.1:8000")

# 主界面：对话输入
st.subheader(f"当前使用：{agent_type}")
user_query = st.text_area("输入你的问题", height=100, placeholder="例如：成都今天天气怎么样？帮我算一下123*456等于多少？")
run_button = st.button("🚀 执行", type="primary", use_container_width=True)

# 结果展示
st.divider()
result_container = st.container()

# 执行逻辑
if run_button and user_query:
    with st.spinner("Agent正在执行中..."):
        try:
            # 匹配接口
            api_map = {
                "多工具全能Agent": "/api/multi-tool-agent",
                "ReAct全网检索Agent": "/api/react-search-agent",
                "带长期记忆的专属助理Agent": "/api/memory-agent"
            }
            api_url = api_base_url + api_map[agent_type]

            # 发起请求
            response = requests.post(
                api_url,
                json={
                    "query": user_query,
                    "user_id": user_id,
                    "max_steps": max_steps
                },
                timeout=60
            )

            if response.status_code == 200:
                result = response.json()["result"]
                result_container.subheader("✅ 执行结果")
                result_container.markdown(result)
            else:
                result_container.error(f"请求失败：{response.text}")
        except Exception as e:
            result_container.error(f"执行异常：{str(e)}，请检查API服务是否正常启动")

六、配套文档

00_快速启动指南.md

# 快速启动指南（5分钟跑通）
## 1. 前置准备
- 安装Python 3.10+（官网下载：https://www.python.org/downloads/，安装时勾选「Add Python to PATH」）
- 注册并获取所需的API Key（参考.env.example里的注册地址）

## 2. 快速启动步骤
1. 下载整个代码包到本地
2. 复制`.env.example`文件，改名为`.env`，填入你申请的API Key
3. 打开cmd/终端，进入代码包根目录，执行依赖安装：
   ```bash
   pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
  1. 运行第一个Agent案例： 
    python 入门篇_零基础Agent/案例1_极简天气查询Agent.py
  2. 看到运行结果，恭喜你！你的第一个Agent已经成功跑通了！

3. 进阶启动

  • 运行API服务：python 高手篇_生产级多智能体/部署套件/agent_api.py
  • 运行可视化前端：新开终端，执行streamlit run 高手篇_生产级多智能体/部署套件/agent_webui.py

### 01_环境配置全流程.md
```markdown
# 环境配置全流程（全系统兼容）
## Windows系统
1. 安装Python 3.10+，官网下载安装包，安装时必须勾选「Add Python to PATH」
2. 验证安装：打开cmd，执行`python --version`和`pip --version`，能输出版本号即为成功
3. 安装依赖：cmd进入代码包根目录，执行`pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple`
4. 配置环境变量：复制.env.example为.env，填入API Key

## Mac/Linux系统
1. 安装Python 3.10+：Mac用`brew install python3`，Linux用`sudo apt install python3 python3-pip`
2. 验证安装：终端执行`python3 --version`和`pip3 --version`
3. 安装依赖：终端进入代码包根目录，执行`pip3 install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple`
4. 配置环境变量：复制.env.example为.env，填入API Key

## 常见安装问题解决
1. pip安装失败：换国内镜像源，加上`-i https://pypi.tuna.tsinghua.edu.cn/simple`
2. chromadb安装失败：Windows先安装Visual Studio Build Tools，Mac执行`xcode-select --install`
3. 权限报错：Linux/Mac加上`sudo`，Windows用管理员身份打开cmd

02_常见问题排障手册.md

# 常见问题排障手册
## 1. API Key相关报错
### 报错：Invalid API Key
- 原因：API Key填写错误、未开通对应服务、额度用完
- 解决方案：
  1. 检查.env里的API Key是否正确，无多余空格
  2. 登录对应平台，确认API Key已开通对应服务权限
  3. 确认平台账户有免费额度/余额

### 报错：API调用超时
- 原因：网络问题、代理冲突、国内访问海外API受限
- 解决方案：
  1. 关闭VPN/代理
  2. 优先使用国内的通义千问API，避免海外API访问受限
  3. 检查网络是否正常

## 2. 工具调用相关报错
### 报错：参数解析失败
- 原因：LLM返回的参数格式不是标准JSON，用eval解析报错
- 解决方案：
  1. 已替换为json.loads解析，确保代码是最新版本
  2. 降低temperature到0.1-0.3，提升LLM输出格式稳定性
  3. 优化工具的description，明确参数要求

### 报错：工具执行无返回/报错
- 原因：API接口变更、网络超时、参数错误
- 解决方案：
  1. 单独测试工具函数，确认函数本身能正常运行
  2. 增加timeout参数，避免网络超时
  3. 完善异常捕获，输出详细报错信息
# 人工智能
Logo
DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

cover

VLAN配置实战指南：PVID与接口模式详解

avatar DAMO开发者矩阵

Kalibr 完全指南：从原理推导到ROS实战，掌握相机-IMU高精度标定

多传感器融合是机器人感知的核心，而传感器之间的精确标定是实现高质量融合的前提。Kalibr 作为瑞士苏黎世联邦理工学院开源的标定工具箱，能够同时校准相机内参、多相机外参、相机-IMU 外参以及时间偏移，已成为视觉-惯性SLAM领域的标配工具。本文将带你深入 Kalibr 的数学原理，推导视觉重投影误差与IMU预积分公式，解析 ROS 功能包的整体流程，并教你如何分析标定结果、诊断常见问题。无论你是

avatar DAMO开发者矩阵
cover

国内具身 AI 数据创业公司盘点：8 家玩家的路径对比（2026 Q2）

avatar DAMO开发者矩阵