多智能体系统实战：用LangChain+MCP构建一个自动化周报生成系统

为什么周报需要多智能体？

周报是每个程序员的"周常痛点"——不是写不出来，而是收集数据太麻烦：

• 从Jira拉取本周完成的任务
• 从Git统计代码提交量
• 从飞书/钉钉获取会议记录
• 从监控系统查看线上告警
• 最后还要"润色"成领导爱看的格式

如果一个Agent只能做一件事，那就让多个Agent分工协作——这就是多智能体系统的价值。

---

系统架构

┌─────────────────────────────────────────────────────────────┐
│                    协调Agent (Orchestrator)                   │
│                  LangChain AgentExecutor                    │
│                  负责任务分解、Agent调度、结果汇总              │
└──────────┬──────────┬──────────┬──────────┬────────────────┘
           │          │          │          │
     ┌─────▼────┐ ┌───▼────┐ ┌──▼───┐ ┌───▼────┐
     │ 数据采集  │ │ 代码分析│ │ 会议摘要│ │ 报告生成│
     │  Agent   │ │  Agent  │ │ Agent  │ │  Agent  │
     └─────┬────┘ └───┬────┘ └──┬───┘ └───┬────┘
           │          │          │          │
     ┌─────▼──────────▼──────────▼──────────▼────────────────┐
     │                  MCP Server 层                         │
     │  [Jira MCP] [Git MCP] [飞书 MCP] [监控 MCP] [邮件 MCP] │
     └───────────────────────────────────────────────────────┘

Agent分工

Agent	职责	使用的MCP Server
数据采集Agent	从Jira/Git拉取原始数据	jira-mcp, git-mcp
代码分析Agent	分析代码变更、PR统计	git-mcp
会议摘要Agent	提取会议纪要关键信息	feishu-mcp
报告生成Agent	汇总数据，生成周报	email-mcp

---

第一步：定义Agent

1.1 数据采集Agent

from langchain_openai import ChatOpenAI
from langchain_mcp import MCPToolkit

jira_toolkit = MCPToolkit(url="http://localhost:8081/mcp/sse")
git_toolkit = MCPToolkit(url="http://localhost:8082/mcp/sse")

data_collector_agent = {
    "name": "data_collector",
    "role": "数据采集专家",
    "goal": "从Jira和Git中收集本周的工作数据",
    "backstory": "你擅长从各种系统中提取结构化数据，确保数据完整准确",
    "tools": jira_toolkit.get_tools() + git_toolkit.get_tools(),
    "llm": ChatOpenAI(model="gpt-4o", temperature=0)
}

1.2 代码分析Agent

code_analyst_agent = {
    "name": "code_analyst",
    "role": "代码分析专家",
    "goal": "分析本周的代码变更，提取关键信息",
    "backstory": "你能从Git提交记录中识别重要的技术变更和风险点",
    "tools": git_toolkit.get_tools(),
    "llm": ChatOpenAI(model="gpt-4o", temperature=0)
}

1.3 会议摘要Agent

from langchain_mcp import MCPToolkit

feishu_toolkit = MCPToolkit(url="http://localhost:8083/mcp/sse")

meeting_summarizer_agent = {
    "name": "meeting_summarizer",
    "role": "会议摘要专家",
    "goal": "从飞书会议记录中提取本周关键决策和待办事项",
    "backstory": "你擅长从冗长的会议记录中提炼核心信息",
    "tools": feishu_toolkit.get_tools(),
    "llm": ChatOpenAI(model="gpt-4o", temperature=0)
}

1.4 报告生成Agent

email_toolkit = MCPToolkit(url="http://localhost:8084/mcp/sse")

report_generator_agent = {
    "name": "report_generator",
    "role": "报告生成专家",
    "goal": "将所有数据汇总，生成结构清晰的周报",
    "backstory": "你擅长将技术数据转化为管理层易读的报告格式",
    "tools": email_toolkit.get_tools(),
    "llm": ChatOpenAI(model="gpt-4o", temperature=0.3)
}

---

第二步：实现MCP Server

2.1 Jira MCP Server

from mcp.server import Server, Tool
from jira import JIRA

server = Server("jira-mcp-server")

jira_client = JIRA(
    server="https://yourcompany.atlassian.net",
    basic_auth=("email", "api_token")
)

@server.tool(name="get_completed_issues", description="获取指定时间范围内完成的Jira任务")
async def get_completed_issues(start_date: str, end_date: str, assignee: str = None):
    jql = f'updated >= "{start_date}" AND updated <= "{end_date}" AND status = Done'
    if assignee:
        jql += f' AND assignee = "{assignee}"'

    issues = jira_client.search_issues(jql, maxResults=50)

    results = []
    for issue in issues:
        results.append({
            "key": issue.key,
            "summary": issue.fields.summary,
            "status": issue.fields.status.name,
            "assignee": issue.fields.assignee.displayName if issue.fields.assignee else "Unassigned",
            "priority": issue.fields.priority.name,
            "updated": issue.fields.updated
        })

    return {"issues": results, "total": len(results)}

@server.tool(name="get_sprint_progress", description="获取当前Sprint的进度信息")
async def get_sprint_progress(board_id: int):
    sprints = jira_client.sprints(board_id, state="active")
    if not sprints:
        return {"error": "No active sprint found"}

    sprint = sprints[0]
    issues = jira_client.search_issues(f'sprint = {sprint.id}')

    done = sum(1 for i in issues if i.fields.status.name == "Done")
    in_progress = sum(1 for i in issues if i.fields.status.name == "In Progress")
    todo = len(issues) - done - in_progress

    return {
        "sprint_name": sprint.name,
        "total": len(issues),
        "done": done,
        "in_progress": in_progress,
        "todo": todo,
        "progress": f"{done}/{len(issues)} ({done/len(issues)*100:.1f}%)"
    }

2.2 Git MCP Server

from git import Repo
from datetime import datetime, timedelta

server = Server("git-mcp-server")

@server.tool(name="get_commit_stats", description="获取指定时间范围内的Git提交统计")
async def get_commit_stats(repo_path: str, author: str = None, days: int = 7):
    repo = Repo(repo_path)
    since = datetime.now() - timedelta(days=days)

    commits = list(repo.iter_commits(since=since))

    if author:
        commits = [c for c in commits if author in c.author.name]

    stats = {
        "total_commits": len(commits),
        "files_changed": set(),
        "insertions": 0,
        "deletions": 0,
        "daily_breakdown": {}
    }

    for commit in commits:
        for diff in commit.diff(commit.parents[0] if commit.parents else None):
            stats["files_changed"].add(diff.a_path or diff.b_path)
        stats["insertions"] += commit.stats.total["insertions"]
        stats["deletions"] += commit.stats.total["deletions"]

        day = commit.committed_datetime.strftime("%Y-%m-%d")
        stats["daily_breakdown"][day] = stats["daily_breakdown"].get(day, 0) + 1

    stats["files_changed"] = list(stats["files_changed"])
    return stats

@server.tool(name="get_pr_summary", description="获取本周PR的汇总信息")
async def get_pr_summary(repo_path: str, days: int = 7):
    # 分析PR的merge记录
    repo = Repo(repo_path)
    since = datetime.now() - timedelta(days=days)

    merged_prs = []
    for commit in repo.iter_commits(since=since, merges=True):
        merged_prs.append({
            "title": commit.message.split("\n")[0],
            "author": commit.author.name,
            "date": commit.committed_datetime.isoformat()
        })

    return {"merged_prs": merged_prs, "total": len(merged_prs)}

---

第三步：编排多智能体工作流

3.1 使用LangGraph编排

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Dict, Any

class WeeklyReportState(TypedDict):
    user: str
    week_start: str
    week_end: str
    jira_data: Dict[str, Any]
    git_data: Dict[str, Any]
    meeting_data: Dict[str, Any]
    analysis: str
    report: str

def collect_jira_data(state: WeeklyReportState) -> WeeklyReportState:
    result = data_collector_agent.invoke({
        "input": f"获取{state['user']}在{state['week_start']}到{state['week_end']}之间完成的Jira任务"
    })
    state["jira_data"] = result
    return state

def collect_git_data(state: WeeklyReportState) -> WeeklyReportState:
    result = code_analyst_agent.invoke({
        "input": f"分析{state['user']}在{state['week_start']}到{state['week_end']}之间的代码提交"
    })
    state["git_data"] = result
    return state

def collect_meeting_data(state: WeeklyReportState) -> WeeklyReportState:
    result = meeting_summarizer_agent.invoke({
        "input": f"提取{state['week_start']}到{state['week_end']}之间的会议关键信息"
    })
    state["meeting_data"] = result
    return state

def generate_report(state: WeeklyReportState) -> WeeklyReportState:
    prompt = f"""
    根据以下数据生成周报：

    ## Jira任务完成情况
    {state['jira_data']}

    ## 代码提交统计
    {state['git_data']}

    ## 会议关键信息
    {state['meeting_data']}

    请生成结构清晰的周报，包含：
    1. 本周工作总结
    2. 关键成果
    3. 风险与问题
    4. 下周计划
    """
    result = report_generator_agent.invoke({"input": prompt})
    state["report"] = result
    return state

workflow = StateGraph(WeeklyReportState)

workflow.add_node("collect_jira", collect_jira_data)
workflow.add_node("collect_git", collect_git_data)
workflow.add_node("collect_meeting", collect_meeting_data)
workflow.add_node("generate_report", generate_report)

workflow.add_edge("collect_jira", "collect_git")
workflow.add_edge("collect_git", "collect_meeting")
workflow.add_edge("collect_meeting", "generate_report")
workflow.add_edge("generate_report", END)

workflow.set_entry_point("collect_jira")

app = workflow.compile()

3.2 并行采集优化

from langgraph.graph import StateGraph

# 数据采集可以并行执行
workflow = StateGraph(WeeklyReportState)

workflow.add_node("collect_jira", collect_jira_data)
workflow.add_node("collect_git", collect_git_data)
workflow.add_node("collect_meeting", collect_meeting_data)
workflow.add_node("merge_and_report", generate_report)

# 三个采集节点并行，全部完成后进入报告生成
workflow.add_edge("collect_jira", "merge_and_report")
workflow.add_edge("collect_git", "merge_and_report")
workflow.add_edge("collect_meeting", "merge_and_report")
workflow.add_edge("merge_and_report", END)

# 设置多个入口点实现并行
workflow.set_entry_point("collect_jira")  # 实际通过fan-out实现

---

第四步：执行与输出

result = app.invoke({
    "user": "zhangsan",
    "week_start": "2026-06-05",
    "week_end": "2026-06-11",
    "jira_data": {},
    "git_data": {},
    "meeting_data": {},
    "analysis": "",
    "report": ""
})

print(result["report"])

输出示例：

# 周报：2026年6月5日 - 6月11日

## 本周工作总结

### 已完成任务（5个）
- PROJ-123: 实现用户认证模块的OAuth2.0集成 ✅
- PROJ-124: 修复订单超时未关闭的Bug ✅
- PROJ-126: 优化数据库查询性能，P99延迟降低40% ✅
- PROJ-128: 完成支付网关对接文档 ✅
- PROJ-130: Code Review同事的PR（3个） ✅

### 代码统计
- 提交次数：23次
- 代码变更：+1,247 / -389 行
- 涉及文件：18个
- 合并PR：4个

### 关键会议
- 周二技术评审：确定微服务拆分方案，采用DDD领域驱动设计
- 周四Sprint回顾：当前Sprint完成率85%，剩余3个任务

## 风险与问题
- 订单超时Bug修复后需要观察线上表现，已添加监控告警
- 支付网关沙箱环境联调延迟，预计下周三完成

## 下周计划
- 完成支付网关生产环境部署
- 启动用户画像服务的重构
- 参与架构评审：缓存策略优化方案

---

生产级增强

定时任务调度

from apscheduler.schedulers.asyncio import AsyncIOScheduler

scheduler = AsyncIOScheduler()

@scheduler.scheduled_job('cron', day_of_week='fri', hour=17, minute=0)
async def weekly_report_job():
    users = get_all_team_members()
    for user in users:
        result = await app.ainvoke({
            "user": user,
            "week_start": get_week_start(),
            "week_end": get_week_end(),
        })
        await send_report_email(user, result["report"])

scheduler.start()

错误处理与重试

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
async def collect_jira_data_with_retry(state):
    try:
        return collect_jira_data(state)
    except JiraApiException as e:
        logger.error(f"Jira API调用失败: {e}")
        raise

---

踩坑记录

坑1：MCP Server连接超时

现象：Agent调用MCP Server时偶尔超时
原因：Jira API响应慢（P99 > 5s），MCP默认超时3s
解决：调大MCP Client超时时间

toolkit = MCPToolkit(
    url="http://localhost:8081/mcp/sse",
    timeout=30  # 增加到30秒
)

坑2：Agent幻觉——编造不存在的任务

现象：报告生成Agent有时会"编造"未在数据中出现的任务
原因：LLM的创造性倾向，temperature过高
解决：数据采集Agent用temperature=0，报告生成Agent的prompt中强调"仅基于提供的数据"

report_prompt = """
重要：仅基于以下实际数据生成报告，不要编造任何不存在的任务或数据。

## 实际数据
{jira_data}
{git_data}
{meeting_data}
"""

坑3：并发采集时MCP连接冲突

现象：并行采集时偶尔出现连接错误
原因：多个Agent同时使用同一个MCP Client
解决：每个Agent使用独立的MCP Client实例

# 每个Agent创建自己的MCP连接
def create_agent_with_mcp(mcp_url, agent_config):
    toolkit = MCPToolkit(url=mcp_url)  # 独立连接
    return Agent(tools=toolkit.get_tools(), **agent_config)

---

总结

多智能体系统让周报生成从"手动拼凑"变成"自动流水线"：

1. 分工明确：每个Agent只做一件事，做好一件事
2. MCP解耦：Agent不直接调用API，通过MCP Server间接访问，易于替换和测试
3. LangGraph编排：支持串行、并行、条件分支等复杂工作流
4. 生产可用：定时调度、错误重试、幻觉防控

多智能体不是"多个聊天机器人"，而是"一个软件系统"——每个Agent是一个微服务，MCP是它们之间的API网关。