08 - Agent与MCP集成¶

"""Function Calling工作原理——模型自主决定参数并调用外部工具"""

# 1. 定义工具（Tool/Function）——告诉模型有哪些能力可用
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_database",
            # description是模型选择工具的唯一依据，必须足够详细
            "description": "搜索产品数据库，返回匹配的产品信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词"  # 每个参数都需要清晰的description
                    },
                    "category": {
                        "type": "string",
                        # 使用enum限制取值范围，避免模型传入无效值
                        "enum": ["电子产品", "服饰", "食品"],
                        "description": "产品类别"
                    },
                    "max_results": {
                        "type": "integer",
                        "default": 5,           # 提供默认值，减少必填参数
                        "description": "返回结果数量"
                    }
                },
                "required": ["query"]  # 只有query是必填的，降低调用门槛
            }
        }
    }
]

# 2. 模型决定是否调用工具
# 输入: 用户消息 + tools定义 → 模型理解工具能力
# 输出: tool_calls（模型自主决定调用哪个工具、传什么参数）

# 3. 执行工具并返回结果（由应用程序执行，非模型执行）
# 4. 模型基于工具结果生成最终自然语言回答

"""Dify工具编排设计原则"""

TOOL_DESIGN_PRINCIPLES = {
    "单一职责": {
        "原则": "每个工具只做一件事",
        "正确": "search_products, get_product_detail, create_order",
        "错误": "search_and_order_products（功能太多）"
    },
    "清晰描述": {
        "原则": "description要足够详细，帮助模型正确选择工具",
        "正确": "搜索产品数据库，支持按名称、类别、价格范围筛选",
        "错误": "搜索功能"
    },
    "参数设计": {
        "原则": "参数要有明确的类型、默认值和说明",
        "建议": [
            "必需参数尽量少（1-3个）",
            "使用enum限制有效值范围",
            "为可选参数提供合理默认值"
        ]
    },
    "错误处理": {
        "原则": "工具返回清晰的错误信息",
        "包含": "错误码 + 错误描述 + 建议操作",
        "示例": '{"error": "product_not_found", '
                '"message": "未找到匹配的产品", '
                '"suggestion": "请尝试更宽泛的搜索词"}'
    },
    "安全考虑": {
        "只读优先": "优先提供查询类工具，写入类工具需要确认",
        "权限控制": "敏感操作需要用户二次确认",
        "数据脱敏": "返回结果中脱敏个人信息"
    }
}

"""MCP协议核心概念"""

MCP_CONCEPTS = {
    "Resources（资源）": {
        "定义": "MCP Server暴露的数据或内容",
        "URI格式": "file:///path/to/file, db://table/records",
        "特点": "只读，由应用程序控制何时获取",
        "示例": ["文件内容", "数据库记录", "API响应", "截屏"]
    },
    "Tools（工具）": {
        "定义": "MCP Server暴露的可执行操作",
        "特点": "由模型决定何时调用（类似Function Calling）",
        "示例": ["执行SQL查询", "发送邮件", "创建文件", "调用API"],
        "安全": "需要用户确认执行（human-in-the-loop）"
    },
    "Prompts（提示）": {
        "定义": "预定义的提示模板，可复用上下文",
        "特点": "由用户显式选择使用",
        "示例": ["代码审查模板", "数据分析模板", "报告生成模板"]
    },
    "Sampling（采样）": {
        "定义": "Server可以请求Client执行LLM推理",
        "特点": "实现Agent嵌套（Server端也能调用LLM）",
        "场景": "复杂工具需要LLM辅助决策时使用"
    }
}

"""使用Python开发自定义MCP Server——将企业内部工具暴露给LLM Agent"""

# pip install mcp

from mcp.server import Server
from mcp.types import Tool, TextContent
from mcp.server.stdio import stdio_server
import json

# 创建MCP Server实例，名称用于Client识别和日志
server = Server("company-tools")

# 使用@server.tool()装饰器注册工具——函数签名自动转换为工具参数Schema
@server.tool()
async def query_sales_data(  # async def定义异步函数；用await调用
    start_date: str,
    end_date: str,
    product_category: str = "all"
) -> list[TextContent]:
    """
    查询销售数据。

    Args:
        start_date: 开始日期（YYYY-MM-DD）
        end_date: 结束日期（YYYY-MM-DD）
        product_category: 产品类别，默认all

    Returns:
        销售数据摘要
    """
    # 实际查询数据库（使用参数化查询防止SQL注入）
    results = await db.query(  # await等待异步操作完成
        "SELECT * FROM sales WHERE date BETWEEN %s AND %s",
        (start_date, end_date)
    )

    # 按类别过滤（在应用层而非SQL层，保持灵活性）
    if product_category != "all":
        results = [r for r in results if r["category"] == product_category]

    # 汇总统计信息，返回结构化摘要而非原始数据
    summary = {
        "total_revenue": sum(r["amount"] for r in results),
        "total_orders": len(results),
        "top_products": get_top_products(results, n=5)
    }

    # MCP工具必须返回TextContent列表
    return [TextContent(
        type="text",
        text=json.dumps(summary, ensure_ascii=False, indent=2)  # json.dumps将Python对象转为JSON字符串
    )]

@server.tool()
async def create_report(
    title: str,
    content: str,
    format: str = "markdown"
) -> list[TextContent]:
    """
    生成并保存报告文件。

    Args:
        title: 报告标题
        content: 报告内容
        format: 格式（markdown/html/pdf）
    """
    filepath = f"/reports/{title}.{format}"
    await save_report(filepath, content, format)

    return [TextContent(
        type="text",
        text=f"报告已保存: {filepath}"
    )]

# 定义资源——使用URI模板暴露可查询的数据资源
@server.resource("sales://monthly/{year}/{month}")
async def get_monthly_sales(year: str, month: str) -> str:
    """获取月度销售报告——只读Resource，由应用程序控制访问时机"""
    data = await db.get_monthly_report(int(year), int(month))
    return json.dumps(data, ensure_ascii=False)

# 启动MCP Server——通过stdio与Client通信（适用于本地部署）
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())  # asyncio.run()启动异步事件循环

"""设计一个迭代式研究Agent工作流"""

RESEARCH_AGENT_WORKFLOW = {
    "节点1 - 研究规划": {
        "类型": "LLM",
        "功能": "分析研究主题，生成3-5个子问题",
        "输出": "研究计划（子问题列表）"
    },
    "节点2 - 迭代研究(循环)": {
        "类型": "Agent(循环节点)",
        "功能": "对每个子问题：搜索→阅读→总结",
        "工具": ["web_search", "web_scraper", "knowledge_base"],
        "循环条件": "遍历所有子问题",
        "每轮输出": "该子问题的研究结果"
    },
    "节点3 - 信息整合": {
        "类型": "LLM",
        "功能": "整合所有子问题的研究结果",
        "输出": "完整的研究报告草稿"
    },
    "节点4 - 质量评估": {
        "类型": "LLM",
        "功能": "评估报告质量，识别缺失信息",
        "输出": "质量分数 + 改进建议"
    },
    "节点5 - 条件分支": {
        "类型": "条件",
        "条件": "质量分数 >= 8 → 输出",
        "否则": "回到节点2补充研究（最多2次）"
    },
    "节点6 - 格式化输出": {
        "类型": "模板",
        "功能": "将报告格式化为最终格式"
    }
}

"""Agent调试检查清单"""

AGENT_DEBUG_CHECKLIST = {
    "工具调用问题": {
        "不调用工具": [
            "检查System Prompt是否明确说明何时使用工具",
            "检查工具description是否足够清晰",
            "检查模型是否支持Function Calling",
            "尝试在Prompt中添加Few-shot示例"
        ],
        "调用错误工具": [
            "检查工具命名和描述是否有歧义",
            "减少工具数量（<10个为佳）",
            "在description中添加使用/不使用的场景说明"
        ],
        "参数错误": [
            "检查参数的type和description是否准确",
            "添加参数示例值",
            "使用enum限制有效值"
        ]
    },
    "推理问题": {
        "无限循环": [
            "设置max_iterations（建议5-10）",
            "检查工具返回是否给出了足够信息让Agent停止",
            "添加明确的终止条件在System Prompt中"
        ],
        "推理链质量差": [
            "切换到更强的模型（GPT-4o）",
            "降低temperature（0.0-0.1）",
            "在System Prompt中添加推理步骤指引"
        ]
    },
    "性能问题": {
        "响应太慢": [
            "减少不必要的工具调用",
            "并行执行独立的工具调用",
            "缓存频繁查询的结果",
            "使用更快的模型（GPT-4o-mini）处理简单步骤"
        ]
    }
}

"""Agent评估脚本示例——系统化衡量Agent的任务完成率、工具准确率和性能"""

import json
from typing import List, Dict

class AgentEvaluator:
    """系统化评估Agent应用——基于预定义测试用例自动评分"""

    def __init__(self, test_cases: List[Dict]):
        self.test_cases = test_cases  # 预定义的测试用例集
        self.results = []              # 评估结果存储

    def evaluate(self, agent_fn):
        """
        运行评估：对每个测试用例调用Agent，收集多维度指标

        Args:
            agent_fn: Agent调用函数，输入问题返回结果字典
        """
        for case in self.test_cases:
            # 调用Agent处理测试输入
            result = agent_fn(case["input"])

            # 收集评估维度：输出内容、工具使用、步骤数、延迟、Token消耗
            evaluation = {
                "input": case["input"],
                "expected": case["expected_output"],
                "actual": result["output"],
                "tools_used": result["tools_called"],        # 实际调用的工具列表
                "expected_tools": case.get("expected_tools", []),  # 期望调用的工具
                "steps": result["num_steps"],               # 推理步骤数
                "latency": result["latency_ms"],             # 端到端延迟(ms)
                "tokens": result["total_tokens"],            # Token消耗量
            }

            # 判断任务是否成功（基于输出匹配）
            evaluation["success"] = self._check_success(
                evaluation["actual"],
                evaluation["expected"]
            )

            # 检查工具调用是否符合预期（集合比较，不关心顺序）
            evaluation["correct_tools"] = set(
                evaluation["tools_used"]
            ) == set(evaluation["expected_tools"])

            self.results.append(evaluation)

        return self._summary()

    def _summary(self):
        """汇总评估结果——计算各指标的均值"""
        n = len(self.results)
        return {
            "success_rate": sum(r["success"] for r in self.results) / n,      # 任务成功率
            "tool_accuracy": sum(r["correct_tools"] for r in self.results) / n, # 工具调用准确率
            "avg_steps": sum(r["steps"] for r in self.results) / n,            # 平均推理步骤
            "avg_latency_ms": sum(r["latency"] for r in self.results) / n,      # 平均延迟
            "avg_tokens": sum(r["tokens"] for r in self.results) / n,           # 平均Token消耗
        }

# 测试用例示例——每个用例包含输入、期望输出和期望调用的工具
test_cases = [
    {
        "input": "帮我查一下订单 ORD-12345 的物流状态",
        "expected_output": "包含物流状态信息",
        # Agent应先查订单，再查物流，需要两个工具协作
        "expected_tools": ["query_order", "track_shipment"]
    },
    {
        "input": "我要退货，订单号ORD-67890",
        "expected_output": "退货申请已提交",
        # Agent应先查询订单确认存在，再创建退货单
        "expected_tools": ["query_order", "create_return"]
    }
]

"""Agent应用成本控制策略——在保证质量的前提下最小化Token消耗"""

COST_CONTROL = {
    "模型选择": {
        "策略": "简单任务用小模型，复杂任务用大模型",
        # 分级策略：约80%的请求可用小模型处理，节省大量费用
        "实现": {
            "意图分类": "GPT-4o-mini（几乎不用大模型）",
            "Agent推理": "GPT-4o（需要强推理能力）",
            "简单生成": "GPT-4o-mini",
        },
        "节省": "~60% Token费用"
    },
    "工具调用优化": {
        "缓存": "常用工具结果缓存（TTL=5min）",       # 相同参数避免重复调用
        "批量": "支持批量查询的工具一次查多个",     # 减少工具调用次数
        "预检": "先检查参数合法性再调用API"       # 避免无效调用浪费Token
    },
    "对话管理": {
        "上下文窗口": "只保留最近10-20轮对话",       # 控制输入Token数
        "摘要压缩": "超过窗口的历史用摘要代替",     # 保留关键信息，减少冗余
        "Session超时": "30分钟无活动自动关闭"       # 及时释放资源
    },
    "限制策略": {
        "每用户每日": "最多100次Agent调用",           # 防止滥用
        "单次最大Token": "4096 output tokens",          # 控制单次成本
        "最大迭代": "10步（防无限循环）"                  # 安全上限
    }
}

"""研究助手评估用例"""

RESEARCH_ASSISTANT_TESTS = [
    {
        "input": "调研2024年大模型推理优化的最新技术",
        "expected_coverage": [
            "Speculative Decoding", "KV Cache优化",
            "量化技术", "vLLM/TGI"
        ],
        "quality_criteria": {
            "had_sources": True,
            "min_sources": 3,
            "structured_output": True,
            "recency": "2024"
        }
    },
    {
        "input": "对比分析LangChain和LlamaIndex的优缺点",
        "expected_coverage": [
            "架构对比", "RAG能力", "Agent能力",
            "社区生态", "使用场景推荐"
        ],
        "quality_criteria": {
            "balanced_comparison": True,
            "has_recommendation": True
        }
    }
]