Monism Agents By Implementing Mixture of Ego

{
  "mixtureOfEgo": [
    {
      "roleName": "Investment Analyst",
      "expertise": ["Financial Analysis", "Market Research", "Valuation"],
      "decisionAuthority": "Research and Recommendations",
      "weightedScore": 0.25
    },
    {
      "roleName": "Portfolio Manager",
      "expertise": [
        "Strategic Planning",
        "Risk Management",
        "Asset Allocation"
      ],
      "decisionAuthority": "Final Investment Decisions",
      "weightedScore": 0.5
    },
    {
      "roleName": "Risk Officer",
      "expertise": ["Risk Assessment", "Compliance", "Regulatory Requirements"],
      "decisionAuthority": "Risk Approval",
      "weightedScore": 0.25
    }
  ]
}

pip install langchain langchain-openai langgraph

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langchain.agents import load_tools
from typing import Dict, Any

personas = {
    "analyst": {
        "prompt": ChatPromptTemplate.from_messages([
            ("system", """You are a financial analyst specializing in market research and financial analysis.
            Provide comprehensive, evidence-based analysis using available data.
            Be thorough, analytical, and data-driven in your responses."""),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ]),
        "llm": ChatOpenAI(temperature=0, model="gpt-3.5-turbo"),
        "tools": load_tools(["serpapi"], llm=ChatOpenAI(temperature=0, model="gpt-3.5-turbo")),
        "description": "Conducts market research and financial analysis"
    },

    "manager": {
        "prompt": ChatPromptTemplate.from_messages([
            ("system", """You are a portfolio manager making strategic investment decisions.
            Base decisions on analyst findings while maintaining risk awareness.
            Ensure alignment with investment objectives and risk tolerance."""),
            ("human", "{input}"),
            ("ai", "Analyst's findings: {agent_analyst_output}"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ]),
        "llm": ChatOpenAI(temperature=0, model="gpt-3.5-turbo"),
        "tools": [],
        "description": "Makes strategic investment decisions"
    },

    "executor": {
        "prompt": ChatPromptTemplate.from_messages([
            ("system", """You are an execution specialist translating decisions into action.
            Convert strategic decisions into clear, implementable steps.
            Ensure proper execution through detailed guidance."""),
            ("human", "{input}"),
            ("ai", "Manager's decision: {agent_manager_output}"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ]),
        "llm": ChatOpenAI(temperature=0, model="gpt-3.5-turbo"),
        "tools": [],
        "description": "Translates decisions into actionable steps"
    }
}

def select_persona(task: str) -> str:
    """Dynamic persona selection based on task keywords."""
    task = task.lower()

    task_mapping = {
        "analyze": "analyst", "research": "analyst", "investigate": "analyst",
        "decide": "manager", "strategy": "manager", "plan": "manager",
        "execute": "executor", "implement": "executor", "action": "executor"
    }

    return next((persona for keyword, persona in task_mapping.items()
                if keyword in task), "analyst")

from langchain.agents import AgentExecutor
from langchain.agents.react.agent import ReActAgent

def create_react_agent(persona_name: str) -> AgentExecutor:
    """
    Creates a ReAct agent for a specific persona with appropriate tools and prompts.
    """
    persona = personas[persona_name]

    agent = ReActAgent.from_llm_and_tools(
        llm=persona["llm"],
        tools=persona["tools"],
        prompt=persona["prompt"],
    )

    return AgentExecutor(
        agent=agent,
        tools=persona["tools"],
        verbose=True,
        handle_parsing_errors=True
    )

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

class AgentState(Dict, Any):
    """
    Represents the state of the multi-agent system.
    Maintains context across different personas and execution phases.
    """
    messages: list
    agent_analyst_output: str = ""
    agent_manager_output: str = ""
    agent_executor_output: str = ""
    current_task: str = ""
    selected_persona: str = ""
    execution_history: list = []

# Create the workflow graph
workflow = StateGraph(AgentState)

# Add nodes for each persona
workflow.add_node("analyst", create_react_agent("analyst"))
workflow.add_node("manager", create_react_agent("manager"))
workflow.add_node("executor", create_react_agent("executor"))

def route_tasks(state: AgentState) -> str:
    """
    Routes tasks to appropriate personas based on the current task requirements.
    This function implements the core MoE routing logic.
    """
    selected_persona = select_persona(state["current_task"])
    state["selected_persona"] = selected_persona

    # Route to the selected persona
    if selected_persona == "analyst":
        return "analyst"
    elif selected_persona == "manager":
        return "manager"
    elif selected_persona == "executor":
        return "executor"
    else:
        return END

# Define conditional edges for dynamic routing
workflow.add_conditional_edges("analyst", route_tasks)
workflow.add_conditional_edges("manager", route_tasks)
workflow.add_conditional_edges("executor", route_tasks)

# Set entry point and compile the graph
workflow.set_entry_point("analyst")
graph = workflow.compile()

def execute_decision_pipeline(task: str) -> Dict[str, Any]:
    """
    Executes the complete decision pipeline using the MoE framework.

    Args:
        task: The initial task description

    Returns:
        Dictionary containing the complete execution results
    """
    # Initialize the system state
    initial_state = {
        "messages": [HumanMessage(content=task)],
        "current_task": task,
        "execution_history": []
    }

    # Execute the workflow
    result = graph.invoke(initial_state)

    return {
        "task": task,
        "selected_persona": result["selected_persona"],
        "analyst_output": result["agent_analyst_output"],
        "manager_output": result["agent_manager_output"],
        "executor_output": result["agent_executor_output"],
        "execution_history": result["execution_history"]
    }

# Example usage
if __name__ == "__main__":
    task = "Analyze the current market trends for technology stocks and provide investment recommendations"

    result = execute_decision_pipeline(task)

    print("=== Decision Pipeline Results ===")
    print(f"Task: {result['task']}")
    print(f"Selected Persona: {result['selected_persona']}")
    print(f"Analyst Output: {result['analyst_output']}")
    print(f"Manager Decision: {result['manager_output']}")
    print(f"Execution Plan: {result['executor_output']}")

def collect_persona_votes(decision: str, state: AgentState) -> Dict[str, str]:
    """
    Collects votes from all personas on a given decision.
    Implements consensus-based decision making in the MoE framework.
    """
    votes = {}

    for persona_name, persona_config in personas.items():
        if persona_name != state["selected_persona"]:
            # Create voting prompt for each persona
            vote_prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are a {persona_name}. Evaluate the following decision and vote 'approve' or 'reject' with reasoning."),
                ("human", f"Decision: {decision}\nPlease vote and provide your reasoning.")
            ])

            # Get vote from persona
            chain = vote_prompt | persona_config["llm"]
            vote_result = chain.invoke({"decision": decision})
            votes[persona_name] = vote_result.content

    return votes

def calculate_dynamic_weights(task: str, historical_performance: Dict[str, float]) -> Dict[str, float]:
    """
    Calculates dynamic weights for personas based on task requirements and performance history.
    """
    base_weights = {
        "analyst": 0.3,
        "manager": 0.5,
        "executor": 0.2
    }

    # Adjust weights based on task characteristics
    if "research" in task.lower() or "analysis" in task.lower():
        base_weights["analyst"] += 0.2
        base_weights["manager"] -= 0.1
        base_weights["executor"] -= 0.1

    # Normalize weights
    total = sum(base_weights.values())
    normalized_weights = {k: v/total for k, v in base_weights.items()}

    return normalized_weights

# Analyze recent price trends for specific stocks
python main.py -q "Analyze recent price trends for 600960"

# Custom date range analysis
python main.py -q "symbol=000001,start_date=20230801,end_date=20230810,period=daily,adjust=hfq"

# Get financial announcements
python main.py -q "Get notifications for notice_type=财务报告,date=20220511"