AutoGen — 多智能体协作框架

简介

AutoGen 是微软开源的多智能体对话框架，支持多个 AI Agent 相互协作完成复杂任务，适合需要多角色分工的金融分析场景。

pip install pyautogen

双 Agent 对话

import autogen

config_list = [{
    "model": "qwen-plus",
    "api_key": "sk-xxx",
    "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1"
}]

llm_config = {"config_list": config_list, "temperature": 0.7}

# 助手 Agent
assistant = autogen.AssistantAgent(
    name="金融分析师",
    llm_config=llm_config,
    system_message="你是专业的金融分析师，擅长分析企业财务数据和市场风险。"
)

# 用户代理（可执行代码）
user_proxy = autogen.UserProxyAgent(
    name="用户",
    human_input_mode="NEVER",  # 自动运行，不需要人工输入
    max_consecutive_auto_reply=5,
    code_execution_config={"work_dir": "workspace", "use_docker": False}
)

# 启动对话
user_proxy.initiate_chat(
    assistant,
    message="分析招商银行2024年的财务状况，重点关注不良贷款率和资本充足率"
)

多 Agent 协作（GroupChat）

# 金融分析团队
risk_analyst = autogen.AssistantAgent(
    name="风险分析师",
    llm_config=llm_config,
    system_message="你专注于识别和评估金融风险，包括信用风险、市场风险和操作风险。"
)

compliance_officer = autogen.AssistantAgent(
    name="合规官",
    llm_config=llm_config,
    system_message="你负责确保所有分析和建议符合监管要求和内部合规政策。"
)

investment_advisor = autogen.AssistantAgent(
    name="投资顾问",
    llm_config=llm_config,
    system_message="你基于风险分析和合规检查，提供具体的投资建议。"
)

user_proxy = autogen.UserProxyAgent(
    name="项目经理",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=1
)

# 创建群聊
groupchat = autogen.GroupChat(
    agents=[user_proxy, risk_analyst, compliance_officer, investment_advisor],
    messages=[],
    max_round=10,
    speaker_selection_method="auto"  # 自动选择发言者
)

manager = autogen.GroupChatManager(
    groupchat=groupchat,
    llm_config=llm_config
)

# 启动多 Agent 协作
user_proxy.initiate_chat(
    manager,
    message="评估某科技公司的贷款申请：成立3年，营收500万，负债率65%，主营AI软件"
)

带工具的 Agent

from autogen import AssistantAgent, UserProxyAgent, register_function

def get_company_credit(company_name: str) -> dict:
    """查询企业信用信息"""
    return {
        "company": company_name,
        "credit_score": 720,
        "blacklist": False,
        "legal_cases": 0
    }

def calculate_risk_score(
    revenue: float,
    debt_ratio: float,
    years_operating: int
) -> float:
    """计算风险评分"""
    score = 100
    if debt_ratio > 0.7: score -= 30
    if years_operating < 3: score -= 20
    if revenue < 1000000: score -= 10
    return max(0, score)

# 注册工具
register_function(
    get_company_credit,
    caller=assistant,
    executor=user_proxy,
    description="查询企业信用信息"
)

register_function(
    calculate_risk_score,
    caller=assistant,
    executor=user_proxy,
    description="计算贷款风险评分"
)