# AgentDiff Coordination
AgentDiff is a lightweight coordination library that prevents common concurrency issues in multi-agent systems - such as agents starting before dependencies complete, multiple agents writing to shared resources, or concurrent API calls exceeding rate limits.
Simply add `@coordinate` decorators for resource locks and `@when` for event-driven chaining. AgentDiff integrates with existing agent frameworks like LangChain, CrewAI, or pure Python implementations.
Provides coordination primitives without requiring framework migration or architectural changes.
[](https://badge.fury.io/py/agentdiff-coordination)
[](https://pypi.org/project/agentdiff-coordination/)
[](https://opensource.org/licenses/MIT)
## What AgentDiff Coordination Solves
- **Race conditions** between agents accessing shared resources.
- **Corrupted state** when multiple agents write to the same keys.
- **API rate limit chaos** from concurrent LLM calls.
- **"Two nodes writing to same key"** bugs that require extensive debugging.
- **Framework complexity** that gets in the way of actually building agents.
## Core Features
- **`@coordinate`** - Resource locks + automatic lifecycle events.
- **`@when`** - Event-driven agent chaining (no manual orchestration).
- **`emit()`** - Custom events for complex workflows.
- **Zero Configuration** - Works immediately, configure only what you need.
- **Framework Agnostic** - Works with any agent framework or pure Python.
## Use Cases: Concurrency Issues and Race Conditions
### **Concurrent State Updates**
```python
# Before: Race conditions in shared state
def process_customer_data():
customer_state["status"] = "processing" # Race condition
result = process_data()
customer_state["result"] = result # Overwrites other agent
def update_customer_profile():
customer_state["status"] = "updating" # Conflicts with processor
customer_state["profile"] = new_profile # State corruption
# After: Resource locks prevent conflicts
@coordinate("data_processor", lock_name="customer_123")
def process_customer_data():
customer_state["status"] = "processing" # Exclusive access
result = process_data()
customer_state["result"] = result # Safe update
@coordinate("profile_updater", lock_name="customer_123")
def update_customer_profile():
customer_state["status"] = "updating" # Waits for processor
customer_state["profile"] = new_profile # No conflicts
```
### **API Rate Limit Management**
```python
# Before: Multiple agents hitting APIs simultaneously
def research_agent():
response = openai.chat.completions.create(...) # Rate limited
def analysis_agent():
response = openai.chat.completions.create(...) # Rate limited
def summary_agent():
response = openai.chat.completions.create(...) # Rate limited
# Running in parallel creates debugging challenges
threading.Thread(target=research_agent).start()
threading.Thread(target=analysis_agent).start()
threading.Thread(target=summary_agent).start()
# After: Resource locks queue API calls safely
@coordinate("researcher", lock_name="openai_api")
def research_agent():
response = openai.chat.completions.create(...) # Queued safely
@coordinate("analyzer", lock_name="openai_api")
def analysis_agent():
response = openai.chat.completions.create(...) # Waits for researcher
@coordinate("summarizer", lock_name="openai_api")
def summary_agent():
response = openai.chat.completions.create(...) # Waits for analyzer
```
### **Manual Orchestration Complexity**
```python
# Before: Complex manual coordination
def run_workflow():
research_result = research_agent()
if research_result:
analysis_result = analysis_agent(research_result)
if analysis_result:
summary_result = summary_agent(analysis_result)
if summary_result:
final_report = editor_agent(summary_result)
# Error handling, retries, parallel flows increase complexity
# After: Event-driven coordination
@coordinate("researcher")
def research_agent():
return research_data
@when("researcher_complete")
def start_analysis(event_data):
analysis_agent(event_data['result']) # Auto-triggered
@when("analyzer_complete")
def start_summary(event_data):
summary_agent(event_data['result']) # Auto-chained
# Just start the workflow - coordination happens automatically
research_agent() # Everything else flows automatically
```
## Installation & Requirements
**Python Support**: 3.9+ (tested on 3.9, 3.10, 3.11, 3.12)
```bash
pip install agentdiff-coordination
```
## Documentation
- **[Quick Start Guide](docs/quickstart.md)** - Get up and running in 5 minutes.
- **[API Reference](docs/api-reference.md)** - Complete function documentation.
- **[Use cases](docs/use-cases.md)** - Example use cases.
- **[Configuration Guide](docs/configuration.md)** - Environment variables and settings.
- **[Examples](examples/)** - Example agent coordination patterns.
## Contributing
We welcome contributions! See [CONTRIBUTING.md](CONTRIBUTING.md) for guidelines.
## License
MIT License - see [LICENSE](LICENSE) file for details.
## About AgentDiff
AgentDiff provides practical tools for AI developers working with multi-agent systems. This coordination library addresses common concurrency challenges encountered in production agent workflows.
- **GitHub**: [https://github.com/AgentDiff](https://github.com/AgentDiff)
- **Issues**: [Report bugs and request features](https://github.com/AgentDiff/agentdiff-coordination/issues)
- **Community**: Share coordination patterns and production experiences
Raw data
{
"_id": null,
"home_page": null,
"name": "agentdiff-coordination",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": null,
"keywords": "agentdiff, agents, ai, concurrency, coordination, python, race-conditions",
"author": null,
"author_email": "AgentDiff Team <hello@agentdiff.com>",
"download_url": "https://files.pythonhosted.org/packages/3f/39/57833a0177b3b46c1a412017c820fa61ff6bf3fdd97341fbddbc2f2f4b4a/agentdiff_coordination-0.1.1.tar.gz",
"platform": null,
"description": "# AgentDiff Coordination\n\nAgentDiff is a lightweight coordination library that prevents common concurrency issues in multi-agent systems - such as agents starting before dependencies complete, multiple agents writing to shared resources, or concurrent API calls exceeding rate limits.\n\nSimply add `@coordinate` decorators for resource locks and `@when` for event-driven chaining. AgentDiff integrates with existing agent frameworks like LangChain, CrewAI, or pure Python implementations.\n\nProvides coordination primitives without requiring framework migration or architectural changes.\n\n[](https://badge.fury.io/py/agentdiff-coordination)\n[](https://pypi.org/project/agentdiff-coordination/)\n[](https://opensource.org/licenses/MIT)\n\n## What AgentDiff Coordination Solves\n\n- **Race conditions** between agents accessing shared resources.\n- **Corrupted state** when multiple agents write to the same keys.\n- **API rate limit chaos** from concurrent LLM calls.\n- **\"Two nodes writing to same key\"** bugs that require extensive debugging.\n- **Framework complexity** that gets in the way of actually building agents.\n\n## Core Features\n\n- **`@coordinate`** - Resource locks + automatic lifecycle events.\n- **`@when`** - Event-driven agent chaining (no manual orchestration).\n- **`emit()`** - Custom events for complex workflows.\n- **Zero Configuration** - Works immediately, configure only what you need.\n- **Framework Agnostic** - Works with any agent framework or pure Python.\n\n## Use Cases: Concurrency Issues and Race Conditions\n\n### **Concurrent State Updates**\n\n```python\n# Before: Race conditions in shared state\ndef process_customer_data():\n customer_state[\"status\"] = \"processing\" # Race condition\n result = process_data()\n customer_state[\"result\"] = result # Overwrites other agent\n\ndef update_customer_profile():\n customer_state[\"status\"] = \"updating\" # Conflicts with processor\n customer_state[\"profile\"] = new_profile # State corruption\n\n# After: Resource locks prevent conflicts\n@coordinate(\"data_processor\", lock_name=\"customer_123\")\ndef process_customer_data():\n customer_state[\"status\"] = \"processing\" # Exclusive access\n result = process_data()\n customer_state[\"result\"] = result # Safe update\n\n@coordinate(\"profile_updater\", lock_name=\"customer_123\")\ndef update_customer_profile():\n customer_state[\"status\"] = \"updating\" # Waits for processor\n customer_state[\"profile\"] = new_profile # No conflicts\n```\n\n### **API Rate Limit Management**\n\n```python\n# Before: Multiple agents hitting APIs simultaneously\ndef research_agent():\n response = openai.chat.completions.create(...) # Rate limited\n\ndef analysis_agent():\n response = openai.chat.completions.create(...) # Rate limited\n\ndef summary_agent():\n response = openai.chat.completions.create(...) # Rate limited\n\n# Running in parallel creates debugging challenges\nthreading.Thread(target=research_agent).start()\nthreading.Thread(target=analysis_agent).start()\nthreading.Thread(target=summary_agent).start()\n\n# After: Resource locks queue API calls safely\n@coordinate(\"researcher\", lock_name=\"openai_api\")\ndef research_agent():\n response = openai.chat.completions.create(...) # Queued safely\n\n@coordinate(\"analyzer\", lock_name=\"openai_api\")\ndef analysis_agent():\n response = openai.chat.completions.create(...) # Waits for researcher\n\n@coordinate(\"summarizer\", lock_name=\"openai_api\")\ndef summary_agent():\n response = openai.chat.completions.create(...) # Waits for analyzer\n```\n\n### **Manual Orchestration Complexity**\n\n```python\n# Before: Complex manual coordination\ndef run_workflow():\n research_result = research_agent()\n if research_result:\n analysis_result = analysis_agent(research_result)\n if analysis_result:\n summary_result = summary_agent(analysis_result)\n if summary_result:\n final_report = editor_agent(summary_result)\n # Error handling, retries, parallel flows increase complexity\n\n# After: Event-driven coordination\n@coordinate(\"researcher\")\ndef research_agent():\n return research_data\n\n@when(\"researcher_complete\")\ndef start_analysis(event_data):\n analysis_agent(event_data['result']) # Auto-triggered\n\n@when(\"analyzer_complete\")\ndef start_summary(event_data):\n summary_agent(event_data['result']) # Auto-chained\n\n# Just start the workflow - coordination happens automatically\nresearch_agent() # Everything else flows automatically\n```\n\n## Installation & Requirements\n\n**Python Support**: 3.9+ (tested on 3.9, 3.10, 3.11, 3.12)\n\n```bash\npip install agentdiff-coordination\n```\n\n## Documentation\n\n- **[Quick Start Guide](docs/quickstart.md)** - Get up and running in 5 minutes.\n- **[API Reference](docs/api-reference.md)** - Complete function documentation.\n- **[Use cases](docs/use-cases.md)** - Example use cases.\n- **[Configuration Guide](docs/configuration.md)** - Environment variables and settings.\n- **[Examples](examples/)** - Example agent coordination patterns.\n\n## Contributing\n\nWe welcome contributions! See [CONTRIBUTING.md](CONTRIBUTING.md) for guidelines.\n\n## License\n\nMIT License - see [LICENSE](LICENSE) file for details.\n\n## About AgentDiff\n\nAgentDiff provides practical tools for AI developers working with multi-agent systems. This coordination library addresses common concurrency challenges encountered in production agent workflows.\n\n- **GitHub**: [https://github.com/AgentDiff](https://github.com/AgentDiff)\n- **Issues**: [Report bugs and request features](https://github.com/AgentDiff/agentdiff-coordination/issues)\n- **Community**: Share coordination patterns and production experiences\n",
"bugtrack_url": null,
"license": null,
"summary": "Lightweight coordination library for AI agents to prevent concurrency issues",
"version": "0.1.1",
"project_urls": {
"Documentation": "https://github.com/AgentDiff/agentdiff-coordination#readme",
"Homepage": "https://github.com/AgentDiff/agentdiff-coordination",
"Issues": "https://github.com/AgentDiff/agentdiff-coordination/issues",
"Repository": "https://github.com/AgentDiff/agentdiff-coordination"
},
"split_keywords": [
"agentdiff",
" agents",
" ai",
" concurrency",
" coordination",
" python",
" race-conditions"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "fbc8d9d38bf5355ddf91c768cf7565c1500e065b80bf00b85da11514559f8efb",
"md5": "c149308fac1dbb25552c9a7f36d3a570",
"sha256": "26023af0a5dc3f2da9454f77478efae32ba87cc2d97c24cfb0e8d313be5c5a39"
},
"downloads": -1,
"filename": "agentdiff_coordination-0.1.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "c149308fac1dbb25552c9a7f36d3a570",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 30116,
"upload_time": "2025-08-14T07:16:20",
"upload_time_iso_8601": "2025-08-14T07:16:20.408790Z",
"url": "https://files.pythonhosted.org/packages/fb/c8/d9d38bf5355ddf91c768cf7565c1500e065b80bf00b85da11514559f8efb/agentdiff_coordination-0.1.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "3f3957833a0177b3b46c1a412017c820fa61ff6bf3fdd97341fbddbc2f2f4b4a",
"md5": "f425b95cc2feb6213ac3c45358a8b9e4",
"sha256": "bd2de39482197865cd3563b8f68632c19ceb467e922fc36b456e91e8c54eb8cf"
},
"downloads": -1,
"filename": "agentdiff_coordination-0.1.1.tar.gz",
"has_sig": false,
"md5_digest": "f425b95cc2feb6213ac3c45358a8b9e4",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 330223,
"upload_time": "2025-08-14T07:16:22",
"upload_time_iso_8601": "2025-08-14T07:16:22.806592Z",
"url": "https://files.pythonhosted.org/packages/3f/39/57833a0177b3b46c1a412017c820fa61ff6bf3fdd97341fbddbc2f2f4b4a/agentdiff_coordination-0.1.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-14 07:16:22",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "AgentDiff",
"github_project": "agentdiff-coordination#readme",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "agentdiff-coordination"
}
JSON
