8、Agent的高级用法4：流式输出及模式

8.1 流式输出的说明

通过 invoke 调用Agent时，内部可能经历多次调用，长时间看不到调用情况，用户体验不好，可以通 过 流式调用 （渐进式显示输出）优化用户体验， 实时显示 Agent 运行过程中的更新。特别是在处理 LLM 延迟时尤其有效。

流式输出好处：

• 大型语言模型生成完整响应通常需要几秒钟时间，对于长输出可能达到10-20 秒，用户期望即时反 馈， 流式传输让等待过程更加可控 。
• 相比非流式传输需要用户长时间等待完整响应，流式传输可以立即显示文字逐渐出现的效果， 大 幅降低用户的等待焦虑 。

设置方式：

通过“ agent.stream(stream_mode=指定模式) ”来指定。具体模式有：values、updates（默认）、 messages、custom、checkpoints、tasks、debug。

8.2 具体的输出模式

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

load_dotenv(override=True)

model=init_chat_model(
    model='deepseek:deepseek-v4-pro'
)

values输出模式

当 stream_mode 设置为values模式时，每个步骤执行后，都会输出完整的状态信息，适用于每一步都 要获取完整状态、状态持久化场景。

举例：

from langchain.agents import create_agent
from langchain.tools import tool
from typing import Dict, Any
from rich import print as rprint


@tool
def query_customer_data(customer_id: str) -> Dict[str, Any]:
    """
    查询客户基本信息

    Args:
        customer_id: 客户ID，用于唯一标识客户

    Returns:
        包含客户基本信息的字典，如姓名、等级、加入日期等
    """

    # 模拟数据库查询
    return {
        "name": "张三",
        "level": "VIP",
        "join_date": "2023-01-15"
    }


@tool
def check_order_history(customer_id: str) -> Dict[str, Any]:
    """
    查询客户订单历史

    Args:
        customer_id: 客户ID，用于唯一标识客户

    Returns:
        包含客户订单历史的字典，如总订单数、总花费等
    """

    return {
        "total_orders": 15,
        "total_spent": 25800.00
    }


@tool
def get_current_promotions() -> Dict[str, Any]:
    """
    获取当前可用促销活动

    Returns:
        包含当前可用促销活动的字典，如活动名称、有效日期等
    """

    return {
        "promotions": [
            "老用户优惠",
            "会员专属折扣"
        ],
        "valid_until": "2027-01-31"
    }


# 创建客户服务 Agent
customer_service_agent = create_agent(
    model=model,
    tools=[
        query_customer_data,
        check_order_history,
        get_current_promotions
    ]
)

for chunk in customer_service_agent.stream(
    {
        "messages":[
            {"role":"user","content":"查询客户id为cust1234的完整信息，和历史订单和可用的优惠"}
        ]
    },
    stream_mode="values"
):
    rprint(chunk)
    print("==" * 50)

代码运行后示意结果如下：

updates输出模式

这种模式就是默认模式。该模式中，每个步骤执行后，只增量更新状态中发生变化的内容，用于监控 Agent 执行进度，例如观察Agent决定调用工具、工具执行结果等步骤

# 其他工具代码同上，保持不变
for chunk in customer_service_agent.stream(
    {
        "messages":[
            {"role":"user","content":"查询客户id为cust1234的完整信息，和历史订单和可用的优惠"}
        ]
    },

):
    rprint(chunk)
    print("==" * 50)

运行结果如下：

messages输出模式

该模式中会输出流式返回的Token以及相关的元数据（如：来自哪个节点），可以用在实现类似 ChatGPT 的打字机效果场景，为聊天机器人等交互式应用提供最佳的实时体验。

# 其他工具代码同上，保持不变

for chunk in customer_service_agent.stream(
    {
        "messages":[
            {"role":"user","content":"查询客户id为cust1234的完整信息，和历史订单和可用的优惠"}
        ]
    },
    stream_mode="messages"
):
 print(chunk[0].content,end="",flush=True)

运行结果如下：

tasks输出模式

该模式会输出当前task任务开始和结束的时间，包含任务的结果和错误信息，该模式用于监控任务的生 命周期。

 # 其他工具代码同上，保持不变
for chunk in customer_service_agent.stream(
    {
        "messages":[
            {"role":"user","content":"查询客户id为cust1234的完整信息，和历史订单和可用的优惠"}
        ]
    },
    stream_mode="tasks"
):
    rprint(chunk)
    print("==" * 50)

debug输出模式

该模式与tasks模式类似，比task模式多输出任务步骤、时间戳、task类型（task/task_result），该模式 用于调试、监控task任务的生命周期。

# 其他工具代码同上，保持不变
for chunk in customer_service_agent.stream(
    {
        "messages":[
            {"role":"user","content":"查询客户id为cust1234的完整信息，和历史订单和可用的优惠"}
        ]
    },
    stream_mode="debug"
):
    rprint(chunk)
    print("==" * 50)

custom输出模式

开发者通过 get_stream_writer 在工具或节点内部 自定义发送的数据 ，用于 输出 业务逻辑相关的进 度信息（如“已处理10/100条记录”）、自定义日志或指标。

举例：生成销售报告和库存报告Agent

from langchain.agents import create_agent
from langgraph.config import get_stream_writer
from langchain.tools import tool
import time


@tool
def generate_sales_report() -> str:
    """生成销售报告"""

    writer = get_stream_writer()

    writer({
        "type": "生成销售报告",
        "message": "开始生成销售报告"
    })

    # 模拟数据处理
    for i in range(1, 4):
        time.sleep(0.5)

        writer({
            "type": "生成销售报告",
            "message": f"生成销售报告进度百分比：{i * 25}%"
        })

    writer({
        "type": "生成销售报告",
        "message": "报告生成完成"
    })

    return "销售报告：总收入150万元，同比增长12%"


@tool
def generate_inventory_report() -> str:
    """生成库存报告"""

    writer = get_stream_writer()

    writer("开始库存分析...")

    time.sleep(0.5)

    writer("检查当前库存量...")

    time.sleep(0.5)

    writer("生成库存报告...")

    return "当前库存量为10000件，库存充足，无异常"


# 创建报告生成 Agent
reporting_agent = create_agent(
    model=model,
    tools=[
        generate_sales_report,
        generate_inventory_report
    ]
)


for chunk in reporting_agent.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": "生成销售报告和库存报告"
            }
        ]
    },
    stream_mode="custom"
):
    print(chunk)
    print("-" * 50)

运行结果如下

{'type': '生成销售报告', 'message': '开始生成销售报告'}
--------------------------------------------------
开始库存分析...
--------------------------------------------------
{'type': '生成销售报告', 'message': '生成销售报告进度百分比：25%'}
--------------------------------------------------
检查当前库存量...
--------------------------------------------------
{'type': '生成销售报告', 'message': '生成销售报告进度百分比：50%'}
--------------------------------------------------
生成库存报告...
--------------------------------------------------
{'type': '生成销售报告', 'message': '生成销售报告进度百分比：75%'}
--------------------------------------------------
{'type': '生成销售报告', 'message': '报告生成完成'}
--------------------------------------------------

8.3 流式输出模式总结

如下是LangChain Agent输出模式对比：

我们可以根据不同的目标来选择不同的输出模式。例如：

• 实现 实时对话交互 ，优先选择messages模式；
• 观察Agent的 思考与执行步骤 ，优先选择updates模式；
• 需要查看 每一步状态 优先选择values/tasks/debug模式；
• 工具执行时 输出自定义业务 日志优先选择custom模式。

此外，以上这些模式还可以组合使用，例如，可以同时指定 stream_mode=[“tasks”，“updates”] ，这 样在同一个循环里 既能 查看Agent task任务执行内容， 又能 显示Agent每步的更新

当指定多模式后，可以通过“ for stream_mode, chunk in customer_service_agent.stream... ”来遍历 dict，dict的key(stream_mode)是执行模式，value(chunk)是该模式输出的结果。

9、实战：多功能智能助手

项目需求：开发一个多功能智能助手，支持：

• 1. 天气查询：查询城市天气
• 2. 数学计算：复杂数学运算
• 3. 时间查询：获取当前时间、日期计算
• 4. 货币转换：多种货币之间转换 5. 信息搜索：搜索产品、新闻等信息

9.1 模型的初始化

import  os
from dotenv import load_dotenv
from langchain.chat_models import  init_chat_model

#  从env加载环境变量
load_dotenv(override=True)

#模型的初始化
model=init_chat_model(
    model='deepseek-v4-pro',
    model_provider="openai",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL")
)

9.2 工具的定义

from langchain_core.tools import tool
import math
from datetime import datetime, timedelta


@tool
def get_weather(city: str) -> str:
    """
    获取指定城市的实时天气信息

    支持中国主要城市的天气查询

    Args:
        city: 城市名称，如"北京"、"上海"、"深圳"等

    Returns:
        包含温度、天气状况、空气质量的详细信息

    Examples:
        get_weather("北京") 返回 "多云，15-22℃，空气质量良"
    """

    weather_db = {
        "北京": "多云，15-22℃，空气质量良，湿度 45%",
        "上海": "晴天，18-25℃，空气质量优，湿度 60%",
        "深圳": "小雨，22-28℃，空气质量优，湿度 75%",
        "成都": "阴天，16-23℃，空气质量良，湿度 70%",
        "杭州": "晴天，17-24℃，空气质量优，湿度 55%",
        "广州": "多云，21-29℃，空气质量良，湿度 72%"
    }

    result = weather_db.get(city)

    if result:
        return f"{city}：{result}"

    return (
        f"抱歉，暂不支持查询{city}的天气信息。"
        "当前支持：北京、上海、深圳、成都、杭州、广州"
    )


@tool
def calculator(expression: str) -> str:
    """
    执行数学计算

    支持基本运算符（+、-、*、/、**）和常用数学函数

    Args:
        expression:
            数学表达式，可以包含：
            - 基本运算：2 + 3, 10 * 5, 100 / 4
            - 幂运算：2 ** 10
            - 函数：sqrt(16), abs(-5), pow(2, 3)

    Returns:
        计算结果或错误信息

    Examples:
        calculator("2 + 3 * 4") 返回 "14"
        calculator("sqrt(16)") 返回 "4.0"
    """

    try:
        safe_functions = {
            "sqrt": math.sqrt,
            "pow": pow,
            "abs": abs,
            "round": round,
            "sin": math.sin,
            "cos": math.cos,
            "tan": math.tan,
            "log": math.log,
            "pi": math.pi,
            "e": math.e
        }

        result = eval(
            expression,
            {"__builtins__": {}},
            safe_functions
        )

        return f"{expression} = {result}"

    except Exception as e:
        return (
            f"计算出错：{str(e)}\n"
            "提示：请检查表达式格式，支持的函数有 "
            "sqrt, abs, pow, sin, cos, tan, log"
        )


@tool
def get_time_info(query_type: str = "current") -> str:
    """
    获取时间相关信息

    Args:
        query_type:
            - current: 当前时间
            - date: 今天日期
            - tomorrow: 明天日期
            - yesterday: 昨天日期
            - weekday: 星期几

    Returns:
        时间信息字符串
    """

    now = datetime.now()

    if query_type == "current":
        return now.strftime(
            "当前时间：%Y年%m月%d日 %H:%M:%S"
        )

    elif query_type == "date":
        return now.strftime(
            "今天是：%Y年%m月%d日"
        )

    elif query_type == "tomorrow":
        tomorrow = now + timedelta(days=1)
        return tomorrow.strftime(
            "明天是：%Y年%m月%d日"
        )

    elif query_type == "yesterday":
        yesterday = now - timedelta(days=1)
        return yesterday.strftime(
            "昨天是：%Y年%m月%d日"
        )

    elif query_type == "weekday":
        weekdays = [
            "星期一",
            "星期二",
            "星期三",
            "星期四",
            "星期五",
            "星期六",
            "星期日"
        ]

        return f"今天是{weekdays[now.weekday()]}"

    else:
        return (
            f"不支持的查询类型：{query_type}。"
            "支持：current, date, tomorrow, yesterday, weekday"
        )


@tool
def convert_currency(
    amount: float,
    from_curr: str,
    to_curr: str
) -> str:
    """
    货币转换工具

    Args:
        amount: 金额数值
        from_curr: 源货币代码（CNY/USD/EUR/GBP/JPY/HKD）
        to_curr: 目标货币代码（CNY/USD/EUR/GBP/JPY/HKD）

    Returns:
        转换结果
    """

    exchange_rates = {
        "CNY": 1.0,
        "USD": 0.14,
        "EUR": 0.13,
        "GBP": 0.11,
        "JPY": 20.8,
        "HKD": 1.09
    }

    currency_names = {
        "CNY": "人民币",
        "USD": "美元",
        "EUR": "欧元",
        "GBP": "英镑",
        "JPY": "日元",
        "HKD": "港币"
    }

    from_curr = from_curr.upper()
    to_curr = to_curr.upper()

    if from_curr not in exchange_rates:
        return f"不支持的源货币：{from_curr}"

    if to_curr not in exchange_rates:
        return f"不支持的目标货币：{to_curr}"

    cny_amount = amount / exchange_rates[from_curr]
    result_amount = cny_amount * exchange_rates[to_curr]

    return (
        f"{amount} {currency_names[from_curr]}（{from_curr}）"
        f" = {result_amount:.2f}"
        f" {currency_names[to_curr]}（{to_curr}）"
    )


@tool
def search_info(
    keyword: str,
    category: str = "all"
) -> str:
    """
    搜索各类信息

    Args:
        keyword: 搜索关键词
        category:
            - product: 搜索产品
            - news: 搜索新闻
            - all: 搜索所有

    Returns:
        搜索结果
    """

    products = {
        "手机": "iPhone 15 (¥5999), 小米14 (¥3999), 华为Mate60 (¥6999)",
        "笔记本": "MacBook Pro (¥12999), ThinkPad X1 (¥9999), 华为MateBook (¥7999)",
        "耳机": "AirPods Pro (¥1999), Sony WH-1000XM5 (¥2499)"
    }

    news = {
        "AI": "1. GPT-5 即将发布 2. AI 芯片市场增长 30% 3. 新AI法规出台",
        "科技": "1. 量子计算新突破 2. 6G 技术测试 3. 新能源汽车销量创新高"
    }

    results = []

    if category in ["product", "all"]:
        for key, value in products.items():
            if keyword in key:
                results.append(
                    f"【产品】{key}：{value}"
                )

    if category in ["news", "all"]:
        for key, value in news.items():
            if keyword in key or keyword in value:
                results.append(
                    f"【新闻】{key} 相关：{value}"
                )

    if results:
        return "\n".join(results)

    return f"未找到关于 '{keyword}' 的{category}信息"

9.3 agent的创建

from langchain.agents import create_agent


class SmartAssistant:
    """多功能智能助手"""

    def __init__(self):
        # 初始化模型
        self.model = model

        # 初始化工具
        self.tools = [
            get_weather,
            calculator,
            get_time_info,
            convert_currency,
            search_info
        ]

        # 系统提示词
        system_prompt = """
你是一个多功能智能助手，可以帮助用户：

🌤 查询天气：使用 get_weather 工具

🔢 数学计算：使用 calculator 工具

⏰ 时间查询：使用 get_time_info 工具

💱 货币转换：使用 convert_currency 工具

🔍 信息搜索：使用 search_info 工具

重要提示：

1. 仔细阅读用户问题，确定需要使用哪个工具。

2. 如果需要多个工具，按顺序调用。

3. 总是用友好、专业的语气回答。

4. 如果工具返回了数据，要用通俗易懂的语言解释给用户。

5. 如果无法完成任务，诚实地告诉用户原因。

请始终使用中文回答。
"""

        # 创建 Agent
        self.agent = create_agent(
            model=self.model,
            tools=self.tools,
            system_prompt=system_prompt
        )

        # 保存历史消息
        self.message = []


    def chat(self, user_input: str) -> str:
        """对话接口"""

        # 添加用户消息
        self.message.append(
            {
                "role": "user",
                "content": user_input
            }
        )

        # 调用 Agent
        result = self.agent.invoke(
            {
                "messages": self.message
            }
        )

        # 更新历史消息
        self.message = result["messages"]

        # 返回最终答案
        for msg in reversed(self.message):
            if msg.type == "ai" and msg.content:
                return msg.content

        return "抱歉，我无法处理这个问题"


    def reset(self):
        """重置对话历史"""

        self.message = []

 1. model 模型
 2. tools 工具
 3. system_prompt 系统提示词
 4. ToolStrategy (体现结构话输出)
 5. invoke 消息列表

9.4 主程序

# ==================== 主程序 ====================

def main():
    assistant = SmartAssistant()

    print("=" * 40)
    print("🤖 多功能智能助手（LangChain 1.2）")
    print("=" * 40)

    print("\n我可以帮你：")

    print(" 🌤 查询天气")
    print(" 🔢 数学计算")
    print(" ⏰ 时间查询")
    print(" 💱 货币转换")
    print(" 🔍 信息搜索")

    print("\n输入 'quit' 退出，输入 'reset' 重置对话\n")


    demos = [
        "北京今天天气怎么样？",
        "帮我算一下 (25 + 17) * 3",
        "现在几点了？",
        "100 美元等于多少人民币？"
    ]


    for demo in demos:
        print(f"👤 {demo}")

        response = assistant.chat(demo)

        print(f"🤖 {response}\n")


    # 重置对话
    assistant.reset()


    # 交互模式
    print("=" * 40)
    print("💬 进入交互模式")
    print("=" * 40)


    while True:
        user_input = input("\n👤 你: ")


        if user_input.lower() == "quit":
            print("再见！👋")
            break


        if user_input.lower() == "reset":
            assistant.reset()
            print("✅ 对话已重置")
            continue


        if not user_input.strip():
            continue


        # 调用助手
        response = assistant.chat(user_input)

        print(f"🤖 助手: {response}")


if __name__ == "__main__":
    main()