无需框架构建AI Agent：分步指南

def search_wikipedia(query: str) -> str:
    """
    Searches Wikipedia and returns the first paragraph of the most relevant article.

    Args:
        query: The search term to look up on Wikipedia

    Returns:
        str: The first paragraph of the most relevant Wikipedia article
    """
    return wikipedia.summary(query, sentences=3)

# This is how we describe the tool to the LLM
wikipedia_tool_description = """
Tool: search_wikipedia
Description: Searches Wikipedia and returns a summary of the topic
Input: A search query string
Example: To learn about Python programming, use: search_wikipedia("Python programming language")
"""

# This is what an agent interaction basically looks like
def agent_interaction(user_input, available_tools, conversation_history):
    # 1. We tell the LLM what it can do
    prompt = f"""You have access to these tools:
    - query_database: Executes SQL queries
    - search_wikipedia: Searches Wikipedia articles

    Previous conversation:
    {conversation_history}

    User question: {user_input}

    Decide if you need to use a tool to answer. If yes, specify which tool and how to use it."""

    # 2. The LLM decides what to do
    llm_response = gpt4_call(prompt)

    # 3. We execute the tool if needed and get the final answer
    if "use query_database" in llm_response:
        # Execute database query and get result
        data = execute_query(extract_query(llm_response))
        return get_final_answer(data)

    return llm_response

# With a framework:
agent = Agent.from_toolkit(SQLToolkit, WikiToolkit)
result = agent.run("Find users who like pizza")

# Without a framework - what's actually happening:
def process_query(user_input: str):
    prompt = f"""You have these tools:
    1. SQL queries: Execute database queries
    2. Wikipedia search: Look up information

    User question: {user_input}

    Think step by step:
    1. Do I need any tools to answer this?
    2. If yes, which tool and how should I use it?
    """

    response = gpt4_call(prompt)
    if "SQL" in response:
        query = extract_sql(response)
        data = execute_query(query)
    if "Wikipedia" in response:
        data = search_wikipedia(topic)
        return get_final_answer(data)

# Framework error:
AgentError: Tool execution failed: SQLToolkit.query failed with error:
Invalid chain configuration...

# Without framework:
try:
    data = execute_query(query)
except SQLException as e:
    print(f"SQL error: {e}")  # Clear what went wrong

import json
import psycopg2
import wikipedia
from typing import Dict, List, Any
from openai import OpenAI

# First, we define how the LLM should understand our tools
tools = [
    {
        "type": "function",
        "function": {
            "name": "query_database",
            "description": """Execute a PostgreSQL SELECT query and return the results.
                            Only SELECT queries are allowed for security reasons.
                            Returns data in JSON format.""",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": """The SQL SELECT query to execute.
                                        Must start with SELECT and should not
                                        contain any data modification commands."""
                    }
                },
                "required": ["query"],
                "additionalProperties": False
            },
            "strict": True
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_wikipedia",
            "description": """Search Wikipedia and return a concise summary.
                            Returns the first three sentences of the most
                            relevant article.""",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "The topic to search for on Wikipedia"
                    }
                },
                "required": ["query"],
                "additionalProperties": False
            },
            "strict": True
        }
    }
]

# Now implement the actual tool functions
def query_database(query: str) -> str:
    """
    Execute a PostgreSQL query and return results as a JSON string.
    Includes safety checks and error handling.
    """
    # Security check: Only allow SELECT queries
    if not query.lower().strip().startswith('select'):
        return json.dumps({
            "error": "Only SELECT queries are allowed for security reasons."
        })

    try:
        # You should use environment variables for these in production
        conn = psycopg2.connect(
            "dbname=your_db user=your_user password=your_pass host=localhost"
        )

        with conn.cursor() as cur:
            cur.execute(query)

            # Get column names from cursor description
            columns = [desc[0] for desc in cur.description]

            # Fetch results and convert to list of dictionaries
            results = []
            for row in cur.fetchall():
                results.append(dict(zip(columns, row)))

            return json.dumps({
                "success": True,
                "data": str(results),
                "row_count": len(results)
            })

    except psycopg2.Error as e:
        return json.dumps({
            "error": f"Database error: {str(e)}"
        })
    except Exception as e:
        return json.dumps({
            "error": f"Unexpected error: {str(e)}"
        })
    finally:
        if 'conn' in locals():
            conn.close()

def search_wikipedia(query: str) -> str:
    """
    Search Wikipedia and return a concise summary.
    Handles disambiguation and missing pages gracefully.
    """
    try:
        # Try to get the most relevant page summary
        summary = wikipedia.summary(
            query,
            sentences=3,
            auto_suggest=True,
            redirect=True
        )

        return json.dumps({
            "success": True,
            "summary": summary,
            "url": wikipedia.page(query).url
        })

    except wikipedia.DisambiguationError as e:
        # Handle multiple matching pages
        return json.dumps({
            "error": "Disambiguation error",
            "options": e.options[:5],  # List first 5 options
            "message": "Topic is ambiguous. Please be more specific."
        })
    except wikipedia.PageError:
        return json.dumps({
            "error": "Page not found",
            "message": f"No Wikipedia article found for: {query}"
        })
    except Exception as e:
        return json.dumps({
            "error": "Unexpected error",
            "message": str(e)
        })

from typing import List, Dict, Optional, Any
from openai import OpenAI
import json

class Agent:
    def __init__(self, system_prompt: Optional[str] = None):
        """
        Initialize an AI Agent with optional system prompt.

        Args:
            system_prompt: Initial instructions for the AI
        """
        # Initialize OpenAI client - expects OPENAI_API_KEY in environment
        self.client = OpenAI()

        # Initialize conversation history
        self.messages = []

        # Set up system prompt if provided, otherwise use default
        default_prompt = """You are a helpful AI assistant with access to a database
        and Wikipedia. Follow these rules:
        1. When asked about data, always check the database first
        2. For general knowledge questions, use Wikipedia
        3. If you're unsure about data, query the database to verify
        4. Always mention your source of information
        5. If a tool returns an error, explain the error to the user clearly
        """

        self.messages.append({
            "role": "system",
            "content": system_prompt or default_prompt
        })

    def execute_tool(self, tool_call: Any) -> str:
        """
        Execute a tool based on the LLM's decision.

        Args:
            tool_call: The function call object from OpenAI's API

        Returns:
            str: JSON-formatted result of the tool execution
        """
        try:
            function_name = tool_call.function.name
            function_args = json.loads(tool_call.function.arguments)

            # Execute the appropriate tool. Add more here as needed.
            if function_name == "query_database":
                result = query_database(function_args["query"])
            elif function_name == "search_wikipedia":
                result = search_wikipedia(function_args["query"])
            else:
                result = json.dumps({
                    "error": f"Unknown tool: {function_name}"
                })

            return result

        except json.JSONDecodeError:
            return json.dumps({
                "error": "Failed to parse tool arguments"
            })
        except Exception as e:
            return json.dumps({
                "error": f"Tool execution failed: {str(e)}"
            })

    def process_query(self, user_input: str) -> str:
        """
        Process a user query through the AI agent.

        Args:
            user_input: The user's question or command

        Returns:
            str: The agent's response
        """
        # Add user input to conversation history
        self.messages.append({
            "role": "user",
            "content": user_input
        })

        try:
            max_iterations = 5
            current_iteration = 0

            while current_iteration < max_iterations:  # Limit to 5 iterations
                current_iteration += 1
                completion = self.client.chat.completions.create(
                    model="gpt-4o",
                    messages=self.messages,
                    tools=tools,  # Global tools list from Step 1
                    tool_choice="auto"  # Let the model decide when to use tools
                )

                response_message = completion.choices[0].message

                # If no tool calls, we're done
                if not response_message.tool_calls:
                    self.messages.append(response_message)
                    return response_message.content

                # Add the model's thinking to conversation history
                self.messages.append(response_message)

                # Process all tool calls
                for tool_call in response_message.tool_calls:
                    try:
                        print("Tool call:", tool_call)
                        result = self.execute_tool(tool_call)
                        print("Tool executed......")
                    except Exception as e:
                        print("Execution failed......")
                        result = json.dumps({
                            "error": f"Tool execution failed: {str(e)}"
                        })

                    print(f"Tool result custom: {result}")

                    self.messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": str(result)
                    })
                    print("Messages:", self.messages)

            # If we've reached max iterations, return a message indicating this
            max_iterations_message = {
                "role": "assistant",
                "content": "I've reached the maximum number of tool calls (5) without finding a complete answer. Here's what I know so far: " + response_message.content
            }
            self.messages.append(max_iterations_message)
            return max_iterations_message["content"]

        except Exception as e:
            error_message = f"Error processing query: {str(e)}"
            self.messages.append({
                "role": "assistant",
                "content": error_message
            })
            return error_message

    def get_conversation_history(self) -> List[Dict[str, str]]:
        """
        Get the current conversation history.

        Returns:
            List[Dict[str, str]]: The conversation history
        """
        return self.messages

from datetime import datetime
import uuid
from typing import List, Dict, Optional
from dataclasses import dataclass
import psycopg2
from psycopg2.extras import Json, UUID_adapter

@dataclass
class MemoryConfig:
    """Configuration for memory management"""
    max_messages: int = 20  # When to summarize
    summary_length: int = 2000  # Max summary length in words
    db_connection: str = "dbname=your_db user=your_user password=your_pass"

class AgentMemory:
    def __init__(self, config: Optional[MemoryConfig] = None):
        self.config = config or MemoryConfig()
        self.session_id = uuid.uuid4()
        self.setup_database()

    def setup_database(self):
        """Create necessary database tables if they don't exist"""
        queries = [
            """
            CREATE TABLE IF NOT EXISTS conversations (
                id SERIAL PRIMARY KEY,
                session_id UUID NOT NULL,
                user_input TEXT NOT NULL,
                agent_response TEXT NOT NULL,
                tool_calls JSONB,
                timestamp TIMESTAMPTZ DEFAULT NOW()
            );
            """,
            """
            CREATE TABLE IF NOT EXISTS conversation_summaries (
                id SERIAL PRIMARY KEY,
                session_id UUID NOT NULL,
                summary TEXT NOT NULL,
                start_time TIMESTAMPTZ NOT NULL,
                end_time TIMESTAMPTZ NOT NULL,
                message_count INTEGER NOT NULL
            );
            """
        ]

        with psycopg2.connect(self.config.db_connection) as conn:
            with conn.cursor() as cur:
                for query in queries:
                    cur.execute(query)

    def store_interaction(self,
                         user_input: str,
                         agent_response: str,
                         tool_calls: Optional[List[Dict]] = None):
        """Store a single interaction in the database"""
        query = """
        INSERT INTO conversations
            (session_id, user_input, agent_response, tool_calls)
        VALUES (%s, %s, %s, %s)
        """

        with psycopg2.connect(self.config.db_connection) as conn:
            with conn.cursor() as cur:
                cur.execute(query, (
                    self.session_id,
                    user_input,
                    agent_response,
                    Json(tool_calls) if tool_calls else None
                ))

    def create_summary(self, messages: List[Dict]) -> str:
        """Create a summary of messages using the LLM"""
        client = OpenAI()

        # Prepare messages for summarization
        summary_prompt = f"""
        Summarize the following conversation in less than {self.config.summary_length} words.
        Focus on key points, decisions, and important information discovered through tool usage.

        Conversation:
        {messages}
        """

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": summary_prompt}]
        )

        return response.choices[0].message.content

    def store_summary(self, summary: str, start_time: datetime, end_time: datetime, message_count: int):
        """Store a conversation summary"""
        query = """
        INSERT INTO conversation_summaries
            (session_id, summary, start_time, end_time, message_count)
        VALUES (%s, %s, %s, %s, %s)
        """

        with psycopg2.connect(self.config.db_connection) as conn:
            with conn.cursor() as cur:
                cur.execute(query, (
                    self.session_id,
                    summary,
                    start_time,
                    end_time,
                    message_count
                ))

    def get_recent_context(self) -> str:
        """Get recent conversations and summaries for context"""
        # First, get the most recent summary
        summary_query = """
        SELECT summary, end_time
        FROM conversation_summaries
        WHERE session_id = %s
        ORDER BY end_time DESC
        LIMIT 1
        """

        # Then get conversations after the summary
        conversations_query = """
        SELECT user_input, agent_response, tool_calls, timestamp
        FROM conversations
        WHERE session_id = %s
        AND timestamp > %s
        ORDER BY timestamp ASC
        """

        with psycopg2.connect(self.config.db_connection) as conn:
            with conn.cursor() as cur:
                # Get latest summary
                cur.execute(summary_query, (self.session_id,))
                summary_row = cur.fetchone()

                if summary_row:
                    summary, last_summary_time = summary_row
                    # Get conversations after the summary
                    cur.execute(conversations_query, (self.session_id, last_summary_time))
                else:
                    # If no summary exists, get recent conversations
                    cur.execute("""
                        SELECT user_input, agent_response, tool_calls, timestamp
                        FROM conversations
                        WHERE session_id = %s
                        ORDER BY timestamp DESC
                        LIMIT %s
                    """, (self.session_id, self.config.max_messages))

                conversations = cur.fetchall()

        # Format context
        context = []
        if summary_row:
            context.append(f"Previous conversation summary: {summary}")

        for conv in conversations:
            user_input, agent_response, tool_calls, _ = conv
            context.append(f"User: {user_input}")
            if tool_calls:
                context.append(f"Tool Usage: {tool_calls}")
            context.append(f"Assistant: {agent_response}")

        return "\n".join(context)

    def check_and_summarize(self):
        """Check if we need to summarize and do it if necessary"""
        query = """
        SELECT COUNT(*)
        FROM conversations
        WHERE session_id = %s
        AND timestamp > (
            SELECT COALESCE(MAX(end_time), '1970-01-01'::timestamptz)
            FROM conversation_summaries
            WHERE session_id = %s
        )
        """

        with psycopg2.connect(self.config.db_connection) as conn:
            with conn.cursor() as cur:
                cur.execute(query, (self.session_id, self.session_id))
                count = cur.fetchone()[0]

                if count >= self.config.max_messages:
                    # Get messages to summarize
                    cur.execute("""
                        SELECT user_input, agent_response, tool_calls, timestamp
                        FROM conversations
                        WHERE session_id = %s
                        ORDER BY timestamp ASC
                        LIMIT %s
                    """, (self.session_id, count))

                    messages = cur.fetchall()
                    if messages:
                        # Create and store summary
                        summary = self.create_summary(messages)
                        self.store_summary(
                            summary,
                            messages[0][3],  # start_time
                            messages[-1][3],  # end_time
                            len(messages)
                        )

# Update Agent class to use memory.
class MemoryAgent(Agent):
    def __init__(self, memory_config: Optional[MemoryConfig] = None):
        self.client = OpenAI()
        self.memory = AgentMemory(memory_config)
        self.messages = []

        # Initialize with system prompt
        self.messages.append({
            "role": "system",
            "content": "You are a helpful AI assistant..."
        })

    def process_query(self, user_input: str) -> str:
        try:
            # Check if we need to summarize before adding new messages
            self.memory.check_and_summarize()

            # Get context (including summaries) from memory
            context = self.memory.get_recent_context()

            # Add context to messages if it exists
            if context:
                self.messages = [
                    self.messages[0],  # Keep system prompt
                    {
                        "role": "system",
                        "content": f"Previous conversation context:\n{context}"
                    }
                ]

            # Process the query as before...
            response = super().process_query(user_input)

            # Store the interaction in memory
            tool_calls = None
            if hasattr(self, 'last_tool_calls'):
                tool_calls = self.last_tool_calls

            self.memory.store_interaction(
                user_input=user_input,
                agent_response=response,
                tool_calls=tool_calls
            )

            return response

        except Exception as e:
            error_message = f"Error processing query: {str(e)}"
            self.memory.store_interaction(
                user_input=user_input,
                agent_response=error_message
            )
            return error_message

    def execute_tool(self, tool_call: Any) -> str:
        # Store tool calls for memory
        if not hasattr(self, 'last_tool_calls'):
            self.last_tool_calls = []

        result = super().execute_tool(tool_call)

        # Store tool call information
        self.last_tool_calls.append({
            'tool': tool_call.function.name,
            'arguments': tool_call.function.arguments,
            'result': result
        })

        return result

from your_agent import Agent, MemoryConfig  # assuming we saved our code in your_agent.py

# Create an agent
agent = Agent(
    memory_config=MemoryConfig(
        max_messages=10,  # summarize after 10 messages
        db_connection="postgresql://user:pass@localhost/your_db"
    )
)

# Simple question-answer interaction
def chat_with_agent(question: str):
    print(f"\nUser: {question}")
    print(f"Assistant: {agent.process_query(question)}")

if __name__ == "__main__":
    chat_with_agent("How many users do we have in our database?")
    chat_with_agent("What's the average age of users from New York?")
    chat_with_agent("Compare our user base in California with PostgreSQL's popularity there")
    chat_with_agent("What were the numbers you just mentioned?")
    chat_with_agent("What are the most interesting trends you can find in our user data?")

def get_database_schema() -> str:
    """Retrieve the database schema information"""
    schema_query = """
    SELECT
        t.table_name,
        array_agg(
            c.column_name || ' ' ||
            c.data_type ||
            CASE
                WHEN c.is_nullable = 'NO' THEN ' NOT NULL'
                ELSE ''
            END
        ) as columns
    FROM information_schema.tables t
    JOIN information_schema.columns c
        ON c.table_name = t.table_name
    WHERE t.table_schema = 'public'
        AND t.table_type = 'BASE TABLE'
    GROUP BY t.table_name;
    """

    try:
        with psycopg2.connect("dbname=your_db user=your_user password=your_pass") as conn:
            with conn.cursor() as cur:
                cur.execute(schema_query)
                schema = cur.fetchall()

                # Format schema information
                schema_str = "Database Schema:\n"
                for table_name, columns in schema:
                    schema_str += f"\n{table_name}\n"
                    for col in columns:
                        schema_str += f"  - {col}\n"

                return schema_str
    except Exception as e:
        return f"Error fetching schema: {str(e)}"

# Updated tool definition with schema information;
# Note that we could also dynamically fetch the schema when needed
tools = [
    {
        "type": "function",
        "function": {
            "name": "query_database",
            "description": """Execute a PostgreSQL SELECT query and return the results.
            Available tables and their schemas:

            users
              - id SERIAL PRIMARY KEY
              - email VARCHAR(255) NOT NULL
              - age INTEGER
              - location VARCHAR(100)
              - signup_date DATE NOT NULL
              - last_login TIMESTAMP WITH TIME ZONE
              - job_industry VARCHAR(100)

            user_activity
              - id SERIAL PRIMARY KEY
              - user_id INTEGER REFERENCES users(id)
              - activity_date DATE NOT NULL
              - activity_type VARCHAR(50) NOT NULL
            """,
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "The SQL SELECT query to execute. Must start with SELECT for security."
                    }
                },
                "required": ["query"],
                "additionalProperties": False
            },
            "strict": True
        }
    }
]

# Updated database query function
def query_database(query: str) -> str:
    """
    Execute a PostgreSQL query with schema awareness
    """
    if not query.lower().strip().startswith('select'):
        return json.dumps({
            "error": "Only SELECT queries are allowed for security reasons.",
            "schema": get_database_schema()  # Return schema for reference
        })

    try:
        with psycopg2.connect("dbname=your_db user=your_user password=your_pass") as conn:
            with conn.cursor() as cur:
                cur.execute(query)

                # Get column names from cursor description
                columns = [desc[0] for desc in cur.description]

                # Fetch results and convert to list of dictionaries
                results = []
                for row in cur.fetchall():
                    results.append(dict(zip(columns, row)))

                return json.dumps({
                    "success": True,
                    "data": str(results),
                    "row_count": len(results),
                    "columns": columns
                })

    except Exception as e:
        return json.dumps({
            "error": str(e),
            "schema": get_database_schema()  # Return schema on error for help
        })

# Usage example:
agent = Agent()

# The agent now knows about the schema and can write better queries
response = agent.process_query(
    "What's the age distribution of users by location and industry?"
)