# linedify
đŦ⥠linedify: Supercharging your LINE Bot with Dify power!
## ⨠Features
- 𧊠Seamless Dify-LINE Bot Integration
- Connect Dify with LINE Bot using minimal code
- Build powerful and efficient chatbots in no time
- đ¸ Rich Input Support
- Handle images, location data, and stickers out of the box
- Customize to work with LINE-specific UI like Flex Messages
- đĒ Developer-Friendly
- Built on FastAPI for high performance and easy scaling
- Asynchronous processing for smooth operations
## đĻ Install
```sh
pip install linedify
```
## đ Quick Start
Make the following script as `run.py` as the handler for WebHook from LINE API server.
By passing the HTTP request body and signature to `line_dify.process_request`, the entire process from receiving user messages to calling Dify and responding to the user is executed.
```python
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, BackgroundTasks
from linedify import LineDify
# LINE Bot - Dify Agent Integrator
line_dify = LineDify(
line_channel_access_token=YOUR_CHANNEL_ACCESS_TOKEN,
line_channel_secret=YOUR_CHANNEL_SECRET,
dify_api_key=DIFY_API_KEY,
dify_base_url=DIFY_BASE_URL, # e.g. http://localhost/v1
dify_user=DIFY_USER
)
# FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
yield
await line_dify.shutdown()
app = FastAPI(lifespan=lifespan)
@app.post("/linebot")
async def handle_request(request: Request, background_tasks: BackgroundTasks):
background_tasks.add_task(
line_dify.process_request,
request_body=(await request.body()).decode("utf-8"),
signature=request.headers.get("X-Line-Signature", "")
)
return "ok"
```
Start server.
```
uvicorn run:app
```
NOTE: You have to expose the host:port to where the LINE API server can access.
## đšī¸ Switching Types
linedify supports Agent and Chatbot for now. (You can add support for TextGenerator and Workflow on your own!)
You can switch the types by setting `dify_type` to the constructor of LineDify. Default is `DifyType.Agent`.
```python
line_dify = LineDify(
line_channel_access_token=YOUR_CHANNEL_ACCESS_TOKEN,
line_channel_secret=YOUR_CHANNEL_SECRET,
dify_api_key=DIFY_API_KEY,
dify_base_url=DIFY_BASE_URL,
dify_user=DIFY_USER,
dify_type=DifyType.Chatbot # <- DifyType.Agent or DifyType.Chatbot
)
```
## đ Use UI Components
Implement function to edit reply message below the decorator `@line_dify.to_reply_message`.
```python
from typing import List
from linebot.v3.messaging import Message, TextMessage, QuickReply, QuickReplyItem, MessageAction
from linedify.session import ConversationSession
@line_dify.to_reply_message
async def to_reply_message(text: str, data: dict, session: ConversationSession) -> List[Message]:
response_message = TextMessage(text=text)
# Show QuickReply buttons when tool "reservation" was executed on Dify
if tool := data.get("tool"):
if tool == "reservation":
response_message.quick_reply = QuickReply([
QuickReplyItem(action=MessageAction(label="Checkout", text="Checkout")),
QuickReplyItem(action=MessageAction(label="Cancel", text="Cancel"))
])
return [response_message]
```
## đ¨ Custom Logic
### Event Validation
Use `@line_dify.validate_event` to validate event before handling.
```python
banned_users = ["U123456", "U234567"]
@line_dify.validate_event
async def validate_event(event):
line_user_id = event.source.user_id
if line_user_id in banned_users:
# Return the list of TextMessage to reply immediately without processing the event
return [TextMessage("Forbidden")]
```
### Handle events
Use `@line_dify.event(event_type)` to customize event handlers.
```python
# Add handler for Postback event
@line_dify.event("postback")
async def handle_message_event(event: PostbackEvent):
# Do something here
# Return reply messages
return [TextMessage(f"Response for postback event: {event.postback.data}")]
# Add handler for unspecified event
@line_dify.event()
async def handle_event(event):
# Do something here
# Return reply messages
return [TextMessage(f"Response for event type: {event.type}")]
```
### Parse messages
Use `@line_dify.parse_message(message_type)` to customize message parsers.
```python
@line_dify.parse_message("location")
async def parse_location_message(message):
text, _ = await line_dify.parse_location_message(message)
map_image = get_map_image(message.address)
return (text, map_image)
```
### Inputs
Use `@line_dify.make_inputs` to customize `inputs` as arguments for Dify conversation threads.
```python
@line_dify.make_inputs
async def make_inputs(session: ConversationSession):
# You can use session to customize inputs dynamically here
inputs = {
"line_user_id": session.user_id,
"favorite_food": "apple"
}
return inputs
```
### Error Message
Use `@line_dify.to_error_message` to customize reply message when error occurs.
```python
@line_dify.to_error_message
async def to_error_message(event: Event, ex: Exception, session: ConversationSession = None):
# Custom logic here
text = random.choice(["Error đĨ˛", "đĩ Something wrong...", "đ"])
# Return reply messages
return [TextMessage(text=text)]
```
## đž Conversation Session
Conversation sessions are managed by a database. By default, SQLite is used, but you can specify the file path or database type using `session_db_url`. For the syntax, please refer to SQLAlchemy's documentation.
Additionally, you can specify the session validity period with `session_timeout`. The default is 3600 seconds. If this period elapses since the last conversation, a new conversation thread will be created on Dify when the next conversation starts.
```python
line_dify = LineDify(
line_channel_access_token=YOUR_CHANNEL_ACCESS_TOKEN,
line_channel_secret=YOUR_CHANNEL_SECRET,
dify_api_key=DIFY_API_KEY,
dify_base_url=DIFY_BASE_URL,
dify_user=DIFY_USER,
session_db_url="sqlite:///your_sessions.db", # SQLAlchemy database url
session_timeout=1800, # Timeout in seconds
)
```
## đ Debug
Set `verbose=True` to see the request and response, both from/to LINE and from/to Dify.
```python
line_dify = LineDify(
line_channel_access_token=YOUR_CHANNEL_ACCESS_TOKEN,
line_channel_secret=YOUR_CHANNEL_SECRET,
dify_api_key=DIFY_API_KEY,
dify_base_url=DIFY_BASE_URL,
dify_user=DIFY_USER,
verbose=True
)
```
## đ Long-Term Memory
Make function to store message histories. Here is the example for [Zep](https://www.getzep.com):
```sh
pip install zep-python
```
```python
import logging
from zep_python.client import AsyncZep
from zep_python.errors import NotFoundError
from zep_python.types import Message
logger = logging.getLogger(__name__)
class ZepIntegrator:
def __init__(self, *, api_key: str, base_url: str, cache_size: int = 1000, debug: bool = False):
self.zep_client = AsyncZep(
api_key=api_key,
base_url=base_url
)
self.cache_size = cache_size
self.user_ids = []
self.session_ids = []
self.debug = debug
async def add_user(self, user_id: str):
try:
user = await self.zep_client.user.get(
user_id=user_id
)
if self.debug:
logger.info(f"User found: {user}")
except NotFoundError:
user = await self.zep_client.user.add(
user_id=user_id
)
if self.debug:
logger.info(f"User created: {user}")
self.user_ids.append(user_id)
while len(self.user_ids) > self.cache_size:
self.user_ids.pop(0)
async def add_session(self, user_id: str, session_id: str):
try:
session = await self.zep_client.memory.get_session(
session_id=session_id
)
if self.debug:
logger.info(f"Session found: {session}")
except NotFoundError:
session = await self.zep_client.memory.add_session(
session_id=session_id,
user_id=user_id,
)
if self.debug:
logger.info(f"Session created: {session}")
self.session_ids.append(session_id)
while len(self.session_ids) > self.cache_size:
self.session_ids.pop(0)
async def add_messages(self, user_id: str, session_id: str, request_text: str, response_text: str):
if not user_id or not session_id or (not request_text and not response_text):
return
if user_id not in self.user_ids:
await self.add_user(user_id)
if session_id not in self.session_ids:
await self.add_session(user_id, session_id)
# Add messages
messages = []
if request_text:
messages.append(Message(role_type="user", content=request_text))
if response_text:
messages.append(Message(role_type="assistant", content=response_text))
if messages:
await self.zep_client.memory.add(session_id=session_id, messages=messages)
```
Call `add_messages` at the end of processing message event.
```python
zep = ZepIntegrator(
api_key="YOUR_ZEP_API_KEY",
base_url="ZEP_BASE_URL"
)
@line_dify.on_message_handling_end
async def on_message_handling_end(
conversation_session: ConversationSession,
request_text: str,
response_text: str,
response_data: any
):
await zep.add_messages(
conversation_session.user_id,
conversation_session.conversation_id,
request_text,
response_text
)
```
Then you can retrieve the facts about the user from wherever you like, including Dify.
## âī¸ License
linedify is distributed under the Apache v2 license.
(c)uezo, made with big â¤ī¸ in Tokyo.
Raw data
{
"_id": null,
"home_page": "https://github.com/uezo/linedify",
"name": "linedify",
"maintainer": "uezo",
"docs_url": null,
"requires_python": null,
"maintainer_email": "uezo@uezo.net",
"keywords": null,
"author": "uezo",
"author_email": "uezo@uezo.net",
"download_url": null,
"platform": null,
"description": "# linedify\n\n\ud83d\udcac\u26a1 linedify: Supercharging your LINE Bot with Dify power!\n\n\n## \u2728 Features\n\n- \ud83e\udde9 Seamless Dify-LINE Bot Integration\n\n - Connect Dify with LINE Bot using minimal code\n - Build powerful and efficient chatbots in no time\n\n- \ud83d\udcf8 Rich Input Support\n\n - Handle images, location data, and stickers out of the box\n - Customize to work with LINE-specific UI like Flex Messages\n\n- \ud83e\ude84 Developer-Friendly\n\n - Built on FastAPI for high performance and easy scaling\n - Asynchronous processing for smooth operations\n\n\n## \ud83d\udce6 Install\n\n```sh\npip install linedify\n```\n\n\n## \ud83d\ude80 Quick Start\n\nMake the following script as `run.py` as the handler for WebHook from LINE API server.\n\nBy passing the HTTP request body and signature to `line_dify.process_request`, the entire process from receiving user messages to calling Dify and responding to the user is executed.\n\n```python\nfrom contextlib import asynccontextmanager\nfrom fastapi import FastAPI, Request, BackgroundTasks\nfrom linedify import LineDify\n\n# LINE Bot - Dify Agent Integrator\nline_dify = LineDify(\n line_channel_access_token=YOUR_CHANNEL_ACCESS_TOKEN,\n line_channel_secret=YOUR_CHANNEL_SECRET,\n dify_api_key=DIFY_API_KEY,\n dify_base_url=DIFY_BASE_URL, # e.g. http://localhost/v1\n dify_user=DIFY_USER\n)\n\n# FastAPI\n@asynccontextmanager\nasync def lifespan(app: FastAPI):\n yield\n await line_dify.shutdown()\n\napp = FastAPI(lifespan=lifespan)\n\n@app.post(\"/linebot\")\nasync def handle_request(request: Request, background_tasks: BackgroundTasks):\n background_tasks.add_task(\n line_dify.process_request,\n request_body=(await request.body()).decode(\"utf-8\"),\n signature=request.headers.get(\"X-Line-Signature\", \"\")\n )\n return \"ok\"\n```\n\nStart server.\n\n```\nuvicorn run:app\n```\n\nNOTE: You have to expose the host:port to where the LINE API server can access.\n\n\n## \ud83d\udd79\ufe0f Switching Types\n\nlinedify supports Agent and Chatbot for now. (You can add support for TextGenerator and Workflow on your own!)\n\nYou can switch the types by setting `dify_type` to the constructor of LineDify. Default is `DifyType.Agent`.\n\n```python\nline_dify = LineDify(\n line_channel_access_token=YOUR_CHANNEL_ACCESS_TOKEN,\n line_channel_secret=YOUR_CHANNEL_SECRET,\n dify_api_key=DIFY_API_KEY,\n dify_base_url=DIFY_BASE_URL,\n dify_user=DIFY_USER,\n dify_type=DifyType.Chatbot # <- DifyType.Agent or DifyType.Chatbot\n)\n```\n\n\n## \ud83d\udc8e Use UI Components\n\nImplement function to edit reply message below the decorator `@line_dify.to_reply_message`.\n\n```python\nfrom typing import List\nfrom linebot.v3.messaging import Message, TextMessage, QuickReply, QuickReplyItem, MessageAction\nfrom linedify.session import ConversationSession\n\n@line_dify.to_reply_message\nasync def to_reply_message(text: str, data: dict, session: ConversationSession) -> List[Message]:\n response_message = TextMessage(text=text)\n\n # Show QuickReply buttons when tool \"reservation\" was executed on Dify\n if tool := data.get(\"tool\"):\n if tool == \"reservation\":\n response_message.quick_reply = QuickReply([\n QuickReplyItem(action=MessageAction(label=\"Checkout\", text=\"Checkout\")),\n QuickReplyItem(action=MessageAction(label=\"Cancel\", text=\"Cancel\"))\n ])\n\n return [response_message]\n```\n\n## \ud83c\udfa8 Custom Logic\n\n### Event Validation\n\nUse `@line_dify.validate_event` to validate event before handling.\n\n```python\nbanned_users = [\"U123456\", \"U234567\"]\n\n@line_dify.validate_event\nasync def validate_event(event):\n line_user_id = event.source.user_id\n if line_user_id in banned_users:\n # Return the list of TextMessage to reply immediately without processing the event\n return [TextMessage(\"Forbidden\")]\n```\n\n\n### Handle events\n\nUse `@line_dify.event(event_type)` to customize event handlers.\n\n```python\n# Add handler for Postback event\n@line_dify.event(\"postback\")\nasync def handle_message_event(event: PostbackEvent):\n # Do something here\n # Return reply messages\n return [TextMessage(f\"Response for postback event: {event.postback.data}\")]\n\n# Add handler for unspecified event\n@line_dify.event()\nasync def handle_event(event):\n # Do something here\n # Return reply messages\n return [TextMessage(f\"Response for event type: {event.type}\")]\n```\n\n\n### Parse messages\n\nUse `@line_dify.parse_message(message_type)` to customize message parsers.\n\n```python\n@line_dify.parse_message(\"location\")\nasync def parse_location_message(message):\n text, _ = await line_dify.parse_location_message(message)\n map_image = get_map_image(message.address)\n return (text, map_image)\n```\n\n\n### Inputs\n\nUse `@line_dify.make_inputs` to customize `inputs` as arguments for Dify conversation threads.\n\n```python\n@line_dify.make_inputs\nasync def make_inputs(session: ConversationSession):\n # You can use session to customize inputs dynamically here\n inputs = {\n \"line_user_id\": session.user_id,\n \"favorite_food\": \"apple\"\n }\n \n return inputs\n```\n\n\n### Error Message\n\nUse `@line_dify.to_error_message` to customize reply message when error occurs.\n\n```python\n@line_dify.to_error_message\nasync def to_error_message(event: Event, ex: Exception, session: ConversationSession = None):\n # Custom logic here\n text = random.choice([\"Error \ud83e\udd72\", \"\ud83d\ude35 Something wrong...\", \"\ud83d\ude43\"])\n # Return reply messages\n return [TextMessage(text=text)]\n```\n\n\n## \ud83d\udcbe Conversation Session\n\nConversation sessions are managed by a database. By default, SQLite is used, but you can specify the file path or database type using `session_db_url`. For the syntax, please refer to SQLAlchemy's documentation.\n\nAdditionally, you can specify the session validity period with `session_timeout`. The default is 3600 seconds. If this period elapses since the last conversation, a new conversation thread will be created on Dify when the next conversation starts.\n\n```python\nline_dify = LineDify(\n line_channel_access_token=YOUR_CHANNEL_ACCESS_TOKEN,\n line_channel_secret=YOUR_CHANNEL_SECRET,\n dify_api_key=DIFY_API_KEY,\n dify_base_url=DIFY_BASE_URL,\n dify_user=DIFY_USER,\n session_db_url=\"sqlite:///your_sessions.db\", # SQLAlchemy database url\n session_timeout=1800, # Timeout in seconds\n)\n```\n\n\n## \ud83d\udc1d Debug\n\nSet `verbose=True` to see the request and response, both from/to LINE and from/to Dify.\n\n```python\nline_dify = LineDify(\n line_channel_access_token=YOUR_CHANNEL_ACCESS_TOKEN,\n line_channel_secret=YOUR_CHANNEL_SECRET,\n dify_api_key=DIFY_API_KEY,\n dify_base_url=DIFY_BASE_URL,\n dify_user=DIFY_USER,\n verbose=True\n)\n```\n\n\n## \ud83d\udcdc Long-Term Memory\n\nMake function to store message histories. Here is the example for [Zep](https://www.getzep.com):\n\n```sh\npip install zep-python\n```\n\n```python\nimport logging\nfrom zep_python.client import AsyncZep\nfrom zep_python.errors import NotFoundError\nfrom zep_python.types import Message\n\nlogger = logging.getLogger(__name__)\n\nclass ZepIntegrator:\n def __init__(self, *, api_key: str, base_url: str, cache_size: int = 1000, debug: bool = False):\n self.zep_client = AsyncZep(\n api_key=api_key,\n base_url=base_url\n )\n self.cache_size = cache_size\n self.user_ids = []\n self.session_ids = []\n self.debug = debug\n\n async def add_user(self, user_id: str):\n try:\n user = await self.zep_client.user.get(\n user_id=user_id\n )\n if self.debug:\n logger.info(f\"User found: {user}\")\n except NotFoundError:\n user = await self.zep_client.user.add(\n user_id=user_id\n )\n if self.debug:\n logger.info(f\"User created: {user}\")\n\n self.user_ids.append(user_id)\n while len(self.user_ids) > self.cache_size:\n self.user_ids.pop(0)\n\n async def add_session(self, user_id: str, session_id: str):\n try:\n session = await self.zep_client.memory.get_session(\n session_id=session_id\n )\n if self.debug:\n logger.info(f\"Session found: {session}\")\n except NotFoundError:\n session = await self.zep_client.memory.add_session(\n session_id=session_id,\n user_id=user_id,\n )\n if self.debug:\n logger.info(f\"Session created: {session}\")\n\n self.session_ids.append(session_id)\n while len(self.session_ids) > self.cache_size:\n self.session_ids.pop(0)\n\n async def add_messages(self, user_id: str, session_id: str, request_text: str, response_text: str):\n if not user_id or not session_id or (not request_text and not response_text):\n return\n\n if user_id not in self.user_ids:\n await self.add_user(user_id)\n\n if session_id not in self.session_ids:\n await self.add_session(user_id, session_id)\n\n # Add messages\n messages = []\n if request_text:\n messages.append(Message(role_type=\"user\", content=request_text))\n if response_text:\n messages.append(Message(role_type=\"assistant\", content=response_text))\n\n if messages:\n await self.zep_client.memory.add(session_id=session_id, messages=messages)\n```\n\n\nCall `add_messages` at the end of processing message event.\n\n```python\nzep = ZepIntegrator(\n api_key=\"YOUR_ZEP_API_KEY\",\n base_url=\"ZEP_BASE_URL\"\n)\n\n@line_dify.on_message_handling_end\nasync def on_message_handling_end(\n conversation_session: ConversationSession,\n request_text: str,\n response_text: str,\n response_data: any\n):\n await zep.add_messages(\n conversation_session.user_id,\n conversation_session.conversation_id,\n request_text,\n response_text\n )\n```\n\nThen you can retrieve the facts about the user from wherever you like, including Dify.\n\n\n## \u2696\ufe0f License\n\nlinedify is distributed under the Apache v2 license.\n\n(c)uezo, made with big \u2764\ufe0f in Tokyo.\n",
"bugtrack_url": null,
"license": "Apache v2",
"summary": "\ud83d\udcac\u26a1 linedify: Supercharging your LINE Bot with Dify power!",
"version": "0.3.2",
"project_urls": {
"Homepage": "https://github.com/uezo/linedify"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "b98b7c97d446953ce959a8a036ec7e6ae4d5afc4ed8e78bc964cef4de8475a73",
"md5": "336749e345d9572851cf47974d9a10a0",
"sha256": "68abc03eb70d921b5886d719c5855b2bf4c70328196a723288f6f292f85a20c6"
},
"downloads": -1,
"filename": "linedify-0.3.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "336749e345d9572851cf47974d9a10a0",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": null,
"size": 14307,
"upload_time": "2025-01-13T15:39:47",
"upload_time_iso_8601": "2025-01-13T15:39:47.414042Z",
"url": "https://files.pythonhosted.org/packages/b9/8b/7c97d446953ce959a8a036ec7e6ae4d5afc4ed8e78bc964cef4de8475a73/linedify-0.3.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-01-13 15:39:47",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "uezo",
"github_project": "linedify",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"requirements": [
{
"name": "aiohttp",
"specs": [
[
"==",
"3.9.5"
]
]
},
{
"name": "line-bot-sdk",
"specs": [
[
"==",
"3.11.0"
]
]
},
{
"name": "fastapi",
"specs": [
[
"==",
"0.111.0"
]
]
},
{
"name": "uvicorn",
"specs": [
[
"==",
"0.30.1"
]
]
},
{
"name": "SQLAlchemy",
"specs": [
[
"==",
"2.0.31"
]
]
}
],
"lcname": "linedify"
}
JSON
