# PyAgent Framework - AI Development Rules (English Version)

> **Note**: This is the English version of PyAgent development rules for AI assistants like Cursor, Windsurf, etc.
> 
> **中文版本**: See `.cursorrules` for the Chinese version with additional details.

## Quick Reference

### Most Critical Rule

```python
# ✅ ALWAYS CORRECT
from pyagent import Chater, get_chater_cfg
chater = Chater(get_chater_cfg("ali"))

# ❌ ALWAYS WRONG - Will cause AttributeError
chater = get_chater_cfg("ali")  # This is just a config object!
```

**Why**: `get_chater_cfg()` returns a `ChaterCfg` configuration object, NOT a `Chater` instance. You MUST wrap it with `Chater()` to use it.

### Core Principles

1. **Simple** - Minimalist design, only 12 core modules
2. **Async** - Everything is async, always use `await`
3. **Composable** - Mix and match agents, tools, pipelines

## Architecture Overview

```
PyAgent Framework
├── Model Layer (Chater, ChaterPool, Embedder)
├── Agent Layer (Agent, ReActAgent, AgenticMemoryAgent)
├── Tool Layer (ToolKit, MCP)
├── Runner Layer (Runner - unified execution)
├── Communication Layer (Pipeline, MsgHub)
├── Trace Layer (Zero-overhead tracing)
├── Storage Layer (Memory, VectorStore)
└── Speaker Layer (Output control)
```

## Essential APIs

### Agent Creation

```python
from pyagent import Agent, Memory, Chater, get_chater_cfg

agent = Agent(
    name="Assistant",
    chater=Chater(get_chater_cfg("ali")),  # MUST wrap!
    memory=Memory(),
    system_prompt="You are helpful",
    tools=None,              # Optional ToolKit
    max_iterations=5,        # Tool call limit
    tool_timeout=None,       # Timeout in seconds
    enable_logging=False,
)
```

### Running Agent

```python
from pyagent import Runner

# Single execution
result = await Runner.run(agent, "Hello")
print(result.content)

# Streaming
async for chunk in Runner.run_streamed(agent, "Tell a story"):
    print(chunk.content, end="", flush=True)

# Sequential multi-agent
result = await Runner.run_sequential([agent1, agent2], "Task")

# Parallel multi-agent
results = await Runner.run_parallel([agent1, agent2, agent3], "Task")
```

### Tool Definition

```python
from pyagent import ToolKit

toolkit = ToolKit()

@toolkit.tool
async def search(query: str) -> str:
    """Search for information
    
    Args:
        query: Search query
    
    Returns:
        Search results
    """
    return f"Results for: {query}"

# IMPORTANT:
# 1. MUST be async function
# 2. MUST have type hints
# 3. MUST have docstring
# 4. Parameter name CANNOT be 'name'
```

### Hook System

```python
# Instance-level hooks
@agent.pre_reply
def log_input(msg):
    print(f"Input: {msg}")
    return msg  # MUST return

@agent.post_reply
def format_output(response):
    response.content = f"✨ {response.content}"
    return response  # MUST return

# Class-level hooks (affects ALL instances)
from pyagent import BaseAgent

@BaseAgent.pre_reply
def global_hook(agent, msg):
    print(f"{agent.name} received: {msg}")
    return msg
```

## Common Patterns

### Pattern 1: Simple Chat
```python
async def chat():
    agent = Agent(
        name="Bot",
        chater=Chater(get_chater_cfg("ali")),
        memory=Memory(),
        system_prompt="You are helpful"
    )
    result = await Runner.run(agent, "Hello")
    return result.content
```

### Pattern 2: With Tools
```python
async def with_tools():
    toolkit = ToolKit()
    
    @toolkit.tool
    async def get_time() -> str:
        """Get current time"""
        from datetime import datetime
        return datetime.now().strftime("%H:%M:%S")
    
    agent = Agent(
        name="Bot",
        chater=Chater(get_chater_cfg("ali")),
        memory=Memory(),
        tools=toolkit
    )
    
    result = await Runner.run(agent, "What time is it?")
    return result.content
```

### Pattern 3: Multi-Agent
```python
async def multi_agent():
    planner = Agent(
        name="Planner",
        chater=Chater(get_chater_cfg("ali")),
        memory=Memory(),
        system_prompt="You plan tasks"
    )
    
    executor = Agent(
        name="Executor",
        chater=Chater(get_chater_cfg("ali")),
        memory=Memory(),
        system_prompt="You execute tasks"
    )
    
    result = await Runner.run_sequential(
        [planner, executor],
        "Create a todo app"
    )
    return result.content
```

### Pattern 4: Streaming
```python
async def streaming():
    agent = Agent(
        name="Storyteller",
        chater=Chater(get_chater_cfg("ali")),
        memory=Memory()
    )
    
    async for chunk in Runner.run_streamed(agent, "Tell a story"):
        if chunk.content:
            print(chunk.content, end="", flush=True)
```

### Pattern 5: Pipeline
```python
from pyagent import sequential_pipeline, ChatResponse

async def pipeline():
    result = await sequential_pipeline(
        [agent1, agent2, agent3],
        ChatResponse(role="user", content="Input")
    )
    return result.content
```

## Common Errors & Solutions

### Error 1: AttributeError: 'ChaterCfg' object has no attribute 'chat'

```python
# ❌ Wrong
chater = get_chater_cfg("ali")

# ✅ Correct
chater = Chater(get_chater_cfg("ali"))
```

**Cause**: Using config object directly instead of wrapping with `Chater()`

### Error 2: Forgot await

```python
# ❌ Wrong - returns coroutine
result = Runner.run(agent, "Input")

# ✅ Correct
result = await Runner.run(agent, "Input")
```

**Cause**: All agent operations are async

### Error 3: Sync tool function

```python
# ❌ Wrong
def my_tool(x: int) -> int:
    return x * 2

# ✅ Correct
async def my_tool(x: int) -> int:
    return x * 2
```

**Cause**: Tools must be async functions

### Error 4: Parameter name conflict

```python
# ❌ Wrong - 'name' conflicts with execute()
async def my_tool(name: str) -> str:
    return f"Hello {name}"

# ✅ Correct - use different parameter name
async def my_tool(username: str) -> str:
    return f"Hello {username}"
```

**Cause**: `name` parameter conflicts with `toolkit.execute(name=...)`

### Error 5: Missing type hints

```python
# ❌ Wrong - no type hints
async def my_tool(x):
    return x * 2

# ✅ Correct - with type hints
async def my_tool(x: int) -> int:
    return x * 2
```

**Cause**: Type hints required for OpenAI tool format generation

## Testing with Mock

```python
import pytest
from pyagent import Agent, Memory, ChatResponse

class MockChater:
    async def chat(self, messages, stream=False, tools=None, tool_choice=None):
        if stream:
            async def generate():
                yield ChatResponse(role="assistant", content="Mock")
            return generate()
        return ChatResponse(role="assistant", content="Mock response")

@pytest.mark.asyncio
async def test_agent():
    agent = Agent(
        name="Test",
        chater=MockChater(),
        memory=Memory()
    )
    
    result = None
    async for resp in agent.reply("Test"):
        result = resp
    
    assert result.content == "Mock response"
```

## Performance Tips

1. **Use parallel execution**: `parallel_pipeline` instead of loops
2. **Limit memory**: `Memory(max_messages=20)`
3. **Set timeouts**: `tool_timeout=30`
4. **Disable logging**: `enable_logging=False` in production
5. **Use ChaterPool**: Automatic failover

## Debugging

1. **Enable logging**: `enable_logging=True, log_level="DEBUG"`
2. **Use trace**: `with trace("debug"): ...`
3. **Check memory**: `print(agent.memory.to_openai())`
4. **Console trace**: `enable_console_output()`

## Supported LLM Providers

- `ali` - Alibaba Cloud Qwen (Recommended)
- `zhipuai` - Zhipu AI
- `siliconflow` - SiliconFlow
- `ark` - Volcano Engine

## Environment Setup

```python
import os

# Set API keys
os.environ["DASHSCOPE_API_KEY"] = "your_key"  # Alibaba
os.environ["ZHIPUAI_API_KEY"] = "your_key"    # Zhipu
```

## Code Style

- Use Black formatter (line-length=100)
- Add type hints to all functions
- Prefer async functions
- Write detailed docstrings
- Use English for code, Chinese for comments (optional)

## Import Checklist

```python
from pyagent import (
    # Core
    Agent, BaseAgent, Memory, Runner,
    Chater, ChaterPool, Embedder, EmbedderPool,
    get_chater_cfg, get_embedder_cfg,
    
    # Tools
    ToolKit, MCPServerConfig,
    
    # Communication
    MsgHub, msghub,
    sequential_pipeline, parallel_pipeline,
    conditional_pipeline, loop_pipeline,
    
    # Trace
    trace, export_traces,
    
    # Advanced Agents
    ReActAgent, AgenticMemoryAgent,
    
    # Data Models
    ChatResponse, ToolCall, ToolResult,
    
    # Output
    Speaker, ConsoleSpeaker, SilentSpeaker,
    
    # Storage
    VectorStore, JsonVectorStore, ChromaVectorStore,
)
```

## Minimal Example

```python
import asyncio
from pyagent import Agent, Memory, Chater, get_chater_cfg, Runner

async def main():
    agent = Agent(
        name="Assistant",
        chater=Chater(get_chater_cfg("ali")),
        memory=Memory(),
        system_prompt="You are helpful"
    )
    
    result = await Runner.run(agent, "Hello")
    print(result.content)

asyncio.run(main())
```

---

## Remember These 5 Rules

1. ✅ **ALWAYS** wrap `get_chater_cfg()` with `Chater()`
2. ✅ **ALWAYS** use `await` for agent operations
3. ✅ **ALWAYS** make tool functions `async def`
4. ✅ **ALWAYS** add type hints and docstrings to tools
5. ✅ **NEVER** use `name` as tool parameter

## Development Workflow

1. Create Agent (remember Chater wrapping!)
2. Define tools (async + type hints)
3. Use Runner to execute
4. Handle results

## When You Encounter Issues

1. Check if Chater is properly wrapped
2. Check if you used await
3. Check if tool functions are async
4. Enable logging and tracing for debugging
5. Check the examples folder for reference

---

**Framework Philosophy**: Simple, Async, Composable

**Documentation**: See `examples/` for 15 complete examples
**Tests**: See `tests/` for comprehensive test suite
**GitHub**: https://github.com/aixiasang/pyagent
