AI Agents and Autonomous Systems 2025

import autogen

# Configure LLM
config_list = [{
    "model": "gpt-4",
    "api_key": "sk-..."
}]

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

# Create specialized agents
researcher = autogen.AssistantAgent(
    name="Researcher",
    system_message="""You are a research specialist. Your role is to
    gather information, analyze data sources, and provide comprehensive
    research summaries. Use web search and databases.""",
    llm_config=llm_config,
)

writer = autogen.AssistantAgent(
    name="Writer",
    system_message="""You are a content writer. Your role is to create
    engaging, well-structured content based on research. Focus on clarity
    and readability.""",
    llm_config=llm_config,
)

critic = autogen.AssistantAgent(
    name="Critic",
    system_message="""You are a critical reviewer. Your role is to
    evaluate content quality, fact-check, and suggest improvements.
    Be constructive but thorough.""",
    llm_config=llm_config,
)

# User proxy for human-in-the-loop
user_proxy = autogen.UserProxyAgent(
    name="User",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=5,
    code_execution_config={"work_dir": "output"},
)

# Create group chat
groupchat = autogen.GroupChat(
    agents=[user_proxy, researcher, writer, critic],
    messages=[],
    max_round=10,
    speaker_selection_method="round_robin"
)

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

# Start collaboration
user_proxy.initiate_chat(
    manager,
    message="""Create a comprehensive blog post about quantum computing.
    Researcher: gather latest developments. Writer: create engaging article.
    Critic: review for accuracy and clarity."""
)

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated

# Define state
class AgentState(TypedDict):
    task: str
    research: str
    draft: str
    review: str
    final: str

# Agent functions
def research_agent(state: AgentState) -> AgentState:
    llm = ChatOpenAI(model="gpt-4")
    result = llm.invoke(f"Research this topic: {state['task']}")
    return {"research": result.content}

def writer_agent(state: AgentState) -> AgentState:
    llm = ChatOpenAI(model="gpt-4")
    result = llm.invoke(f"Write article based on: {state['research']}")
    return {"draft": result.content}

def critic_agent(state: AgentState) -> AgentState:
    llm = ChatOpenAI(model="gpt-4")
    result = llm.invoke(f"Review and improve: {state['draft']}")
    return {"review": result.content}

def should_continue(state: AgentState) -> str:
    # Decision logic: iterate or finalize
    if "needs improvement" in state.get("review", "").lower():
        return "writer"
    return "end"

# Build graph
workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("researcher", research_agent)
workflow.add_node("writer", writer_agent)
workflow.add_node("critic", critic_agent)

# Add edges
workflow.set_entry_point("researcher")
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "critic")
workflow.add_conditional_edges(
    "critic",
    should_continue,
    {
        "writer": "writer",
        "end": END
    }
)

# Compile and run
app = workflow.compile()
result = app.invoke({
    "task": "Explain quantum computing to beginners"
})

from crewai import Agent, Task, Crew, Process

# Define agents
researcher = Agent(
    role='Research Analyst',
    goal='Gather comprehensive information on quantum computing',
    backstory="""You are an expert researcher with deep knowledge in
    physics and computer science. You excel at finding reliable sources.""",
    verbose=True,
    allow_delegation=False,
)

writer = Agent(
    role='Tech Writer',
    goal='Create engaging, accurate technical content',
    backstory="""You are a skilled technical writer who can explain
    complex topics clearly. You prioritize reader comprehension.""",
    verbose=True,
    allow_delegation=False,
)

editor = Agent(
    role='Editor',
    goal='Ensure content quality and accuracy',
    backstory="""You are a meticulous editor with expertise in tech
    communication. You catch errors and improve clarity.""",
    verbose=True,
    allow_delegation=False,
)

# Define tasks
research_task = Task(
    description="""Research quantum computing: current state,
    key concepts, recent breakthroughs, practical applications.""",
    agent=researcher,
    expected_output="Comprehensive research summary"
)

writing_task = Task(
    description="""Write a 1000-word article about quantum computing
    for tech-savvy beginners. Use research provided.""",
    agent=writer,
    expected_output="Complete article draft"
)

editing_task = Task(
    description="""Review article for accuracy, clarity, and engagement.
    Make final improvements.""",
    agent=editor,
    expected_output="Polished final article"
)

# Create crew
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential,  # or Process.hierarchical
    verbose=2
)

# Execute
result = crew.kickoff()
print(result)

from langchain.callbacks import get_openai_callback
from tenacity import retry, stop_after_attempt, wait_exponential
import logging

class ProductionAgent:
    def __init__(self):
        self.max_iterations = 10
        self.timeout = 60
        self.budget_limit = 0.50  # $0.50 per execution

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    def execute_with_retry(self, task: str) -> str:
        """Execute agent with automatic retry on failure"""
        try:
            with get_openai_callback() as cb:
                result = self.agent.invoke(task)

                # Cost guard
                if cb.total_cost > self.budget_limit:
                    raise ValueError(f"Budget exceeded: {cb.total_cost}")

                # Log metrics
                logging.info(f"Cost: {cb.total_cost}, Tokens: {cb.total_tokens}")
                return result

        except Exception as e:
            logging.error(f"Agent execution failed: {e}")
            raise

    def execute_with_circuit_breaker(self, task: str) -> str:
        """Circuit breaker pattern for preventing cascading failures"""
        if self.circuit_breaker.is_open():
            raise Exception("Circuit breaker open - too many failures")

        try:
            result = self.execute_with_retry(task)
            self.circuit_breaker.record_success()
            return result
        except Exception as e:
            self.circuit_breaker.record_failure()
            raise

    def execute_with_timeout(self, task: str) -> str:
        """Timeout protection"""
        import signal

        def timeout_handler(signum, frame):
            raise TimeoutError("Agent execution timeout")

        signal.signal(signal.SIGALRM, timeout_handler)
        signal.alarm(self.timeout)

        try:
            result = self.execute_with_circuit_breaker(task)
            signal.alarm(0)  # Cancel alarm
            return result
        except TimeoutError:
            logging.error("Agent timeout - execution too long")
            raise

import os
from langsmith import Client
from langchain.callbacks.tracers import LangChainTracer

# Configure LangSmith
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls-..."
os.environ["LANGCHAIN_PROJECT"] = "production-agent"

# Create tracer
tracer = LangChainTracer(
    project_name="production-agent",
    client=Client()
)

# Use with agent
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[tracer],
    verbose=True
)

# Every execution is traced in LangSmith dashboard:
# - Full conversation history
# - Tool calls and results
# - Token usage and costs
# - Latency per step
# - Error traces

# Query traces programmatically
client = Client()
runs = client.list_runs(
    project_name="production-agent",
    filter='eq(status, "error")',
    limit=100
)

for run in runs:
    print(f"Failed run: {run.id}")
    print(f"Error: {run.error}")
    print(f"Input: {run.inputs}")

from enum import Enum
from typing import Callable, Any

class ToolPermission(Enum):
    READ = "read"
    WRITE = "write"
    EXECUTE = "execute"

class SecureTool:
    def __init__(
        self,
        name: str,
        func: Callable,
        required_permissions: list[ToolPermission],
        allowed_users: list[str] = None
    ):
        self.name = name
        self.func = func
        self.required_permissions = required_permissions
        self.allowed_users = allowed_users or []

    def execute(self, user_id: str, **kwargs) -> Any:
        # Permission check
        if not self._has_permission(user_id):
            raise PermissionError(f"User {user_id} lacks permission")

        # Audit log
        self._log_execution(user_id, kwargs)

        # Rate limiting
        if self._is_rate_limited(user_id):
            raise Exception("Rate limit exceeded")

        # Sandboxed execution
        try:
            result = self._execute_sandboxed(kwargs)
            return result
        except Exception as e:
            self._log_error(user_id, e)
            raise

    def _execute_sandboxed(self, kwargs: dict) -> Any:
        """Execute in isolated environment"""
        # Docker container or restricted subprocess
        import subprocess

        # Example: run in container
        result = subprocess.run(
            ["docker", "run", "--rm", "--network=none",
             "agent-sandbox", "python", "tool.py"],
            input=str(kwargs),
            capture_output=True,
            timeout=30
        )
        return result.stdout

# Usage
database_tool = SecureTool(
    name="database_query",
    func=execute_sql,
    required_permissions=[ToolPermission.READ],
    allowed_users=["admin", "analyst"]
)

file_write_tool = SecureTool(
    name="file_write",
    func=write_file,
    required_permissions=[ToolPermission.WRITE, ToolPermission.EXECUTE],
    allowed_users=["admin"]
)