from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Literal

class AgentState(TypedDict):
    input: str
    risk_score: float
    approval_status: Literal["pending", "approved", "rejected"]
    audit_trail: list

def risk_assessment(state: AgentState) -> AgentState:
    # 调用风控模型
    state["risk_score"] = calculate_risk(state["input"])
    state["audit_trail"].append(f"risk_score={state['risk_score']}")
    return state

def human_approval(state: AgentState) -> AgentState:
    # 通过Webhook通知人工审批
    send_approval_request(state)
    return state

def auto_approve(state: AgentState) -> AgentState:
    state["approval_status"] = "approved"
    return state

def route_by_risk(state: AgentState) -> Literal["human_approval", "auto_approve"]:
    return "human_approval" if state["risk_score"] > 0.7 else "auto_approve"

# 构建图
graph = StateGraph(AgentState)
graph.add_node("risk_assessment", risk_assessment)
graph.add_node("human_approval", human_approval)
graph.add_node("auto_approve", auto_approve)

graph.set_entry_point("risk_assessment")
graph.add_conditional_edges(
    "risk_assessment",
    route_by_risk,
    {"human_approval": "human_approval", "auto_approve": "auto_approve"}
)
graph.add_edge("human_approval", END)
graph.add_edge("auto_approve", END)

app = graph.compile()

from autogen import ConversableAgent, GroupChat, GroupChatManager

# 定义三个具有不同"人格"的Agent
researcher = ConversableAgent(
    name="researcher",
    system_message="你是顶级量化研究员，擅长数据挖掘。",
    llm_config={"config_list": [{"model": "gpt-4o", "api_key": "..."}]},
)

critic = ConversableAgent(
    name="critic",
    system_message="你是严苛的审计员，专门找逻辑漏洞。",
    llm_config={"config_list": [{"model": "gpt-4o", "api_key": "..."}]},
)

synthesizer = ConversableAgent(
    name="synthesizer",
    system_message="你负责汇总观点，输出最终报告。",
    llm_config={"config_list": [{"model": "gpt-4o", "api_key": "..."}]},
)

group_chat = GroupChat(
    agents=[researcher, critic, synthesizer],
    messages=[],
    max_round=12,
)

manager = GroupChatManager(
    groupchat=group_chat,
    llm_config={"config_list": [{"model": "gpt-4o", "api_key": "..."}]},
)

# 启动对话
result = manager.initiate_chat(
    researcher,
    message="分析2026年Q1全球AI芯片市场格局，并给出投资建议。",
)

from crewai import Agent, Task, Crew, Process

# 定义角色
content_writer = Agent(
    role="资深内容策略师",
    goal="产出SEO友好的深度文章",
    backstory="你拥有10年B2B内容营销经验",
    tools=[search_tool, seo_tool],
    allow_delegation=True,
)

seo_analyst = Agent(
    role="SEO数据分析师",
    goal="优化关键词布局",
    backstory="你精通Google Analytics与Search Console",
    tools=[analytics_tool],
)

# 定义任务（注意：任务是静态的，无法动态路由）
write_task = Task(
    description="撰写关于AI Agent框架对比的5000字长文",
    agent=content_writer,
    expected_output="Markdown格式文章",
)

analyze_task = Task(
    description="对文章进行SEO关键词密度分析",
    agent=seo_analyst,
    expected_output="优化建议列表",
)

crew = Crew(
    agents=[content_writer, seo_analyst],
    tasks=[write_task, analyze_task],
    process=Process.sequential,  # 仅支持顺序或层级
    verbose=True,
)

result = crew.kickoff()

# Dify DSL (部分)
app:
  mode: workflow
  workflow:
    nodes:
      - id: "start"
        type: "start"
        data:
          inputs:
            - variable: "query"
              value: "用户问题"
      - id: "rag_retrieval"
        type: "knowledge-retrieval"
        data:
          dataset_id: "ds_xxx"
          top_k: 5
      - id: "llm_chain"
        type: "llm"
        data:
          model: "gpt-4o"
          prompt_template: |
            基于以下上下文回答：{{#context#}}
            问题：{{#query#}}
      - id: "end"
        type: "end"

from langgraph.checkpoint.postgres import PostgresSaver

# 使用PostgreSQL保存检查点
with PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost:5432/agent_db"
) as checkpointer:
    graph = StateGraph(AgentState)
    # ... 构建图 ...
    app = graph.compile(checkpointer=checkpointer)
    
    # 支持时间旅行
    config = {"configurable": {"thread_id": "user_123"}}
    app.invoke({"input": "查询用户信用额度"}, config)
    
    # 回滚到历史状态
    state_history = app.get_state_history(config)
    last_state = next(state_history)
    app.invoke(last_state.values, {"configurable": {"thread_id": "user_123", "checkpoint_id": last_state.config["configurable"]["checkpoint_id"]}})

{
  "name": "get_stock_price",
  "description": "获取实时股票价格",
  "inputSchema": {
    "type": "object",
    "properties": {
      "symbol": {"type": "string", "description": "股票代码"}
    },
    "required": ["symbol"]
  }
}

from langchain_mcp_adapters import load_mcp_tools
from mcp import ClientSession, StdioServerParameters

# 启动MCP服务器
server_params = StdioServerParameters(
    command="python",
    args=["mcp_server.py"],
)

async with ClientSession(server_params) as session:
    tools = await load_mcp_tools(session)
    
    def mcp_tool_node(state):
        # 在图中调用MCP工具
        result = tools["get_stock_price"].invoke({"symbol": "AAPL"})
        return {"stock_price": result}

# requirements.txt
# langgraph==0.8.4
# langchain-openai==0.3.2
# psycopg2-binary==2.9.10
# redis==5.2.1

import asyncio
import json
import uuid
from typing import TypedDict, Literal, Annotated
from datetime import datetime

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
import redis

# ========== 状态定义 ==========
class LoanState(TypedDict):
    application_id: str
    applicant_data: dict
    risk_score: float
    risk_factors: list[str]
    approval_status: Literal["pending", "auto_approved", "pending_human", "approved", "rejected"]
    human_decision: str | None
    audit_trail: Annotated[list[str], lambda x, y: x + y]
    retry_count: int

# ========== LLM与工具初始化 ==========
llm = ChatOpenAI(model="gpt-4o", temperature=0.2)

async def get_credit_score(ssn: str) -> int:
    """模拟信用评分API调用"""
    await asyncio.sleep(0.5)  # 模拟网络延迟
    return 650 + (hash(ssn) % 300)

async def get_income_verification(applicant_id: str) -> dict:
    """模拟收入验证"""
    await asyncio.sleep(0.3)
    return {"verified_income": 85000, "source": "W2"}

# ========== 节点函数 ==========
async def data_collection(state: LoanState) -> LoanState:
    """收集并验证申请数据"""
    state["application_id"] = str(uuid.uuid4())
    
    # 并行获取数据
    credit_task = get_credit_score(state["applicant_data"]["ssn"])
    income_task = get_income_verification(state["applicant_data"]["applicant_id"])
    
    credit_score, income_data = await asyncio.gather(credit_task, income_task)
    
    state["applicant_data"]["credit_score"] = credit_score
    state["applicant_data"]["verified_income"] = income_data["verified_income"]
    
    state["audit_trail"].append(f"数据收集完成: credit_score={credit_score}, income={income_data['verified_income']}")
    return state

async def risk_assessment(state: LoanState) -> LoanState:
    """使用LLM进行风险评估"""
    prompt = f"""
    作为贷款审批专家，评估以下申请人的风险：
    - 信用评分: {state['applicant_data']['credit_score']}
    - 年收入: {state['applicant_data']['verified_income']}
    - 贷款金额: {state['applicant_data']['loan_amount']}
    - 负债率: {state['applicant_data']['debt_ratio']}
    
    请输出JSON格式：{{"risk_score": 0-100, "risk_factors": ["原因1", "原因2"]}}
    """
    
    response = await llm.ainvoke([SystemMessage(content=prompt)])
    result = json.loads(response.content)
    
    state["risk_score"] = result["risk_score"]
    state["risk_factors"] = result["risk_factors"]
    state["audit_trail"].append(f"风险评分: {result['risk_score']}, 因素: {result['risk_factors']}")
    return state

async def auto_approve(state: LoanState) -> LoanState:
    """自动批准低风险贷款"""
    state["approval_status"] = "auto_approved"
    state["audit_trail"].append("自动批准")
    return state

async def request_human_approval(state: LoanState) -> LoanState:
    """提交人工审批"""
    state["approval_status"] = "pending_human"
    
    # 发送到审批队列（Redis）
    r = redis.Redis(host='localhost', port=6379, db=0)
    approval_task = {
        "application_id": state["application_id"],
        "risk_score": state["risk_score"],
        "risk_factors": state["risk_factors"],
        "timestamp": datetime.now().isoformat()
    }
    r.lpush("human_approval_queue", json.dumps(approval_task))
    
    state["audit_trail"].append("已提交人工审批")
    return state

async def final_decision(state: LoanState) -> LoanState:
    """最终决策节点"""
    if state["human_decision"]:
        state["approval_status"] = state["human_decision"]
        state["audit_trail"].append(f"人工决策: {state['human_decision']}")
    return state

def route_decision(state: LoanState) -> Literal["auto_approve", "request_human_approval"]:
    """路由函数：根据风险分数决定路径"""
    if state["risk_score"] < 30:
        return "auto_approve"
    else:
        return "request_human_approval"

# ========== 构建图 ==========
def build_loan_graph():
    graph = StateGraph(LoanState)
    
    # 添加节点
    graph.add_node("data_collection", data_collection)
    graph.add_node("risk_assessment", risk_assessment)
    graph.add_node("auto_approve", auto_approve)
    graph.add_node("request_human_approval", request_human_approval)
    graph.add_node("final_decision", final_decision)
    
    # 设置入口
    graph.set_entry_point("data_collection")
    
    # 添加边
    graph.add_edge("data_collection", "risk_assessment")
    graph.add_conditional_edges(
        "risk_assessment",
        route_decision,
        {
            "auto_approve": "auto_approve",
            "request_human_approval": "request_human_approval"
        }
    )
    graph.add_edge("auto_approve", "final_decision")
    graph.add_edge("request_human_approval", "final_decision")
    graph.add_edge("final_decision", END)
    
    return graph.compile()

# ========== 执行 ==========
async def main():
    # 使用PostgreSQL持久化
    with PostgresSaver.from_conn_string(
        "postgresql://postgres:postgres@localhost:5432/agent_db"
    ) as checkpointer:
        app = build_loan_graph()
        
        # 初始状态
        initial_state = {
            "applicant_data": {
                "applicant_id": "APP-2026-001",
                "ssn": "123-45-6789",
                "loan_amount": 50000,
                "debt_ratio": 0.35
            },
            "risk_score": 0.0,
            "risk_factors": [],
            "approval_status": "pending",
            "human_decision": None,
            "audit_trail": [],
            "retry_count": 0
        }
        
        config = {"configurable": {"thread_id": "loan-2026-001"}}
        
        # 执行
        result = await app.ainvoke(initial_state, config)
        
        print("=== 最终状态 ===")
        print(f"申请ID: {result['application_id']}")
        print(f"风险评分: {result['risk_score']}")
        print(f"审批状态: {result['approval_status']}")
        print(f"审计日志: {result['audit_trail']}")
        
        # 模拟人工审批
        if result["approval_status"] == "pending_human":
            print("\n=== 模拟人工审批 ===")
            # 获取最新检查点
            state_history = app.get_state_history(config)
            latest_state = next(state_history)
            
            # 注入人工决策
            updated_state = latest_state.values.copy()
            updated_state["human_decision"] = "approved"
            
            # 重新执行
            final_result = await app.ainvoke(updated_state, config)
            print(f"最终状态: {final_result['approval_status']}")

if __name__ == "__main__":
    asyncio.run(main())

# requirements.txt
# pyautogen==0.9.1
# docker==7.1.0

import asyncio
from autogen import ConversableAgent, GroupChat, GroupChatManager
from autogen.coding import DockerCommandLineCodeExecutor

# ========== 配置LLM ==========
llm_config = {
    "config_list": [
        {
            "model": "gpt-4o",
            "api_key": "your-api-key",
            "temperature": 0.3,
        }
    ],
    "timeout": 120,
}

# ========== 创建带代码执行能力的Agent ==========
# 使用Docker沙箱执行代码
executor = DockerCommandLineCodeExecutor(
    image="python:3.11-slim",
    timeout=60,
    work_dir="/workspace",
)

architect = ConversableAgent(
    name="Architect",
    system_message="""你是资深系统架构师。你的职责：
1. 提出技术方案
2. 使用Python代码进行架构验证
3. 输出最终方案文档

回复格式：先给出方案，然后用代码验证，最后总结。""",
    llm_config=llm_config,
    code_execution_config={"executor": executor},
    human_input_mode="NEVER",
)

security_expert = ConversableAgent(
    name="SecurityExpert",
    system_message="""你是安全专家。职责：
1. 审查方案的安全漏洞
2. 使用代码模拟攻击场景
3. 给出加固建议

重点关注：SQL注入、XSS、认证授权问题。""",
    llm_config=llm_config,
    code_execution_config={"executor": executor},
    human_input_mode="NEVER",
)

performance_engineer = ConversableAgent(
    name="PerformanceEngineer",
    system_message="""你是性能工程师。职责：
1. 评估方案性能瓶颈
2. 编写基准测试代码
3. 给出优化建议

重点关注：并发、延迟、资源消耗。""",
    llm_config=llm_config,
    code_execution_config={"executor": executor},
    human_input_mode="NEVER",
)

# ========== 创建群聊 ==========
group_chat = GroupChat(
    agents=[architect, security_expert, performance_engineer],
    messages=[],
    max_round=15,
    speaker_selection_method="round_robin",  # 轮转发言
    enable_clear_history=True,
)

manager = GroupChatManager(
    groupchat=group_chat,
    llm_config=llm_config,
    is_termination_msg=lambda msg: "TERMINATE" in msg.get("content", ""),
)

# ========== 启动辩论 ==========
async def run_debate():
    # 初始化对话
    initial_message = """我们需要设计一个高并发的电商订单系统。请开始讨论。

要求：
- 支持每秒10万订单
- 数据最终一致性
- 可水平扩展

请架构师先提出方案，然后安全专家和性能工程师依次审查。最后输出完整的方案文档。"""

    await manager.a_initiate_chat(
        architect,
        message=initial_message,
    )
    
    # 获取完整对话历史
    print("\n=== 完整对话记录 ===")
    for msg in group_chat.messages:
        print(f"\n[{msg['name']}]:")
        print(msg['content'][:500] + "..." if len(msg['content']) > 500 else msg['content'])

if __name__ == "__main__":
    asyncio.run(run_debate())

# requirements.txt
# crewai==0.2.6
# langchain-community==0.3.12

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, ScrapeWebsiteTool
from langchain_openai import ChatOpenAI

# ========== 初始化工具 ==========
search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()

# ========== 定义Agent ==========
researcher = Agent(
    role="市场研究员",
    goal="发现2026年AI Agent领域的最新趋势与数据",
    backstory="你有10年科技行业研究经验，擅长从Crunchbase、Gartner报告中提取数据。",
    tools=[search_tool, scrape_tool],
    llm=ChatOpenAI(model="gpt-4o", temperature=0.4),
    allow_delegation=False,
    verbose=True,
    max_iterations=5,  # 防止无限循环
)

content_writer = Agent(
    role="资深技术作者",
    goal="将研究数据转化为有深度的技术文章",
    backstory="你曾为TechCrunch、InfoQ撰写过50+篇深度报道。",
    tools=[search_tool],
    llm=ChatOpenAI(model="gpt-4o", temperature=0.7),
    allow_delegation=True,  # 可以委托给SEO专家
    verbose=True,
    max_iterations=8,
)

seo_expert = Agent(
    role="SEO策略师",
    goal="优化文章关键词布局，提升搜索排名",
    backstory="你是Google SEO算法专家，精通语义搜索优化。",
    tools=[search_tool],
    llm=ChatOpenAI(model="gpt-4o", temperature=0.3),
    allow_delegation=False,
    verbose=True,
)

publisher = Agent(
    role="发布经理",
    goal="制定发布计划并安排社交媒体推广",
    backstory="你负责管理公司的内容日历和社交渠道。",
    llm=ChatOpenAI(model="gpt-4o", temperature=0.2),
    allow_delegation=False,
    verbose=True,
)

# ========== 定义任务（支持动态工具调用） ==========
research_task = Task(
    description="""研究2026年AI Agent框架市场，重点收集：
1. 四大框架（LangGraph、AutoGen、CrewAI、Dify）的最新版本与star数
2. 企业采用率数据（可从Gartner、IDC报告获取）
3. 3个真实案例（金融、医疗、电商各一个）

输出格式：结构化数据列表，包含来源URL。""",
    expected_output="包含数据的Markdown表格与来源链接",
    agent=researcher,
)

write_task = Task(
    description="""基于研究数据，撰写一篇3000字的深度文章，标题为《2026 AI Agent框架选型指南》。

要求：
- 开篇用数据冲击力吸引读者
- 中间部分对比LangGraph、AutoGen、CrewAI、Dify的架构差异
- 结尾给出选型建议

文章需包含：3个对比表格、2个架构图描述、1个实战案例。""",
    expected_output="完整的Markdown格式文章",
    agent=content_writer,
    context=[research_task],  # 依赖研究结果
)

seo_task = Task(
    description="""对文章进行SEO优化：
1. 提取5个目标关键词（如：AI Agent框架、LangGraph vs AutoGen）
2. 优化标题和副标题
3. 添加meta描述和alt文本建议

输出：优化建议列表。""",
    expected_output="包含关键词密度分析的SEO报告",
    agent=seo_expert,
    context=[write_task],
)

publish_task = Task(
    description="""制定发布计划：
1. 推荐发布平台（TechCrunch、Medium、InfoQ）
2. 安排社交媒体推广时间表
3. 预估流量与转化指标

输出：完整的发布日历。""",
    expected_output="包含时间线表格的发布计划",
    agent=publisher,
    context=[write_task, seo_task],
)

# ========== 创建Crew并执行 ==========
marketing_crew = Crew(
    agents=[researcher, content_writer, seo_expert, publisher],
    tasks=[research_task, write_task, seo_task, publish_task],
    process=Process.sequential,  # 顺序执行
    verbose=True,
    memory=True,  # 开启记忆