# Cogency (Python)
> **Multi-step reasoning agents with clean architecture**
## Installation
```bash
pip install cogency
```
## Quick Start
### Standard Execution
```python
from cogency.agent import Agent
from cogency.llm import GeminiLLM
from cogency.tools import CalculatorTool, WebSearchTool, FileManagerTool
# Create agent with multiple tools
llm = GeminiLLM(api_keys="your-key") # New cleaner interface
agent = Agent(
name="MyAgent",
llm=llm,
tools=[
CalculatorTool(),
WebSearchTool(),
FileManagerTool()
]
)
# Execute with tracing
result = await agent.run("What is 15 * 23?", enable_trace=True, print_trace=True)
print(result["response"])
```
### Streaming Execution (NEW in 0.3.0)
```python
# Stream agent execution in real-time - see thinking process as it happens
async for chunk in agent.stream("What is 15 * 23?"):
if chunk["type"] == "thinking":
print(f"š {chunk['content']}")
elif chunk["type"] == "chunk":
print(f"š§ {chunk['content']}", end="")
elif chunk["type"] == "result":
print(f"\nā
{chunk['node']}: {chunk['data']}")
# Example output:
# š Analyzing user request and available tools...
# š Available tools: calculator (Basic arithmetic operations)
# š Generating plan decision...
# š§ {"action": "tool_needed", "reasoning": "Math calculation", "strategy": "calculator"}
# ā
plan: {"decision": "..."}
# š Analyzing task and selecting appropriate tool...
# š Available tools: ['calculator']
# š§ TOOL_CALL: calculator(operation='multiply', x1=15, x2=23)
# ā
reason: {"tool_call": "..."}
```
## Core Architecture
Cogency uses a clean 5-step reasoning loop:
1. **Plan** - Decide strategy and if tools are needed
2. **Reason** - Select tools and prepare inputs
3. **Act** - Execute tools with validation
4. **Reflect** - Evaluate results and decide next steps
5. **Respond** - Format clean answer for user
This separation enables emergent reasoning behavior - agents adapt their tool usage based on results without explicit programming.
### Stream-First Design (0.3.0)
Cogency implements a stream-first architecture where agents are defined by their execution streams rather than traditional callback patterns.
- Every node is an **async generator** that yields thinking steps in real-time
- Stream IS the execution, not a view of it
- Natural cancellation, unified interfaces, and transparent reasoning
- Configurable yield intervals for rate limiting
## Built-in Tools
- **CalculatorTool** - Basic arithmetic operations
- **WebSearchTool** - Web search using DuckDuckGo
- **FileManagerTool** - File system operations
## Adding Custom Tools
Create a new tool by extending the `BaseTool` class:
```python
from cogency.tools.base import BaseTool
class WeatherTool(BaseTool):
def __init__(self):
super().__init__(
name="weather",
description="Get current weather for a location"
)
def run(self, location: str) -> dict:
# Your implementation here
return {"temperature": 72, "condition": "sunny"}
```
Tools are automatically discovered and available to agents.
## LLM Support
Supports multiple LLM providers with automatic key rotation:
```python
# OpenAI
from cogency.llm import OpenAILLM
llm = OpenAILLM(api_keys="your-key", model="gpt-4")
# Anthropic Claude
from cogency.llm import AnthropicLLM
llm = AnthropicLLM(api_keys="your-key", model="claude-3-sonnet-20240229")
# Google Gemini
from cogency.llm import GeminiLLM
llm = GeminiLLM(api_keys="your-key", model="gemini-1.5-pro")
# Grok (X.AI)
from cogency.llm import GrokLLM
llm = GrokLLM(api_keys="your-key", model="grok-beta")
# Mistral
from cogency.llm import MistralLLM
llm = MistralLLM(api_keys="your-key", model="mistral-large-latest")
# Multiple keys with automatic rotation (all providers)
llm = OpenAILLM(api_keys=["key1", "key2", "key3"])
```
All LLM providers support:
- **Streaming execution** for real-time output
- **Key rotation** for high-volume usage
- **Rate limiting** via yield_interval parameter
- **Unified interface** - switch providers with one line
## Conversation History
Cogency supports conversation history with optional sliding window:
```python
from cogency.context import Context
# Create context with conversation history limit
context = Context("Hello", max_history=10) # Keep last 10 messages
# Run multiple interactions with shared context
result1 = await agent.run("What's 2+2?", context=context)
result2 = await agent.run("What about 3+3?", context=context) # Remembers previous exchange
# Access conversation state
print(f"Messages in context: {len(agent.context.messages)}")
print(f"Full conversation: {agent.context.get_clean_conversation()}")
# Continue conversation across multiple runs
result3 = await agent.run("Add those two results together", context=context)
```
**Key features:**
- **Sliding window**: Automatically trims old messages when `max_history` is reached
- **Context persistence**: Reuse contexts across multiple `agent.run()` calls
- **Context access**: Inspect conversation state via `agent.context` property
- **Clean output**: `get_clean_conversation()` filters internal messages
## Execution Tracing
Enable detailed tracing to see your agent's reasoning:
```python
# Simple trace viewing
result = await agent.run("Complex task", enable_trace=True, print_trace=True)
# Or capture trace data
result = await agent.run("Complex task", enable_trace=True)
trace_data = result["execution_trace"]
```
Example trace output:
```
--- Execution Trace (ID: abc123) ---
PLAN | Need to calculate and then search for information
REASON | TOOL_CALL: calculator(operation='multiply', num1=15, num2=23)
ACT | calculator -> {'result': 345}
REFLECT | Calculation completed, now need to search
REASON | TOOL_CALL: web_search(query='AI developments 2025')
ACT | web_search -> {'results': [...]}
REFLECT | Found relevant search results
RESPOND | 15 multiplied by 23 equals 345. Recent AI developments include...
--- End Trace ---
```
## Error Handling
All tools include built-in validation and graceful error handling:
```python
# Invalid operations are caught and handled
result = await agent.run("Calculate abc + def")
# Agent will respond with helpful error message instead of crashing
```
## CLI Usage
Run examples from the command line:
```bash
cd python
python examples/basic_usage.py
```
## Development
### Running Tests
```bash
pytest
```
### Project Structure
```
cogency/
āāā agent.py # Core agent implementation
āāā llm/ # LLM integrations
āāā tools/ # Built-in tools
āāā utils/ # Utilities and formatting
āāā tests/ # Test suite (115+ tests)
```
## Emergent Behavior
The key insight behind Cogency is that clean architectural separation enables emergent reasoning. When agents fail with one tool, they automatically reflect and try different approaches:
```python
# Agent fails with poor search query, reflects, and tries again
result = await agent.run("Tell me about recent AI developments")
# Trace shows:
# 1. Initial search with generic query
# 2. Poor results returned
# 3. Agent reflects on failure
# 4. Adapts query strategy
# 5. Succeeds with better results
```
This behavior emerges from the Plan ā Reason ā Act ā Reflect ā Respond loop, not from explicit programming.
## License
MIT License - see [LICENSE](../LICENSE) for details.
Raw data
{
"_id": null,
"home_page": null,
"name": "cogency",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.9",
"maintainer_email": null,
"keywords": "ai, agents, reasoning, tools, tracing, cognition",
"author": "Tyson Chan",
"author_email": "tyson.chan@proton.me",
"download_url": "https://files.pythonhosted.org/packages/b1/ff/ba282407e3851f4d7a631171e263e820eecf5c2aa10cb5a9d10c667bde23/cogency-0.3.0.tar.gz",
"platform": null,
"description": "# Cogency (Python)\n\n> **Multi-step reasoning agents with clean architecture**\n\n## Installation\n\n```bash\npip install cogency\n```\n\n## Quick Start\n\n### Standard Execution\n```python\nfrom cogency.agent import Agent\nfrom cogency.llm import GeminiLLM\nfrom cogency.tools import CalculatorTool, WebSearchTool, FileManagerTool\n\n# Create agent with multiple tools\nllm = GeminiLLM(api_keys=\"your-key\") # New cleaner interface\nagent = Agent(\n name=\"MyAgent\", \n llm=llm, \n tools=[\n CalculatorTool(), \n WebSearchTool(), \n FileManagerTool()\n ]\n)\n\n# Execute with tracing\nresult = await agent.run(\"What is 15 * 23?\", enable_trace=True, print_trace=True)\nprint(result[\"response\"])\n```\n\n### Streaming Execution (NEW in 0.3.0)\n```python\n# Stream agent execution in real-time - see thinking process as it happens\nasync for chunk in agent.stream(\"What is 15 * 23?\"):\n if chunk[\"type\"] == \"thinking\":\n print(f\"\ud83d\udcad {chunk['content']}\")\n elif chunk[\"type\"] == \"chunk\":\n print(f\"\ud83e\udde0 {chunk['content']}\", end=\"\")\n elif chunk[\"type\"] == \"result\":\n print(f\"\\n\u2705 {chunk['node']}: {chunk['data']}\")\n\n# Example output:\n# \ud83d\udcad Analyzing user request and available tools...\n# \ud83d\udcad Available tools: calculator (Basic arithmetic operations)\n# \ud83d\udcad Generating plan decision...\n# \ud83e\udde0 {\"action\": \"tool_needed\", \"reasoning\": \"Math calculation\", \"strategy\": \"calculator\"}\n# \u2705 plan: {\"decision\": \"...\"}\n# \ud83d\udcad Analyzing task and selecting appropriate tool...\n# \ud83d\udcad Available tools: ['calculator']\n# \ud83e\udde0 TOOL_CALL: calculator(operation='multiply', x1=15, x2=23)\n# \u2705 reason: {\"tool_call\": \"...\"}\n```\n\n## Core Architecture\n\nCogency uses a clean 5-step reasoning loop:\n\n1. **Plan** - Decide strategy and if tools are needed\n2. **Reason** - Select tools and prepare inputs\n3. **Act** - Execute tools with validation\n4. **Reflect** - Evaluate results and decide next steps\n5. **Respond** - Format clean answer for user\n\nThis separation enables emergent reasoning behavior - agents adapt their tool usage based on results without explicit programming.\n\n### Stream-First Design (0.3.0)\n\nCogency implements a stream-first architecture where agents are defined by their execution streams rather than traditional callback patterns.\n\n- Every node is an **async generator** that yields thinking steps in real-time\n- Stream IS the execution, not a view of it\n- Natural cancellation, unified interfaces, and transparent reasoning\n- Configurable yield intervals for rate limiting\n\n## Built-in Tools\n\n- **CalculatorTool** - Basic arithmetic operations\n- **WebSearchTool** - Web search using DuckDuckGo\n- **FileManagerTool** - File system operations\n\n## Adding Custom Tools\n\nCreate a new tool by extending the `BaseTool` class:\n\n```python\nfrom cogency.tools.base import BaseTool\n\nclass WeatherTool(BaseTool):\n def __init__(self):\n super().__init__(\n name=\"weather\",\n description=\"Get current weather for a location\"\n )\n \n def run(self, location: str) -> dict:\n # Your implementation here\n return {\"temperature\": 72, \"condition\": \"sunny\"}\n```\n\nTools are automatically discovered and available to agents.\n\n## LLM Support\n\nSupports multiple LLM providers with automatic key rotation:\n\n```python\n# OpenAI\nfrom cogency.llm import OpenAILLM\nllm = OpenAILLM(api_keys=\"your-key\", model=\"gpt-4\")\n\n# Anthropic Claude\nfrom cogency.llm import AnthropicLLM\nllm = AnthropicLLM(api_keys=\"your-key\", model=\"claude-3-sonnet-20240229\")\n\n# Google Gemini\nfrom cogency.llm import GeminiLLM\nllm = GeminiLLM(api_keys=\"your-key\", model=\"gemini-1.5-pro\")\n\n# Grok (X.AI)\nfrom cogency.llm import GrokLLM\nllm = GrokLLM(api_keys=\"your-key\", model=\"grok-beta\")\n\n# Mistral\nfrom cogency.llm import MistralLLM\nllm = MistralLLM(api_keys=\"your-key\", model=\"mistral-large-latest\")\n\n# Multiple keys with automatic rotation (all providers)\nllm = OpenAILLM(api_keys=[\"key1\", \"key2\", \"key3\"])\n```\n\nAll LLM providers support:\n- **Streaming execution** for real-time output\n- **Key rotation** for high-volume usage\n- **Rate limiting** via yield_interval parameter\n- **Unified interface** - switch providers with one line\n\n## Conversation History\n\nCogency supports conversation history with optional sliding window:\n\n```python\nfrom cogency.context import Context\n\n# Create context with conversation history limit\ncontext = Context(\"Hello\", max_history=10) # Keep last 10 messages\n\n# Run multiple interactions with shared context\nresult1 = await agent.run(\"What's 2+2?\", context=context)\nresult2 = await agent.run(\"What about 3+3?\", context=context) # Remembers previous exchange\n\n# Access conversation state\nprint(f\"Messages in context: {len(agent.context.messages)}\")\nprint(f\"Full conversation: {agent.context.get_clean_conversation()}\")\n\n# Continue conversation across multiple runs\nresult3 = await agent.run(\"Add those two results together\", context=context)\n```\n\n**Key features:**\n- **Sliding window**: Automatically trims old messages when `max_history` is reached\n- **Context persistence**: Reuse contexts across multiple `agent.run()` calls \n- **Context access**: Inspect conversation state via `agent.context` property\n- **Clean output**: `get_clean_conversation()` filters internal messages\n\n## Execution Tracing\n\nEnable detailed tracing to see your agent's reasoning:\n\n```python\n# Simple trace viewing\nresult = await agent.run(\"Complex task\", enable_trace=True, print_trace=True)\n\n# Or capture trace data\nresult = await agent.run(\"Complex task\", enable_trace=True)\ntrace_data = result[\"execution_trace\"]\n```\n\nExample trace output:\n```\n--- Execution Trace (ID: abc123) ---\nPLAN | Need to calculate and then search for information\nREASON | TOOL_CALL: calculator(operation='multiply', num1=15, num2=23)\nACT | calculator -> {'result': 345}\nREFLECT | Calculation completed, now need to search\nREASON | TOOL_CALL: web_search(query='AI developments 2025')\nACT | web_search -> {'results': [...]}\nREFLECT | Found relevant search results\nRESPOND | 15 multiplied by 23 equals 345. Recent AI developments include...\n--- End Trace ---\n```\n\n## Error Handling\n\nAll tools include built-in validation and graceful error handling:\n\n```python\n# Invalid operations are caught and handled\nresult = await agent.run(\"Calculate abc + def\")\n# Agent will respond with helpful error message instead of crashing\n```\n\n## CLI Usage\n\nRun examples from the command line:\n\n```bash\ncd python\npython examples/basic_usage.py\n```\n\n## Development\n\n### Running Tests\n```bash\npytest\n```\n\n### Project Structure\n```\ncogency/\n\u251c\u2500\u2500 agent.py # Core agent implementation\n\u251c\u2500\u2500 llm/ # LLM integrations\n\u251c\u2500\u2500 tools/ # Built-in tools\n\u251c\u2500\u2500 utils/ # Utilities and formatting\n\u2514\u2500\u2500 tests/ # Test suite (115+ tests)\n```\n\n## Emergent Behavior\n\nThe key insight behind Cogency is that clean architectural separation enables emergent reasoning. When agents fail with one tool, they automatically reflect and try different approaches:\n\n```python\n# Agent fails with poor search query, reflects, and tries again\nresult = await agent.run(\"Tell me about recent AI developments\")\n\n# Trace shows:\n# 1. Initial search with generic query\n# 2. Poor results returned\n# 3. Agent reflects on failure\n# 4. Adapts query strategy\n# 5. Succeeds with better results\n```\n\nThis behavior emerges from the Plan \u2192 Reason \u2192 Act \u2192 Reflect \u2192 Respond loop, not from explicit programming.\n\n## License\n\nMIT License - see [LICENSE](../LICENSE) for details.",
"bugtrack_url": null,
"license": "MIT",
"summary": "Agentic AI out of the box",
"version": "0.3.0",
"project_urls": null,
"split_keywords": [
"ai",
" agents",
" reasoning",
" tools",
" tracing",
" cognition"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "ac46a35beff6d86a6569443f45e6f31861fadf2a545b92d9d5b7b690cfdfac3d",
"md5": "6863c49838128f57a0e48c06f4016262",
"sha256": "a8322f085c31c98d769146712d39b0d7c12d83207e2fd0376043d9286dd7be4c"
},
"downloads": -1,
"filename": "cogency-0.3.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "6863c49838128f57a0e48c06f4016262",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.9",
"size": 42596,
"upload_time": "2025-07-13T01:31:36",
"upload_time_iso_8601": "2025-07-13T01:31:36.502155Z",
"url": "https://files.pythonhosted.org/packages/ac/46/a35beff6d86a6569443f45e6f31861fadf2a545b92d9d5b7b690cfdfac3d/cogency-0.3.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "b1ffba282407e3851f4d7a631171e263e820eecf5c2aa10cb5a9d10c667bde23",
"md5": "ceafcaaa05b781a93367a472f0e27611",
"sha256": "22a131a9bec69aebd4a0f6ab62b9eab881d8c74a8d4822fa3e6e2b53997dbf25"
},
"downloads": -1,
"filename": "cogency-0.3.0.tar.gz",
"has_sig": false,
"md5_digest": "ceafcaaa05b781a93367a472f0e27611",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.9",
"size": 29024,
"upload_time": "2025-07-13T01:31:37",
"upload_time_iso_8601": "2025-07-13T01:31:37.985423Z",
"url": "https://files.pythonhosted.org/packages/b1/ff/ba282407e3851f4d7a631171e263e820eecf5c2aa10cb5a9d10c667bde23/cogency-0.3.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-13 01:31:37",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "cogency"
}
JSON