🚀 打造智能市场洞察报告系统：用 LangGraph + AI 实现每日自动化推送
前言
在建筑工程和造价软件行业，每天都需要关注行业龙头企业、智能标书审查、AI自动编标、标书检查等领域的市场动态。传统的信息搜集方式需要：

手动搜索多个关键词
筛选和整理资讯
撰写分析报告
手动发送到群组
这个过程耗时耗力，而且容易遗漏重要信息。今天，我将分享如何用 LangGraph 工作流引擎 + AI 大模型 + 飞书机器人构建一个全自动化的市场洞察报告系统，每天早上8点自动推送专业报告！

📊 项目亮点
核心功能
✅ 智能搜索：自动搜索 11 个行业关键词的市场动态
✅ AI 分析：基于搜索结果生成专业洞察报告
✅ 定时推送：每天早上8点自动发送到飞书机器人
✅ 结构化输出：报告包含市场焦点、关键动态、行业趋势、竞争格局、商业建议
技术栈
工作流引擎：LangGraph（构建有向无环图工作流）
AI 大模型：豆包 Doubao（文本生成与分析）
Web 搜索：coze-coding-dev-sdk（实时信息检索）
飞书集成：飞书自定义机器人 webhook
定时任务：APScheduler（灵活的 Python 定时调度）
🏗️ 系统架构
工作流设计
整个系统采用线性工作流设计，包含三个核心节点：

┌─────────────────┐
│ 市场信息搜集 │
│ (Web Search) │
└────────┬────────┘
│
↓
┌─────────────────┐
│ 报告生成 │
│ (AI Agent) │
└────────┬────────┘
│
↓
┌─────────────────┐
│ 飞书消息发送 │
│ (Feishu Bot) │
└─────────────────┘
数据流转
关键词列表 → 搜索结果 + AI摘要 → 结构化报告 → 飞书推送
↓ ↓ ↓ ↓
11个关键词 15条资讯 1200+字每日8点
🚀 快速开始

克隆项目
git clone
cd market-insight-workflow
安装依赖
pip install -r requirements.txt
核心依赖：

langgraph>=1.0.0 - 工作流引擎
coze-coding-dev-sdk>=0.5.0 - AI 和搜索 SDK
apscheduler>=3.10.0 - 定时任务
requests>=2.31.0 - HTTP 请求
3. 配置飞书机器人
编辑 src/graphs/nodes/feishu_send_node.py：

修改为你的飞书 webhook 地址

webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/你的webhook地址"
获取飞书 webhook 地址：

在飞书群组中添加"自定义机器人"
复制 webhook URL
替换代码中的地址
4. 立即测试
python scripts/test_run.py
你会看到：

============================================================
🧪 测试：立即执行市场洞察报告
⏰ 时间: 2026-02-12 18:54:02

📊 开始搜索市场信息...
🔑 关键词: 行业龙头企业, 智能标书审查, 标书检查, 某软件公司, ai编标...

✅ 报告生成成功！
📊 报告长度: 1787 字符
📤 发送状态: success
🎉 已成功发送到飞书机器人！
5. 启动定时任务
python scripts/scheduler.py
启动后，系统会每天早上8点自动执行任务！

📝 核心代码解析

状态定义（State）

src/graphs/state.py

class GlobalState(BaseModel):
"""全局状态定义"""
keywords: List[str] = Field(default=[], description="搜索关键词列表")
search_results: List[Dict] = Field(default=[], description="市场信息搜索结果")
ai_summary: str = Field(default="", description="AI生成的搜索摘要")
report_content: str = Field(default="", description="市场洞察报告内容")
send_status: str = Field(default="", description="飞书消息发送状态")
2. 市场信息搜集节点

src/graphs/nodes/market_search_node.py

def market_search_node(
state: MarketSearchInput,
config: RunnableConfig,
runtime: Runtime[Context]
) -> MarketSearchOutput:
"""搜索市场信息"""
client = SearchClient(ctx=new_context(method="search.web"))

# 组合关键词进行搜索
query = " ".join(state.keywords) + " 市场动态 行业趋势"

# 执行搜索并获取AI摘要
response = client.web_search_with_summary(query=query, count=15)

return MarketSearchOutput(
    search_results=[...],
    ai_summary=response.summary
)

关键点：

使用 web_search_with_summary 方法，自动获取 AI 生成的摘要
搜索15条结果，保证信息量充足
3. 报告生成节点（Agent 节点）

src/graphs/nodes/report_generation_node.py

def report_generation_node(
state: ReportGenerationInput,
config: RunnableConfig,
runtime: Runtime[Context]
) -> ReportGenerationOutput:
"""生成市场洞察报告"""
# 读取配置文件
cfg_file = os.path.join(os.getenv("COZE_WORKSPACE_PATH"), config['metadata']['llm_cfg'])
with open(cfg_file, 'r') as fd:
_cfg = json.load(fd)

# 初始化 LLM 客户端
client = LLMClient(ctx=new_context(method="invoke"))

# 渲染提示词模板
up_tpl = Template(_cfg["up"])
user_prompt = up_tpl.render({
    "search_results": state.search_results,
    "ai_summary": state.ai_summary
})

# 调用大模型
response = client.invoke(
    messages=[
        SystemMessage(content=_cfg["sp"]),
        HumanMessage(content=user_prompt)
    ],
    model="doubao-seed-1-8-251228",
    temperature=0.7
)

return ReportGenerationOutput(report_content=response.content)

提示词模板（config/report_generation_cfg.json）：

{
"sp": "你是专业的市场洞察报告撰写专家...",
"up": "请基于以下搜索结果和摘要，生成市场洞察报告：\n\n搜索结果：\n{% for result in search_results %}\n- 标题：{{ result.title }}\n 来源：{{ result.site_name }}\n{% endfor %}\n\nAI摘要：\n{{ ai_summary }}"
}
4. 飞书发送节点

src/graphs/nodes/feishu_send_node.py

def feishu_send_node(
state: FeishuSendInput,
config: RunnableConfig,
runtime: Runtime[Context]
) -> FeishuSendOutput:
"""发送飞书消息"""
# 获取 webhook URL
webhook_url = os.getenv("FEISHU_WEBHOOK_URL", "默认地址")

# 构建富文本消息
payload = {
    "msg_type": "post",
    "content": {
        "post": {
            "zh_cn": {
                "title": "📊 市场洞察报告 - 每日汇总",
                "content": [[{"tag": "text", "text": state.report_content}]]
            }
        }
    }
}

# 发送请求
response = requests.post(webhook_url, json=payload)
result = response.json()

return FeishuSendOutput(send_status="success" if result.get("code") == 0 else "failed")

5. 工作流编排

src/graphs/graph.py

builder = StateGraph(GlobalState, input_schema=GraphInput, output_schema=GraphOutput)

添加节点

builder.add_node("market_search", market_search_node)
builder.add_node("report_generation", report_generation_node,
metadata={"type": "agent", "llm_cfg": "config/report_generation_cfg.json"})
builder.add_node("feishu_send", feishu_send_node)

构建工作流

builder.set_entry_point("market_search")
builder.add_edge("market_search", "report_generation")
builder.add_edge("report_generation", "feishu_send")
builder.add_edge("feishu_send", END)

main_graph = builder.compile()
⏰ 定时任务详解
使用 APScheduler 实现

scripts/scheduler.py

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger

scheduler = BlockingScheduler()

添加定时任务：每天早上8点

scheduler.add_job(
func=send_market_report,
trigger=CronTrigger(hour=8, minute=0),
id='market_report_job',
name='每日市场洞察报告'
)

启动调度器

scheduler.start()
CronTrigger 配置示例

每天早上8点

CronTrigger(hour=8, minute=0)

每天早上9点半

CronTrigger(hour=9, minute=30)

每周一早上8点

CronTrigger(day_of_week=0, hour=8, minute=0) # 0=周一

每周一、三、五早上8点

CronTrigger(day_of_week='0,2,4', hour=8, minute=0)

每月1号早上8点

CronTrigger(day=1, hour=8, minute=0)

每天早上8点和下午2点

CronTrigger(hour='8,14', minute=0)
后台运行（生产环境推荐）
使用 nohup：

nohup python scripts/scheduler.py > scheduler.log 2>&1 &
tail -f scheduler.log
使用 systemd 服务：

创建 /etc/systemd/system/market-report.service：

[Unit]
Description=Market Insight Report Scheduler
After=network.target

[Service]
Type=simple
User=your_username
WorkingDirectory=/path/to/project
ExecStart=/usr/bin/python3 scripts/scheduler.py
Restart=always

[Install]
WantedBy=multi-user.target
启动服务：

sudo systemctl daemon-reload
sudo systemctl enable market-report
sudo systemctl start market-report
sudo systemctl status market-report
📊 报告示例
系统生成的报告结构如下：

工程招投标AI应用市场洞察报告

一、报告概述（今日市场焦点）

当前工程建设招投标领域正加速向数智化转型，AI技术已成为构建全链智慧交易体系的核心驱动力...

二、关键动态

行业标准性报告发布，AI应用场景明确
2025年5月，中国日报网发布《工程建设项目招标投标AI应用报告》...
行业龙头企业清标AI功能全面落地，效率提升显著
某行业龙头企业清标+AI方案实现智能推荐、智慧分析、智能决策、智能报告四大核心突破...

三、行业趋势分析

全链条智慧交易体系加速构建
从招标阶段的文件编制、合规审查，到投标阶段的标书检查、清标分析...
大模型技术驱动场景化应用深化
以某行业龙头企业AECGPT大模型为代表的行业大模型，正推动AI应用从单一功能向场景化解决方案升级...

四、竞争格局洞察

头部企业技术壁垒凸显
某行业龙头企业凭借"AI+数据"的核心战略，沉淀海量建筑行业数据...
市场集中度将逐步提升
招投标AI应用对技术研发能力、行业数据积累及场景落地经验要求较高...

五、商业建议

聚焦政策落地节点，加快场景化产品布局
企业应紧跟国家发改委提出的2026-2027年AI落地时间表...
强化"AI+数据"能力建设
借鉴某行业龙头企业的核心战略，加大行业数据沉淀与AI模型训练投入...
🎯 最佳实践
关键词优化
根据业务需求调整搜索关键词：

default_keywords = [
"行业龙头企业", # 行业头部企业
"智能标书审查", # 核心业务场景
"AI编标", # 新兴技术应用
"建筑工程", # 行业分类
"造价软件", # 产品类别
# 添加更多关键词...
]
2. 提示词优化
在 config/report_generation_cfg.json 中优化系统提示词：

{
"sp": "你是专注于建筑工程和造价软件领域的资深市场分析师，具有10年以上的行业经验。你的报告需要：\n\n1. 数据驱动：基于搜索结果提供具体数据和案例\n2. 观点鲜明：明确表达对趋势的判断\n3. 建议实用：提供可执行的商业建议\n\n请保持客观中立，避免过度推测。"
}
3. 错误处理
代码已包含完善的异常处理：

try:
response = requests.post(webhook_url, json=payload)
result = response.json()

if result.get("code") == 0:
    return FeishuSendOutput(send_status="success")
else:
    error_msg = result.get("msg", "Unknown error")
    raise Exception(f"飞书发送失败: {error_msg}")

except Exception as e:
raise Exception(f"发送飞书消息异常: {str(e)}")
4. 日志监控
查看执行日志：

查看最近的执行

grep "任务完成" scheduler.log | tail -n 10

查看错误日志

grep "ERROR" scheduler.log

实时监控

tail -f scheduler.log
🔧 常见问题
Q1: 如何修改发送时间？
编辑 scripts/scheduler.py，修改 CronTrigger 参数：

修改为每天早上9点

CronTrigger(hour=9, minute=0)
Q2: 飞书消息发送失败？
检查以下几点：

Webhook URL 是否正确
网络连接是否正常
飞书机器人是否被移出群组
测试 webhook：

curl -X POST <webhook_url> -d '{"msg_type":"text","content":{"text":"测试"}}'
Q3: 如何增加搜索结果数量？
编辑 market_search_node.py：

response = client.web_search_with_summary(
query=query,
count=20 # 增加到20条
)
Q4: 如何发送到多个飞书群？
在 feishu_send_node.py 中添加多个 webhook：

webhook_urls = [
os.getenv("FEISHU_WEBHOOK_URL_1"),
os.getenv("FEISHU_WEBHOOK_URL_2"),
]

for url in webhook_urls:
requests.post(url, json=payload)
📈 扩展方向

支持多渠道推送
除了飞书，还可以支持：

微信企业号机器人
钉钉机器人
Slack Webhook
邮件发送
2. 数据可视化
生成图表展示：

关键词趋势图
搜索结果数量统计
发送成功率监控
3. 智能摘要提取
添加摘要高亮、关键词提取、情感分析等。

历史数据存储
使用数据库存储历史报告，支持：

历史报告查询
趋势对比分析
报告导出
🎉 总结
通过这个项目，我们实现了一个完整的市场洞察报告自动化系统：

✅ 自动化：从信息搜集到报告生成再到消息推送，全流程自动化
✅ 智能化：AI 驱动的报告生成，内容专业且结构化
✅ 定时化：每天固定时间推送，无需人工干预
✅ 可扩展：基于 LangGraph，易于添加新节点和功能

这个系统不仅适用于建筑行业，也可以轻松适配到其他领域，只需修改搜索关键词和提示词即可。

📚 相关资源
项目代码：[GitHub 仓库链接]
LangGraph 官方文档：https://langchain-ai.github.io/langgraph/
APScheduler 文档：https://apscheduler.readthedocs.io/
飞书开放平台：https://open.feishu.cn/
项目已开源，欢迎 Star 和 Fork！如果有任何问题，欢迎提 Issue！ 🚀

希望这篇博文对你有帮助！如果你有任何疑问或建议，欢迎在评论区交流！💬

🚀 打造智能市场洞察报告系统：用 LangGraph + AI 实现每日自动化推送

修改为你的飞书 webhook 地址

============================================================
🧪 测试：立即执行市场洞察报告
⏰ 时间: 2026-02-12 18:54:02

src/graphs/state.py

src/graphs/nodes/market_search_node.py

src/graphs/nodes/report_generation_node.py

src/graphs/nodes/feishu_send_node.py

src/graphs/graph.py

添加节点

构建工作流

scripts/scheduler.py

添加定时任务：每天早上8点

启动调度器

每天早上8点

每天早上9点半

每周一早上8点

每周一、三、五早上8点

每月1号早上8点

每天早上8点和下午2点

工程招投标AI应用市场洞察报告

一、报告概述（今日市场焦点）

二、关键动态

三、行业趋势分析

四、竞争格局洞察

五、商业建议

查看最近的执行

查看错误日志

实时监控

修改为每天早上9点

灰狗的糖果屋

分享文章

文章目录

修改为你的飞书 webhook 地址

============================================================ 🧪 测试：立即执行市场洞察报告 ⏰ 时间: 2026-02-12 18:54:02

src/graphs/state.py

src/graphs/nodes/market_search_node.py

src/graphs/nodes/report_generation_node.py

src/graphs/nodes/feishu_send_node.py

src/graphs/graph.py

添加节点

构建工作流

scripts/scheduler.py

添加定时任务：每天早上8点

启动调度器

每天早上8点

每天早上9点半

每周一早上8点

每周一、三、五早上8点

每月1号早上8点

每天早上8点和下午2点

工程招投标AI应用市场洞察报告

一、报告概述（今日市场焦点）

二、关键动态

三、行业趋势分析

四、竞争格局洞察

五、商业建议

查看最近的执行

查看错误日志

实时监控

修改为每天早上9点

分享文章

文章目录

============================================================
🧪 测试：立即执行市场洞察报告
⏰ 时间: 2026-02-12 18:54:02